close consumer once completed

Signed-off-by: Jorge Aguilera <[email protected]>
panaxging · Oct 27, 2022 · 9fd1844 · 9fd1844
1 parent 5b31131
commit 9fd1844
Showing 1 changed file with 12 additions and 1 deletion.
diff --git a/plugins/nf-kafka/src/main/nextflow/events/kafa/TopicHandler.groovy b/plugins/nf-kafka/src/main/nextflow/events/kafa/TopicHandler.groovy
@@ -72,7 +72,12 @@ class TopicHandler {
 
     TopicHandler perform() {
         createConsumer()
-        listening ? runAsync() : consume()
+        if( listening ) {
+            runAsync()
+        }else{
+            consume()
+            closeConsumer()
+        }
         return this
     }
 
@@ -92,6 +97,10 @@ class TopicHandler {
         consumer
     }
 
+    void closeConsumer(){
+        consumer.close()
+    }
+
     void consume(){
         try {
             final records = consumer.poll(duration)
@@ -114,6 +123,8 @@ class TopicHandler {
                 log.trace "Closing $topic kafka thread"
             }catch(Exception e){
                 log.error "Exception reading kafka topic $topic",e
+            }finally{
+                closeConsumer()
             }
         })
     }