Some failure test

it-test/src/test/java/io/datanerds/verteiler/it_test/BlockingQueueConsumerTest.java

@@ -129,4 +129,40 @@ public void testSendOneMessageRestartConsumerEnsureOneMessageOnly() throws Excep
         await().atMost(2, SECONDS).until(() -> messageCounter.get() == 1);
         consumer1.stop();
     }
+
+    @Test
+    public void handleMessageWithExceptionTest() throws Exception {
+        final String topic = "handle_exception_topic";
+        createTopic(topic);
+
+        AtomicInteger messageCounter = new AtomicInteger();
+
+        Consumer<String> action1 = (message) -> {
+            throw new RuntimeException("unexpected");
+        };
+
+        Consumer<String> action2 = (message) -> {
+            messageCounter.incrementAndGet();
+        };
+
+        BlockingQueueConsumer<String, String> consumer1 = new BlockingQueueConsumer<>(topic, props,
+                42, action1);
+        consumer1.start();
+
+        BlockingQueueConsumer<String, String> consumer2 = new BlockingQueueConsumer<>(topic, props,
+                42, action2);
+        consumer2.start();
+
+        SimpleTestProducer testProducer = new SimpleTestProducer("Lorem-Radio", topic, kafkaConnect);
+        logger.info("Sending {} messages", NUMBER_OF_MESSAGES);
+
+        for (int i = 0; i < NUMBER_OF_MESSAGES; i++) {
+            testProducer.send(LOREM_IPSUM.paragraph());
+        }
+
+        await().atMost(60, SECONDS).until(() -> messageCounter.get() >= NUMBER_OF_MESSAGES);
+
+        testProducer.close();
+        consumer2.stop();
+    }
 }

verteiler/src/main/java/io/datanerds/verteiler/BlockingQueueConsumer.java

@@ -109,7 +109,13 @@ void relay(ConsumerRecord<K, V> message) throws InterruptedException {
         if (!topic.equals(message.topic())) {
             throw new ConsumerException(String.format("Message from unexpected topic: '%s'", message.topic()));
         }
+
         Processor<K, V> processor = processors.get(message.partition());
+        if (processor.isStopped()) {
+            throw new ConsumerException(
+                    String.format("Processor '%s-%s' has stopped", message.topic(), message.partition())
+            );
+        }
         processor.queue(message);
     }
 

verteiler/src/main/java/io/datanerds/verteiler/ConsumerRecordRelay.java

@@ -41,6 +41,7 @@ public void run() {
                 } catch (Exception ex) {
                     logger.error("Error while relaying messages from kafka to queue: {}", ex.getMessage(), ex);
                     blockingQueueConsumer.stop();
+                    break;
                 }
             }
             commitOffsets();