fix: Fixed lag computation and describeGroup response handling. (#153)

* fix: Fixed lag computation and describeGroup response handling.

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

Author: ShogunPanda

src/clients/admin/admin.ts

@@ -85,6 +85,7 @@ import {
   type DescribeLogDirsOptions,
   type Group,
   type GroupBase,
+  type GroupMember,
   type ListGroupsOptions,
   type ListTopicsOptions
 } from './types.ts'
@@ -705,26 +706,31 @@ export class Admin extends Base<AdminOptions> {
                 for (const member of raw.members) {
                   const reader = Reader.from(member.memberMetadata)
 
-                  const memberMetadata = {
-                    version: reader.readInt16(),
-                    topics: reader.readArray(r => r.readString(false), false, false),
-                    metadata: reader.readBytes(false)
-                  }
+                  let memberMetadata: GroupMember['metadata'] | undefined
+                  let memberAssignments: Map<string, GroupAssignment> | undefined
+
+                  if (reader.remaining > 0) {
+                    memberMetadata = {
+                      version: reader.readInt16(),
+                      topics: reader.readArray(r => r.readString(false), false, false),
+                      metadata: reader.readBytes(false)
+                    }
 
-                  reader.reset(member.memberAssignment)
-                  reader.skip(2) // Ignore Version information
+                    reader.reset(member.memberAssignment)
+                    reader.skip(2) // Ignore Version information
 
-                  const memberAssignments: Map<string, GroupAssignment> = reader.readMap(
-                    r => {
-                      const topic = r.readString(false)
+                    memberAssignments = reader.readMap(
+                      r => {
+                        const topic = r.readString(false)
 
-                      return [topic, { topic, partitions: reader.readArray(r => r.readInt32(), false, false) }]
-                    },
-                    false,
-                    false
-                  )
+                        return [topic, { topic, partitions: reader.readArray(r => r.readInt32(), false, false) }]
+                      },
+                      false,
+                      false
+                    )
 
-                  reader.readBytes() // Ignore the user data
+                    reader.readBytes() // Ignore the user data
+                  }
 
                   group.members.set(member.memberId, {
                     id: member.memberId,

src/clients/admin/types.ts

@@ -1,11 +1,11 @@
 import { type AlterClientQuotasRequestEntry } from '../../apis/admin/alter-client-quotas-v1.ts'
+import { type CreateTopicsRequestTopicConfig } from '../../apis/admin/create-topics-v7.ts'
 import { type DescribeClientQuotasRequestComponent } from '../../apis/admin/describe-client-quotas-v0.ts'
 import {
+  type DescribeLogDirsRequestTopic,
   type DescribeLogDirsResponse,
-  type DescribeLogDirsResponseResult,
-  type DescribeLogDirsRequestTopic
+  type DescribeLogDirsResponseResult
 } from '../../apis/admin/describe-log-dirs-v4.ts'
-import { type CreateTopicsRequestTopicConfig } from '../../apis/admin/create-topics-v7.ts'
 import { type ConsumerGroupState } from '../../apis/enumerations.ts'
 import { type NullableString } from '../../protocol/definitions.ts'
 import { type BaseOptions } from '../base/types.ts'
@@ -29,8 +29,8 @@ export interface GroupMember {
   groupInstanceId: NullableString
   clientId: string
   clientHost: string
-  metadata: Omit<ExtendedGroupProtocolSubscription, 'memberId'>
-  assignments: Map<string, GroupAssignment>
+  metadata?: Omit<ExtendedGroupProtocolSubscription, 'memberId'>
+  assignments?: Map<string, GroupAssignment>
 }
 
 export interface GroupBase {

src/clients/consumer/consumer.ts

@@ -520,7 +520,7 @@ export class Consumer<Key = Buffer, Value = Buffer, HeaderKey = Buffer, HeaderVa
       // Now gather the last committed offsets from each stream
       const committeds = new Map<string, bigint>()
       for (const stream of this.#streams) {
-        for (const [topic, offset] of stream.committedOffsets) {
+        for (const [topic, offset] of stream.offsetsCommitted) {
           committeds.set(topic, offset)
         }
       }

src/clients/consumer/messages-stream.ts

@@ -146,6 +146,8 @@ export class MessagesStream<Key, Value, HeaderKey, HeaderValue> extends Readable
     // will have changed so we may have gone from last  with no assignments to
     // having some.
     this.#consumer.on('consumer:group:join', () => {
+      this.#offsetsCommitted.clear()
+
       this.#refreshOffsets((error: Error | null) => {
         /* c8 ignore next 4 - Hard to test */
         if (error) {
@@ -169,6 +171,19 @@ export class MessagesStream<Key, Value, HeaderKey, HeaderValue> extends Readable
     notifyCreation('messages-stream', this)
   }
 
+  get offsetsToFetch (): Map<string, bigint> {
+    return this.#offsetsToFetch
+  }
+
+  get offsetsToCommit (): Map<string, CommitOptionsPartition> {
+    return this.#offsetsToCommit
+  }
+
+  get offsetsCommitted (): Map<string, bigint> {
+    return this.#offsetsCommitted
+  }
+
+  // TODO: This is deprecated alias, remove in future major version
   get committedOffsets (): Map<string, bigint> {
     return this.#offsetsCommitted
   }
@@ -279,6 +294,7 @@ export class MessagesStream<Key, Value, HeaderKey, HeaderValue> extends Readable
   */
   addListener (event: 'autocommit', listener: (err: Error, offsets: CommitOptionsPartition[]) => void): this
   addListener (event: 'fetch', listener: () => void): this
+  addListener (event: 'offsets', listener: () => void): this
   addListener (event: 'data', listener: (message: Message<Key, Value, HeaderKey, HeaderValue>) => void): this
   addListener (event: 'close', listener: () => void): this
   addListener (event: 'end', listener: () => void): this
@@ -293,6 +309,7 @@ export class MessagesStream<Key, Value, HeaderKey, HeaderValue> extends Readable
 
   on (event: 'autocommit', listener: (err: Error, offsets: CommitOptionsPartition[]) => void): this
   on (event: 'fetch', listener: () => void): this
+  on (event: 'offsets', listener: () => void): this
   on (event: 'data', listener: (message: Message<Key, Value, HeaderKey, HeaderValue>) => void): this
   on (event: 'close', listener: () => void): this
   on (event: 'end', listener: () => void): this
@@ -307,6 +324,7 @@ export class MessagesStream<Key, Value, HeaderKey, HeaderValue> extends Readable
 
   once (event: 'autocommit', listener: (err: Error, offsets: CommitOptionsPartition[]) => void): this
   once (event: 'fetch', listener: () => void): this
+  once (event: 'offsets', listener: () => void): this
   once (event: 'data', listener: (message: Message<Key, Value, HeaderKey, HeaderValue>) => void): this
   once (event: 'close', listener: () => void): this
   once (event: 'end', listener: () => void): this
@@ -321,6 +339,7 @@ export class MessagesStream<Key, Value, HeaderKey, HeaderValue> extends Readable
 
   prependListener (event: 'autocommit', listener: (err: Error, offsets: CommitOptionsPartition[]) => void): this
   prependListener (event: 'fetch', listener: () => void): this
+  prependListener (event: 'offsets', listener: () => void): this
   prependListener (event: 'data', listener: (message: Message<Key, Value, HeaderKey, HeaderValue>) => void): this
   prependListener (event: 'close', listener: () => void): this
   prependListener (event: 'end', listener: () => void): this
@@ -335,6 +354,7 @@ export class MessagesStream<Key, Value, HeaderKey, HeaderValue> extends Readable
 
   prependOnceListener (event: 'autocommit', listener: (err: Error, offsets: CommitOptionsPartition[]) => void): this
   prependOnceListener (event: 'fetch', listener: () => void): this
+  prependOnceListener (event: 'offsets', listener: () => void): this
   prependOnceListener (event: 'data', listener: (message: Message<Key, Value, HeaderKey, HeaderValue>) => void): this
   prependOnceListener (event: 'close', listener: () => void): this
   prependOnceListener (event: 'end', listener: () => void): this
@@ -774,6 +794,24 @@ export class MessagesStream<Key, Value, HeaderKey, HeaderValue> extends Readable
       }
     }
 
+    // Rebuild the list of offsetsCommitted (which is used for consumer lag) out of the offsets to fetch
+    for (const topic of this.#topics) {
+      const assignment = this.#assignmentsForTopic(topic)
+
+      // This consumer has no assignment for the topic, continue
+      if (!assignment) {
+        continue
+      }
+
+      const partitions = assignment.partitions
+
+      for (const partition of partitions) {
+        const committed = this.#offsetsToFetch.get(`${topic}:${partition}`)!
+        this.#offsetsCommitted.set(`${topic}:${partition}`, committed - 1n)
+      }
+    }
+
+    this.emit('offsets')
     callback(null)
   }
 

test/clients/consumer/consumer.test.ts

@@ -2188,22 +2188,25 @@ test('getLag should allow to filter out topics and partitions', async t => {
   const topic = await createTopic(t, true, 3)
   const consumer = createConsumer(t)
 
-  await consumer.consume({
+  const stream = await consumer.consume({
     topics: [topic],
     autocommit: true,
     mode: 'earliest',
     maxWaitTime: 1000,
     maxBytes: 10
   })
 
+  // Wait for the topics to be fetched
+  await once(stream, 'offsets')
+
   await new Promise<void>((resolve, reject) => {
     consumer.getLag({ topics: [topic], partitions: { [topic]: [0, 2] } }, (err, offsets) => {
       if (err) {
         reject(err)
         return
       }
 
-      deepStrictEqual(offsets, new Map([[topic, [-1n, -2n, -1n]]]))
+      deepStrictEqual(offsets, new Map([[topic, [0n, -2n, 0n]]]))
       resolve()
     })
   })
@@ -2364,9 +2367,9 @@ test('startLagMonitoring should regularly check consumer lag', async t => {
   unsubscribe(consumerLagChannel.name, onLag as ChannelListener)
 
   deepStrictEqual(lagsViaEvent, [
-    new Map([[topic, [-1n, -1n, -1n]]]),
-    new Map([[topic, [-1n, -1n, -1n]]]),
-    new Map([[topic, [-1n, -1n, -1n]]])
+    new Map([[topic, [-1n, -1n, -1n]]]), // Still joining and not assigned to any partition
+    new Map([[topic, [0n, 0n, 0n]]]),
+    new Map([[topic, [0n, 0n, 0n]]])
   ])
 })