From: Kai Moritz <kai@juplo.de>
Date: Sat, 2 Nov 2024 21:23:16 +0000 (+0100)
Subject: Fix: `close()` muss noch vom `ExampleConsumer` aufgerufen werden
X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=4dccfa10d82ecffa33a544191cb1577b74acd267;p=demos%2Fkafka%2Ftraining

Fix: `close()` muss noch vom `ExampleConsumer` aufgerufen werden

* Der Aufruf von `close()` löst die Abmeldung der Instanz bei dem
  GroupCoordinator aus.
* Dieser Vorgang sollte noch unter der Kontrolle des Anwendungscodes
  erfolgen!
* Wenn die Methode erst von Spring aufgerufen wird, werden dann ggf. noch
  Seiteneffekte ausgelöst, die dann noch im Kontext der Instanz laufen,
  obwohl diese eigentlich schon beendet wurde!
---

diff --git a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
index 116d63d..0046f8f 100644
--- a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
+++ b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
@@ -27,7 +27,7 @@ public class ApplicationConfiguration
             kafkaConsumer);
   }
 
-  @Bean
+  @Bean(destroyMethod = "")
   public KafkaConsumer<String, String> kafkaConsumer(ApplicationProperties properties)
   {
     Properties props = new Properties();
diff --git a/src/main/java/de/juplo/kafka/ExampleConsumer.java b/src/main/java/de/juplo/kafka/ExampleConsumer.java
index 13431d3..3206067 100644
--- a/src/main/java/de/juplo/kafka/ExampleConsumer.java
+++ b/src/main/java/de/juplo/kafka/ExampleConsumer.java
@@ -73,6 +73,8 @@ public class ExampleConsumer implements Runnable
     }
     finally
     {
+      log.info("{} - Closing the KafkaConsumer", id);
+      consumer.close();
       log.info("{}: Consumed {} messages in total, exiting!", id, consumed);
     }
   }