Ordentlich mit `instance.consumer.wakeup()`

diff --git a/src/main/java/de/juplo/kafka/ExampleConsumer.java b/src/main/java/de/juplo/kafka/ExampleConsumer.java
index 867215729072858783739fabfe759ec14261131d..223f0221de2547ca13ae006ae34a60be13554193 100644 (file)
--- a/src/main/java/de/juplo/kafka/ExampleConsumer.java
+++ b/src/main/java/de/juplo/kafka/ExampleConsumer.java
@@ -5,6 +5,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.errors.WakeupException;
 import org.apache.kafka.common.serialization.StringDeserializer;
 
 import java.time.Duration;
@@ -67,6 +68,11 @@ public class ExampleConsumer
         }
       }
     }
+    catch(WakeupException e)
+    {
+      log.info("{}: Wakeup!", id);
+      consumer.close();
+    }
     finally
     {
       log.info("{}: Consumed {} messages in total, exiting!", id, consumed);
@@ -87,7 +93,7 @@ public class ExampleConsumer
 
     Runtime.getRuntime().addShutdownHook(new Thread(() ->
     {
-      instance.consumer.close();
+      instance.consumer.wakeup();
       while (instance.running)
       {
         log.info("Waiting...");