Merge pull request #2452 from pikasTech/fix/kafka-cli-tail-stop

fix: stop kafka query consumers

internal/cloud/kafka-event-bridge.ts

@@ -157,12 +157,20 @@ export async function queryKafkaEventStream({ env = process.env, stream = "hwlab
   const consumer = kafka.consumer({ groupId: `${clientId}-query-${Date.now()}-${randomUUID().slice(0, 8)}`, allowAutoTopicCreation: false });
   const events = [];
   let timer = null;
+  let running = false;
   await consumer.connect();
   await consumer.subscribe({ topic: resolvedTopic, fromBeginning: fromBeginning !== false });
   try {
     await new Promise((resolve, reject) => {
-      timer = setTimeout(resolve, budgetMs);
+      let finished = false;
+      const finish = (value = null) => {
+        if (finished) return;
+        finished = true;
+        resolve(value);
+      };
+      timer = setTimeout(finish, budgetMs);
       timer.unref?.();
+      running = true;
       consumer.run({
         eachMessage: async ({ topic: messageTopic, partition, message }) => {
           const valueText = message.value ? Buffer.from(message.value).toString("utf8") : "";
@@ -176,13 +184,14 @@ export async function queryKafkaEventStream({ env = process.env, stream = "hwlab
             timestamp: message.timestamp ?? null,
             value
           });
-          if (events.length >= maxEvents) resolve(null);
+          if (events.length >= maxEvents) finish();
         }
       }).catch(reject);
     });
   } finally {
     if (timer) clearTimeout(timer);
-    await consumer.disconnect();
+    if (running) await consumer.stop().catch(() => undefined);
+    await boundedDisconnect(consumer);
   }
   return {
     ok: true,
@@ -270,6 +279,21 @@ function defaultKafkaFactory(config) {
   return new Kafka({ clientId: config.clientId, brokers: config.brokers, logLevel: logLevel.NOTHING });
 }
 
+async function boundedDisconnect(consumer, timeoutMs = 2000) {
+  let timer = null;
+  try {
+    await Promise.race([
+      consumer.disconnect(),
+      new Promise((resolve) => {
+        timer = setTimeout(resolve, timeoutMs);
+        timer.unref?.();
+      })
+    ]);
+  } finally {
+    if (timer) clearTimeout(timer);
+  }
+}
+
 function kafkaMessageKey(projected) {
   return firstText(projected.traceId, projected.sessionId, projected.context?.commandId, projected.context?.runId) || "hwlab-event";
 }