feat: Fetch all topics offsets when no topics is provided. (#151)

Signed-off-by: Paolo Insogna <[email protected]>

Author: ShogunPanda

docs/consumer.md

@@ -140,6 +140,8 @@ Options:
 
 Lists available offsets for topics.
 
+If the topics list is empty, then it will fetch the offsets of all topics currently consumed by the consumer.
+
 The return value is a map where keys are in the form `$topic:$partition` and values are arrays of offsets (where the position represents the partition).
 
 Options:
@@ -167,6 +169,8 @@ Options:
 
 Calculates the consumer lag for specified topics.
 
+If the topics list is empty, then it will calculate the lag of all topics currently consumed by the consumer.
+
 The return value is a map where keys are topic names and values are arrays of lag values (where the position represents the partition). A value of `-1n` indicates that the consumer is not assigned to that partition.
 
 If a partition is filtered out via the `partitions` option, then a `-2n` will be returned for that partition.

src/clients/consumer/consumer.ts

@@ -799,15 +799,21 @@ export class Consumer<Key = Buffer, Value = Buffer, HeaderKey = Buffer, HeaderVa
     options: ListOffsetsOptions,
     callback: CallbackWithPromise<Offsets | OffsetsWithTimestamps>
   ): void {
-    this[kMetadata]({ topics: options.topics }, (error, metadata) => {
+    let topics = options.topics
+
+    if (!topics || topics.length === 0) {
+      topics = this.topics.current
+    }
+
+    this[kMetadata]({ topics }, (error, metadata) => {
       if (error) {
         callback(error, undefined as unknown as Offsets)
         return
       }
 
       const requests = new Map<number, Map<string, ListOffsetsRequestTopic>>()
 
-      for (const name of options.topics) {
+      for (const name of topics) {
         const topic = metadata.topics.get(name)!
         const toInclude = new Set(options.partitions?.[name] ?? [])
         const hasPartitionsFilter = toInclude.size > 0

test/clients/consumer/consumer.test.ts

@@ -1599,6 +1599,46 @@ test('listOffsets should return offset values for topics and partitions and supp
   verifyTracingChannel()
 })
 
+test('listOffsets should return offset values for all consumed topics when none are explicitly requested', async t => {
+  const consumer = createConsumer(t)
+  const topic = await createTopic(t, true, 2)
+
+  const verifyTracingChannel = createTracingChannelVerifier(consumerOffsetsChannel, 'client', {
+    start (context: ClientDiagnosticEvent) {
+      deepStrictEqual(context, {
+        client: consumer,
+        operation: 'listOffsets',
+        options: { topics: [] },
+        operationId: mockedOperationId
+      })
+    },
+    asyncStart (context: ClientDiagnosticEvent) {
+      deepStrictEqual((context.result as Offsets).get(topic), [0n, 0n])
+    },
+    error (context: ClientDiagnosticEvent) {
+      ok(typeof context === 'undefined')
+    }
+  })
+
+  // Get offsets for the test topic
+  consumer.topics.track(topic)
+  const offsets = await consumer.listOffsets({ topics: [] })
+
+  // Verify the offsets structure
+  strictEqual(offsets instanceof Map, true, 'Should return a Map of offsets')
+  strictEqual(offsets.has(topic), true, 'Should contain the requested topic')
+
+  const topicOffsets = offsets.get(topic)!
+  strictEqual(Array.isArray(topicOffsets), true, 'Topic offsets should be an array')
+  strictEqual(topicOffsets.length, 2, 'Should have offsets for all partitions')
+
+  // For new topics, offsets should typically be 0
+  strictEqual(typeof topicOffsets[0], 'bigint', 'Offset should be a bigint')
+  strictEqual(typeof topicOffsets[1], 'bigint', 'Offset should be a bigint')
+
+  verifyTracingChannel()
+})
+
 test('listOffsets should support both promise and callback API', async t => {
   const consumer = createConsumer(t)
   const topic = await createTopic(t, true)