Pass up more cluster metadata

on-behalf-of: @SAP [email protected]

Author: lp247

src/clients/base/base.ts

@@ -600,8 +600,8 @@ export class Base<OptionsType extends BaseOptions = BaseOptions> extends EventEm
 
             // This should never change, but we act defensively here
             for (const broker of metadata.brokers) {
-              const { host, port } = broker
-              brokers.set(broker.nodeId, { host, port })
+              const { host, port, rack } = broker
+              brokers.set(broker.nodeId, { host, port, rack })
             }
 
             this.#metadata.brokers = brokers
@@ -619,7 +619,9 @@ export class Base<OptionsType extends BaseOptions = BaseOptions> extends EventEm
                 partitions[rawPartition.partitionIndex] = {
                   leader: rawPartition.leaderId,
                   leaderEpoch: rawPartition.leaderEpoch,
-                  replicas: rawPartition.replicaNodes
+                  replicas: rawPartition.replicaNodes,
+                  isr: rawPartition.isrNodes,
+                  offlineReplicas: rawPartition.offlineReplicas
                 }
               }
 

src/clients/base/types.ts

@@ -1,4 +1,5 @@
 import { type Broker, type ConnectionOptions } from '../../network/connection.ts'
+import { type NullableString } from '../../protocol/definitions.ts'
 import { type Metrics } from '../metrics.ts'
 
 export interface TopicWithPartitionAndOffset {
@@ -11,6 +12,8 @@ export interface ClusterPartitionMetadata {
   leader: number
   leaderEpoch: number
   replicas: number[]
+  isr: number[]
+  offlineReplicas: number[]
 }
 
 export interface ClusterTopicMetadata {
@@ -20,9 +23,13 @@ export interface ClusterTopicMetadata {
   lastUpdate: number
 }
 
+export interface BrokerWithRack extends Broker {
+  rack: NullableString
+}
+
 export interface ClusterMetadata {
   id: string
-  brokers: Map<number, Broker>
+  brokers: Map<number, BrokerWithRack>
   topics: Map<string, ClusterTopicMetadata>
   lastUpdate: number
 }

test/clients/base/sasl-oauthbearer.test.ts

@@ -50,7 +50,7 @@ test('should connect to SASL protected broker using SASL/OAUTHBEARER', async t =
 
   const metadata = await base.metadata({ topics: [] })
 
-  deepStrictEqual(metadata.brokers.get(1), saslBroker)
+  deepStrictEqual(metadata.brokers.get(1), { ...saslBroker, rack: null })
 })
 
 test('should handle authentication errors', async t => {
@@ -103,7 +103,7 @@ test('should accept a function as credential provider', async t => {
 
   const metadata = await base.metadata({ topics: [] })
 
-  deepStrictEqual(metadata.brokers.get(1), saslBroker)
+  deepStrictEqual(metadata.brokers.get(1), { ...saslBroker, rack: null })
 })
 
 test('should accept an async function as credential provider', async t => {
@@ -134,7 +134,7 @@ test('should accept an async function as credential provider', async t => {
 
   const metadata = await base.metadata({ topics: [] })
 
-  deepStrictEqual(metadata.brokers.get(1), saslBroker)
+  deepStrictEqual(metadata.brokers.get(1), { ...saslBroker, rack: null })
 })
 
 test('should handle sync credential provider errors', async t => {

test/clients/base/sasl.test.ts

@@ -49,7 +49,7 @@ for (const mechanism of allowedSASLMechanisms) {
 
     const metadata = await base.metadata({ topics: [] })
 
-    deepStrictEqual(metadata.brokers.get(1), saslBroker)
+    deepStrictEqual(metadata.brokers.get(1), { ...saslBroker, rack: null })
   })
 
   test(`${mechanism} - should handle authentication errors`, async t => {
@@ -90,7 +90,7 @@ for (const mechanism of allowedSASLMechanisms) {
 
     const metadata = await base.metadata({ topics: [] })
 
-    deepStrictEqual(metadata.brokers.get(1), saslBroker)
+    deepStrictEqual(metadata.brokers.get(1), { ...saslBroker, rack: null })
   })
 
   test(`${mechanism} - should accept an async function as credential provider`, async t => {
@@ -113,7 +113,7 @@ for (const mechanism of allowedSASLMechanisms) {
 
     const metadata = await base.metadata({ topics: [] })
 
-    deepStrictEqual(metadata.brokers.get(1), saslBroker)
+    deepStrictEqual(metadata.brokers.get(1), { ...saslBroker, rack: null })
   })
 
   test(`${mechanism} - should handle sync credential provider errors`, async t => {