From 0224b257a2b0ef295765d1b3bfed7c4479c280e7 Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Fri, 15 Apr 2022 11:29:48 +0200 Subject: [PATCH] =?utf8?q?Springify:=20Start/Stop=20pr=C3=BCft,=20ob=20der?= =?utf8?q?=20Container=20schon/noch=20l=C3=A4uft?= MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit --- .../java/de/juplo/kafka/EndlessConsumer.java | 23 ++++++++++++++++++- 1 file changed, 22 insertions(+), 1 deletion(-) diff --git a/src/main/java/de/juplo/kafka/EndlessConsumer.java b/src/main/java/de/juplo/kafka/EndlessConsumer.java index ea899cc..a5a5ce6 100644 --- a/src/main/java/de/juplo/kafka/EndlessConsumer.java +++ b/src/main/java/de/juplo/kafka/EndlessConsumer.java @@ -59,12 +59,18 @@ public class EndlessConsumer public synchronized void start() { + if (registry.getListenerContainer(id).isChildRunning()) + throw new IllegalStateException("Consumer instance " + id + " is already running!"); + log.info("{} - Starting - consumed {} messages before", id, consumed); registry.getListenerContainer(id).start(); } public synchronized void stop() { + if (!registry.getListenerContainer(id).isChildRunning()) + throw new IllegalStateException("Consumer instance " + id + " is not running!"); + log.info("{} - Stopping", id); registry.getListenerContainer(id).stop(); log.info("{} - Stopped - consumed {} messages so far", id, consumed); @@ -74,6 +80,21 @@ public class EndlessConsumer public void destroy() { log.info("{} - Destroy!", id); - stop(); + try + { + stop(); + } + catch (IllegalStateException e) + { + 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