fix: call fetch on resume if necessary

Signed-off-by: Mirza Brunjadze <[email protected]>

Author: xxzefgh

src/clients/consumer/consumer.ts

@@ -325,19 +325,26 @@ export class Consumer<Key = Buffer, Value = Buffer, HeaderKey = Buffer, HeaderVa
       throw new UserError('Cannot resume partitions before joining a consumer group.')
     }
 
+    let emitResumeEvent = false
     for (const { topic, partition } of partitions) {
       const assignment = this.assignments.find(a => a.topic === topic)
       if (!assignment) {
         throw new UserError(`Topic '${topic}' is not assigned to this consumer.`, { topic })
       }
 
       const existing = this.#pausedPartitions.get(topic)
-      if (existing) {
+      if (existing?.has(partition)) {
+        emitResumeEvent = true
         existing.delete(partition)
-        if (existing.size === 0) {
-          this.#pausedPartitions.delete(topic)
-        }
       }
+
+      if (existing?.size === 0) {
+        this.#pausedPartitions.delete(topic)
+      }
+    }
+
+    if (emitResumeEvent) {
+      this.emitWithDebug('consumer', 'user:resume', { partitions })
     }
   }
 

src/clients/consumer/messages-stream.ts

@@ -144,6 +144,14 @@ export class MessagesStream<Key, Value, HeaderKey, HeaderValue> extends Readable
       })
     })
 
+    this.#consumer.on('consumer:user:resume', () => {
+      if (this.#inflightNodes.size > 0 || this.#shouldClose || this.closed || this.destroyed) {
+        return
+      }
+
+      this.#fetch()
+    })
+
     if (consumer[kPrometheus]) {
       this.#metricsConsumedMessages = ensureMetric<Counter>(
         consumer[kPrometheus],
@@ -563,9 +571,7 @@ export class MessagesStream<Key, Value, HeaderKey, HeaderValue> extends Readable
     }
 
     if (canPush && !(this.#shouldClose || this.closed || this.destroyed)) {
-      process.nextTick(() => {
-        this.#fetch()
-      })
+      this.#fetch()
     }
   }
 

test/clients/consumer/consumer.test.ts

@@ -3270,11 +3270,11 @@ test('pause should prevent fetches from paused partitions during consumption', a
   const messages = Array.from({ length: 10 }, (_, i) => ({
     topic,
     key: `key-${i}`,
-    value: `value-${i}`,
-    partition: i % 2
+    value: `value-${i}`
   }))
 
-  await produceTestMessages({ t, messages })
+  await produceTestMessages({ t, messages: messages.slice(0, 5), overrideOptions: { partitioner: () => 0 } })
+  await produceTestMessages({ t, messages: messages.slice(5, 10), overrideOptions: { partitioner: () => 1 } })
 
   const consumer = createConsumer(t)
   const stream = await consumer.consume({