feat: expose cluster ID via getClusterId() and describeCluster()

lib/client.js

@@ -580,6 +580,23 @@ Client.prototype.setSaslCredentials = function(username, password) {
   this._client.setSaslCredentials(username, password);
 };
 
+/**
+ * Get the cluster ID reported by the broker metadata.
+ *
+ * Returns null if the client is not connected or if the cluster ID
+ * could not be retrieved within the given timeout.
+ *
+ * @param {number} timeout - Timeout in milliseconds (default: 200)
+ * @returns {string|null} - The cluster ID, or null
+ */
+Client.prototype.getClusterId = function(timeout) {
+  if (!this.isConnected()) {
+    return null;
+  }
+
+  return this._client.getClusterId(timeout) || null;
+};
+
 /**
  * Wrap a potential RdKafka error.
  *

lib/kafkajs/_admin.js

@@ -513,6 +513,55 @@ class Admin {
     });
   }
 
+  /**
+   * Describe the Kafka cluster.
+   *
+   * Returns the cluster ID, the controller broker, and the list
+   * of brokers in the cluster.
+   *
+   * @param {object?} options
+   * @param {number?} options.timeout - Timeout in ms (default: 5000).
+   * @returns {Promise<{clusterId: string, controller: number,
+   *           brokers: Array<{nodeId: number, host: string, port: number}>}>}
+   */
+  async describeCluster(options = {}) {
+    if (this.#state !== AdminState.CONNECTED) {
+      throw new error.KafkaJSError(
+        "Admin client is not connected.",
+        { code: error.ErrorCodes.ERR__STATE }
+      );
+    }
+
+    const timeout = options.timeout ?? 5000;
+
+    return new Promise((resolve, reject) => {
+      this.#internalClient.getMetadata(
+        { allTopics: true, timeout },
+        (err, metadata) => {
+          if (err) {
+            reject(createKafkaJsErrorFromLibRdKafkaError(err));
+          } else {
+            // Metadata has been fetched, so clusterid() returns
+            // from cache instantly with no blocking.
+            const clusterId =
+              this.#internalClient.getClusterId(0) || null;
+            resolve({
+              clusterId,
+              controller:
+                metadata.orig_broker_id != null
+                  ? metadata.orig_broker_id : null,
+              brokers: (metadata.brokers || []).map(b => ({
+                nodeId: b.id,
+                host: b.host,
+                port: b.port,
+              })),
+            });
+          }
+        }
+      );
+    });
+  }
+
   /**
    * Fetch the offsets for topic partition(s) for consumer group(s).
    *

src/admin.cc

@@ -136,6 +136,7 @@ void AdminClient::Init(v8::Local<v8::Object> exports) {
   Nan::SetPrototypeMethod(tpl, "disconnect", NodeDisconnect);
   Nan::SetPrototypeMethod(tpl, "setSaslCredentials", NodeSetSaslCredentials);
   Nan::SetPrototypeMethod(tpl, "getMetadata", NodeGetMetadata);
+  Nan::SetPrototypeMethod(tpl, "getClusterId", NodeGetClusterId);
   Nan::SetPrototypeMethod(tpl, "setOAuthBearerToken", NodeSetOAuthBearerToken);
   Nan::SetPrototypeMethod(tpl, "setOAuthBearerTokenFailure",
                           NodeSetOAuthBearerTokenFailure);

src/connection.cc

@@ -700,4 +700,30 @@ NAN_METHOD(Connection::NodeName) {
   info.GetReturnValue().Set(Nan::New(name).ToLocalChecked());
 }
 
+NAN_METHOD(Connection::NodeGetClusterId) {
+  Connection* obj = ObjectWrap::Unwrap<Connection>(info.This());
+
+  int timeout_ms = 200;
+  if (info[0]->IsNumber()) {
+    Nan::Maybe<int64_t> maybeTimeout = Nan::To<int64_t>(info[0]);
+    if (!maybeTimeout.IsNothing()) {
+      timeout_ms = static_cast<int>(maybeTimeout.FromJust());
+    }
+  }
+
+  if (!obj->IsConnected()) {
+    info.GetReturnValue().Set(Nan::Null());
+    return;
+  }
+
+  std::string cluster_id = obj->m_client->clusterid(timeout_ms);
+  if (cluster_id.empty()) {
+    info.GetReturnValue().Set(Nan::Null());
+    return;
+  }
+
+  info.GetReturnValue().Set(
+    Nan::New<v8::String>(cluster_id).ToLocalChecked());
+}
+
 }  // namespace NodeKafka

src/connection.h

@@ -110,6 +110,7 @@ class Connection : public Nan::ObjectWrap {
   static NAN_METHOD(NodeSetOAuthBearerToken);
   static NAN_METHOD(NodeSetOAuthBearerTokenFailure);
   static NAN_METHOD(NodeName);
+  static NAN_METHOD(NodeGetClusterId);
 };
 
 }  // namespace NodeKafka

src/kafka-consumer.cc

@@ -546,6 +546,7 @@ void KafkaConsumer::Init(v8::Local<v8::Object> exports) {
   Nan::SetPrototypeMethod(tpl, "connect", NodeConnect);
   Nan::SetPrototypeMethod(tpl, "disconnect", NodeDisconnect);
   Nan::SetPrototypeMethod(tpl, "getMetadata", NodeGetMetadata);
+  Nan::SetPrototypeMethod(tpl, "getClusterId", NodeGetClusterId);
   Nan::SetPrototypeMethod(tpl, "queryWatermarkOffsets", NodeQueryWatermarkOffsets);  // NOLINT
   Nan::SetPrototypeMethod(tpl, "offsetsForTimes", NodeOffsetsForTimes);
   Nan::SetPrototypeMethod(tpl, "getWatermarkOffsets", NodeGetWatermarkOffsets);

src/producer.cc

@@ -70,6 +70,7 @@ void Producer::Init(v8::Local<v8::Object> exports) {
   Nan::SetPrototypeMethod(tpl, "connect", NodeConnect);
   Nan::SetPrototypeMethod(tpl, "disconnect", NodeDisconnect);
   Nan::SetPrototypeMethod(tpl, "getMetadata", NodeGetMetadata);
+  Nan::SetPrototypeMethod(tpl, "getClusterId", NodeGetClusterId);
   Nan::SetPrototypeMethod(tpl, "queryWatermarkOffsets", NodeQueryWatermarkOffsets);  // NOLINT
   Nan::SetPrototypeMethod(tpl, "poll", NodePoll);
   Nan::SetPrototypeMethod(tpl, "setPollInBackground", NodeSetPollInBackground);

test/promisified/admin/describe_cluster.spec.js

@@ -0,0 +1,47 @@
+jest.setTimeout(30000);
+
+const {
+    createAdmin,
+} = require('../testhelpers');
+
+describe('Admin > describeCluster', () => {
+    let admin;
+
+    beforeEach(async () => {
+        admin = createAdmin({});
+    });
+
+    afterEach(async () => {
+        admin && (await admin.disconnect());
+    });
+
+    it('should fail if not connected', async () => {
+        await expect(admin.describeCluster()).rejects.toHaveProperty(
+            'code',
+            -172 // ERR__STATE
+        );
+    });
+
+    it('should describe the cluster', async () => {
+        await admin.connect();
+
+        const result = await admin.describeCluster();
+
+        expect(result).toEqual(
+            expect.objectContaining({
+                clusterId: expect.any(String),
+                controller: expect.any(Number),
+                brokers: expect.arrayContaining([
+                    expect.objectContaining({
+                        nodeId: expect.any(Number),
+                        host: expect.any(String),
+                        port: expect.any(Number),
+                    }),
+                ]),
+            })
+        );
+
+        expect(result.clusterId.length).toBeGreaterThan(0);
+        expect(result.brokers.length).toBeGreaterThan(0);
+    });
+});