fixes consumer pull disconnect

3axap4eHko · Mar 28, 2024 · 52f275c · 52f275c
1 parent 1ad718f
commit 52f275c
Showing 1 changed file with 6 additions and 6 deletions.
diff --git a/src/utils/kafka.ts b/src/utils/kafka.ts
@@ -170,17 +170,17 @@ export async function createConsumer(client: Kafka, group: string, topic: string
   await consumer.connect();
   await consumer.subscribe(consumerOptions);
 
-  const pool = new Pool<EachMessagePayload>([], poolOptions);
-  pool.onDone(() => {
-    consumer.disconnect();
-  });
-
-  consumer.run({
+  await consumer.run({
     eachMessage: async (payload) => {
       pool.push(payload);
     },
   }).catch(e => console.error(e));
 
+  const pool = new Pool<EachMessagePayload>([], poolOptions);
+  pool.onDone(() => {
+    consumer.disconnect();
+  });
+
   return pool;
 }