From abc2ffdb4d2829cd2c4263df9f6cec10b2a60c03 Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Sat, 9 Apr 2022 18:50:43 +0200 Subject: [PATCH] Fehlerbehandlung in EndlessConsumer.destroy() korrigiert --- src/main/java/de/juplo/kafka/EndlessConsumer.java | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/main/java/de/juplo/kafka/EndlessConsumer.java b/src/main/java/de/juplo/kafka/EndlessConsumer.java index b3dd446..f25e93c 100644 --- a/src/main/java/de/juplo/kafka/EndlessConsumer.java +++ b/src/main/java/de/juplo/kafka/EndlessConsumer.java @@ -110,7 +110,7 @@ public class EndlessConsumer implements Runnable { boolean stateChanged = running.compareAndSet(false, true); if (!stateChanged) - throw new RuntimeException("Consumer instance " + id + " is already running!"); + throw new IllegalStateException("Consumer instance " + id + " is already running!"); log.info("{} - Starting - consumed {} messages before", id, consumed); future = executor.submit(this); @@ -120,7 +120,7 @@ public class EndlessConsumer implements Runnable { boolean stateChanged = running.compareAndSet(true, false); if (!stateChanged) - throw new RuntimeException("Consumer instance " + id + " is not running!"); + throw new IllegalStateException("Consumer instance " + id + " is not running!"); log.info("{} - Stopping", id); consumer.wakeup(); @@ -140,6 +140,10 @@ public class EndlessConsumer implements Runnable { log.info("{} - Was already stopped", id); } + catch (Exception e) + { + log.error("{} - Unexpected exception while trying to stop the consumer", id, e); + } finally { log.info("{}: Consumed {} messages in total, exiting!", id, consumed); -- 2.20.1