add a thread join to wait for consumer thread termination

kafka-ops · Oct 20, 2021 · 641f3b8 · 641f3b8
1 parent 0f304c9
commit 641f3b8
Showing 1 changed file with 6 additions and 1 deletion.
diff --git a/src/main/java/com/purbon/kafka/topology/backend/KafkaBackend.java b/src/main/java/com/purbon/kafka/topology/backend/KafkaBackend.java
@@ -43,7 +43,7 @@ public void run() {
       try {
         consumer.retrieve(callback);
       } catch (WakeupException ex) {
-        LOGGER.error(ex);
+        LOGGER.trace(ex);
       }
     }
   }
@@ -97,6 +97,11 @@ public BackendState load() throws IOException {
   public void close() {
     consumer.stop();
     producer.stop();
+    try {
+      thread.join();
+    } catch (InterruptedException e) {
+      LOGGER.error(e);
+    }
     latest = null;
     thread = null;
   }