Subscribe to topics on startConsuming

juriwiens · Jun 17, 2020 · fa81cbc · fa81cbc
1 parent 77664ec
commit fa81cbc
Showing 1 changed file with 1 addition and 0 deletions.
diff --git a/src/batch_consumer.ts b/src/batch_consumer.ts
@@ -148,6 +148,7 @@ export class KafkaBatchConsumer extends (EventEmitter as new () => TypedEmitter)
     if (this.consuming) {
       return false
     }
+    this.consumer.unsubscribe().subscribe(this.topics)
     this.consuming = true
     this.emit("consuming", true)
     return this.startPolling()