DescribeTopics api implemented (#155)

* describeTopics api implemented

* addressing comments

* addressing lint errors

* requested changes

* requested changes

* reverting space indent

Author: PratRanj07

CHANGELOG.md

@@ -7,6 +7,7 @@ v0.5.0 is a limited availability feature release. It is supported for all usage.
 
 1. Add support for an Admin API to delete records.(#141).
 2. Fixes an issue with unresolved raced Promises leaking in the consumer (#151).
+3. Add support for an Admin API to describe topics.(#155).
 
 
 # confluent-kafka-javascript v0.4.0

MIGRATION.md

@@ -335,6 +335,8 @@ The admin-client only has support for a limited subset of methods, with more to
   `requireStableOffsets` option but `resolveOffsets` option is not yet supported.
   * The `deleteTopicRecords` method is supported with additional `timeout`
   and `operationTimeout` option.
+  * The `fetchTopicMetadata` method is supported with additional `timeout`
+  and `includeAuthorizedOperations` option. Fetching for all topics is not advisable.
 
 ### Using the Schema Registry
 

examples/kafkajs/admin/fetch-topic-metadata.js

@@ -0,0 +1,77 @@
+const { Kafka } = require('@confluentinc/kafka-javascript').KafkaJS;
+const { parseArgs } = require('node:util');
+
+async function fetchMetadata() {
+    // Parse command-line arguments
+    const args = parseArgs({
+        allowPositionals: true,
+        options: {
+            'bootstrap-servers': {
+                type: 'string',
+                short: 'b',
+                default: 'localhost:9092',
+            },
+            'timeout': {
+                type: 'string',
+                short: 't',
+                default: '5000',
+            },
+            'include-authorized-operations': {
+                type: 'boolean',
+                short: 'a',
+                default: false,
+            },
+        },
+    });
+
+    const {
+        'bootstrap-servers': bootstrapServers,
+        timeout,
+        'include-authorized-operations': includeAuthorizedOperations,
+    } = args.values;
+
+    const [topicName] = args.positionals;
+
+    if (!topicName) {
+        console.error('Topic name is required');
+        process.exit(1);
+    }
+
+    const kafka = new Kafka({
+        kafkaJS: {
+            brokers: [bootstrapServers],
+        },
+    });
+
+    const admin = kafka.admin();
+    await admin.connect();
+
+    try {
+        // Fetch the topic metadata with specified options
+        const metadata = await admin.fetchTopicMetadata(
+            {
+                topics: [topicName],
+                includeAuthorizedOperations: includeAuthorizedOperations,
+                timeout: Number(timeout), // Convert timeout to a number
+            });
+
+        console.log(`Metadata for topic "${topicName}":`, stringifyBigInt(metadata));
+    } catch (err) {
+        console.error('Error fetching topic metadata:', err);
+    } finally {
+        await admin.disconnect();
+    }
+}
+
+function stringifyBigInt(obj) {
+    return JSON.stringify(
+        obj,
+        (key, value) =>
+            typeof value === 'bigint'
+                ? value.toString() // Convert BigInt to string
+                : value,
+        2
+    );
+}
+
+fetchMetadata();

lib/admin.js

@@ -493,7 +493,7 @@ AdminClient.prototype.listConsumerGroupOffsets = function (listGroupOffsets, opt
  *
  * @param {import("../../types/rdkafka").TopicPartitionOffset[]} delRecords - The list of topic partitions and
  *                                                                            offsets to delete records up to.
- * @param {number} options.operationTimeout - The operation timeout in milliseconds.
+ * @param {number?} options.operationTimeout - The operation timeout in milliseconds.
  *                                             May be unset (default: 60000)
  * @param {number?} options.timeout - The request timeout in milliseconds.
  *                                    May be unset (default: 5000)
@@ -534,3 +534,44 @@ AdminClient.prototype.deleteRecords = function (delRecords, options, cb) {
     }
   });
 };
+
+/**
+ * Describe Topics.
+ *
+ * @param {string[]} topics - The names of the topics to describe.
+ * @param {number?} options.timeout - The request timeout in milliseconds.
+ *                                    May be unset (default: 5000)
+ * @param {boolean?} options.includeAuthorizedOperations - If true, include operations allowed on the topic by the calling client.
+ *                                                         (default: false)
+ * @param {function} cb - The callback to be executed when finished.
+ */
+AdminClient.prototype.describeTopics = function (topics, options, cb) {
+  if (!this._isConnected) {
+    throw new Error('Client is disconnected');
+  }
+
+  if(!options) {
+    options = {};
+  }
+
+  if(!Object.hasOwn(options, 'timeout')) {
+    options.timeout = 5000;
+  }
+
+  if(!Object.hasOwn(options, 'includeAuthorizedOperations')) {
+    options.includeAuthorizedOperations = false;
+  }
+
+  this._client.describeTopics(topics, options, function (err, descriptions) {
+    if (err) {
+      if (cb) {
+        cb(LibrdKafkaError.create(err));
+      }
+      return;
+    }
+
+    if (cb) {
+      cb(null, descriptions);
+    }
+  });
+};

lib/kafkajs/_admin.js

@@ -541,7 +541,7 @@ class Admin {
    *
    * @param {string} options.topic - The topic to delete offsets for.
    * @param {import("../../types/kafkajs").SeekEntry[]} options.partitons - The partitions to delete offsets for.
-   * @param {number} options.operationTimeout - The operation timeout in milliseconds.
+   * @param {number?} options.operationTimeout - The operation timeout in milliseconds.
    *                                             May be unset (default: 60000)
    * @param {number?} options.timeout - The request timeout in milliseconds.
    *                                    May be unset (default: 5000)
@@ -589,6 +589,66 @@ class Admin {
       });
     });
   }
+
+  /**
+   * Describe topics.
+   *
+   * @param {string[]} options.topics - The topics to describe.
+   *                                    If unset, all topics will be described.
+   * @param {number?} options.timeout - The request timeout in milliseconds.
+   *                                    May be unset (default: 5000)
+   * @param {boolean?} options.includeAuthorizedOperations - If true, include operations allowed on the topic
+   *                                                         by the calling client (default: false).
+   *
+   * @returns {Promise<{ topics: Array<import('../../types/kafkajs').ITopicMetadata> }>}
+   */
+  async fetchTopicMetadata(options = {}) {
+    if (this.#state !== AdminState.CONNECTED) {
+      throw new error.KafkaJSError("Admin client is not connected.", { code: error.ErrorCodes.ERR__STATE });
+    }
+
+    let topics = options.topics;
+    if (!Object.hasOwn(options, 'topics')) {
+      try {
+        topics = await this.listTopics();
+      } catch (err) {
+        throw createKafkaJsErrorFromLibRdKafkaError(err);
+      }
+    }
+
+    return new Promise((resolve, reject) => {
+      this.#internalClient.describeTopics(topics, options, (err, metadata) => {
+        if (err) {
+          reject(createKafkaJsErrorFromLibRdKafkaError(err));
+        } else {
+
+          let errs = metadata.filter(topic => topic.error);
+          if (errs.length > 0) {
+            reject(createKafkaJsErrorFromLibRdKafkaError(errs[0].error));
+            return;
+          }
+          const convertedMetadata = metadata.map(topic => ({
+            name: topic.name,
+            topicId: topic.topicId,
+            isInternal: topic.isInternal,
+            partitions: topic.partitions.map(partition => ({
+              partitionErrorCode: error.ErrorCodes.ERR_NO_ERROR,
+              partitionId: partition.partition,
+              leader: partition.leader.id,
+              leaderNode: partition.leader,
+              replicas: partition.replicas.map(replica => replica.id),
+              replicaNodes: partition.replicas,
+              isr: partition.isr.map(isrNode => isrNode.id),
+              isrNodes: partition.isr
+            })),
+            authorizedOperations: topic.authorizedOperations
+          }));
+
+          resolve(convertedMetadata);
+        }
+      });
+    });
+  }
 }
 
 module.exports = {

src/admin.cc

@@ -116,6 +116,7 @@ void AdminClient::Init(v8::Local<v8::Object> exports) {
   Nan::SetPrototypeMethod(tpl, "deleteTopic", NodeDeleteTopic);
   Nan::SetPrototypeMethod(tpl, "createPartitions", NodeCreatePartitions);
   Nan::SetPrototypeMethod(tpl, "deleteRecords", NodeDeleteRecords);
+  Nan::SetPrototypeMethod(tpl, "describeTopics", NodeDescribeTopics);
 
   // Consumer group related operations
   Nan::SetPrototypeMethod(tpl, "listGroups", NodeListGroups);
@@ -812,6 +813,76 @@ Baton AdminClient::DeleteRecords(rd_kafka_DeleteRecords_t **del_records,
   }
 }
 
+Baton AdminClient::DescribeTopics(rd_kafka_TopicCollection_t *topics,
+                                  bool include_authorized_operations,
+                                  int timeout_ms,
+                                  rd_kafka_event_t **event_response) {
+  if (!IsConnected()) {
+    return Baton(RdKafka::ERR__STATE);
+  }
+
+  {
+    scoped_shared_write_lock lock(m_connection_lock);
+    if (!IsConnected()) {
+      return Baton(RdKafka::ERR__STATE);
+    }
+
+    // Make admin options to establish that we are describing topics
+    rd_kafka_AdminOptions_t *options = rd_kafka_AdminOptions_new(
+        m_client->c_ptr(), RD_KAFKA_ADMIN_OP_DESCRIBETOPICS);
+
+    if (include_authorized_operations) {
+      rd_kafka_error_t *error =
+          rd_kafka_AdminOptions_set_include_authorized_operations(
+              options, include_authorized_operations);
+      if (error) {
+        return Baton::BatonFromErrorAndDestroy(error);
+      }
+    }
+
+    char errstr[512];
+    rd_kafka_resp_err_t err = rd_kafka_AdminOptions_set_request_timeout(
+        options, timeout_ms, errstr, sizeof(errstr));
+    if (err != RD_KAFKA_RESP_ERR_NO_ERROR) {
+      return Baton(static_cast<RdKafka::ErrorCode>(err), errstr);
+    }
+
+    // Create queue just for this operation.
+    rd_kafka_queue_t *rkqu = rd_kafka_queue_new(m_client->c_ptr());
+
+    rd_kafka_DescribeTopics(m_client->c_ptr(), topics, options, rkqu);
+
+    // Poll for an event by type in that queue
+    // DON'T destroy the event. It is the out parameter, and ownership is
+    // the caller's.
+    *event_response =
+        PollForEvent(rkqu, RD_KAFKA_EVENT_DESCRIBETOPICS_RESULT, timeout_ms);
+
+    // Destroy the queue since we are done with it.
+    rd_kafka_queue_destroy(rkqu);
+
+    // Destroy the options we just made because we polled already
+    rd_kafka_AdminOptions_destroy(options);
+
+    // If we got no response from that operation, this is a failure
+    // likely due to time out
+    if (*event_response == NULL) {
+      return Baton(RdKafka::ERR__TIMED_OUT);
+    }
+
+    // Now we can get the error code from the event
+    if (rd_kafka_event_error(*event_response)) {
+      // If we had a special error code, get out of here with it
+      const rd_kafka_resp_err_t errcode = rd_kafka_event_error(*event_response);
+      return Baton(static_cast<RdKafka::ErrorCode>(errcode));
+    }
+
+    // At this point, event_response contains the result, which needs
+    // to be parsed/converted by the caller.
+    return Baton(RdKafka::ERR_NO_ERROR);
+  }
+}
+
 void AdminClient::ActivateDispatchers() {
   // Listen to global config
   m_gconfig->listen();
@@ -1298,4 +1369,59 @@ NAN_METHOD(AdminClient::NodeDeleteRecords) {
       callback, client, delete_records, 1, operation_timeout_ms, timeout_ms));
 }
 
+/**
+ * Describe Topics.
+ */
+NAN_METHOD(AdminClient::NodeDescribeTopics) {
+  Nan::HandleScope scope;
+
+  if (info.Length() < 3 || !info[2]->IsFunction()) {
+    return Nan::ThrowError("Need to specify a callback");
+  }
+
+  if (!info[0]->IsArray()) {
+    return Nan::ThrowError("Must provide an array of 'topicNames'");
+  }
+
+  v8::Local<v8::Array> topicNames = info[0].As<v8::Array>();
+
+  if (topicNames->Length() == 0) {
+    return Nan::ThrowError("'topicNames' cannot be empty");
+  }
+
+  std::vector<std::string> topicNamesVector = v8ArrayToStringVector(topicNames);
+
+  const char **topics = static_cast<const char **>(
+      malloc(sizeof(const char *) * topicNamesVector.size()));
+
+  for (size_t i = 0; i < topicNamesVector.size(); i++) {
+    topics[i] = topicNamesVector[i].c_str();
+  }
+
+  /**
+   * The ownership of this is taken by
+   * Workers::AdminClientDescribeTopics and freeing it is also handled
+   * by that class.
+   */
+  rd_kafka_TopicCollection_t *topic_collection =
+      rd_kafka_TopicCollection_of_topic_names(topics, topicNamesVector.size());
+
+  free(topics);
+
+  v8::Local<v8::Object> options = info[1].As<v8::Object>();
+
+  bool include_authorised_operations =
+      GetParameter<bool>(options, "includeAuthorizedOperations", false);
+
+  int timeout_ms = GetParameter<int64_t>(options, "timeout", 5000);
+
+  v8::Local<v8::Function> cb = info[2].As<v8::Function>();
+  Nan::Callback *callback = new Nan::Callback(cb);
+  AdminClient *client = ObjectWrap::Unwrap<AdminClient>(info.This());
+
+  Nan::AsyncQueueWorker(new Workers::AdminClientDescribeTopics(
+      callback, client, topic_collection,
+      include_authorised_operations, timeout_ms));
+}
+
 }  // namespace NodeKafka

src/admin.h

@@ -68,6 +68,9 @@ class AdminClient : public Connection {
   Baton DeleteRecords(rd_kafka_DeleteRecords_t** del_records,
                       size_t del_records_cnt, int operation_timeout_ms,
                       int timeout_ms, rd_kafka_event_t** event_response);
+  Baton DescribeTopics(rd_kafka_TopicCollection_t* topics,
+                       bool include_authorized_operations, int timeout_ms,
+                       rd_kafka_event_t** event_response);
 
  protected:
   static Nan::Persistent<v8::Function> constructor;
@@ -91,6 +94,7 @@ class AdminClient : public Connection {
   static NAN_METHOD(NodeDeleteGroups);
   static NAN_METHOD(NodeListConsumerGroupOffsets);
   static NAN_METHOD(NodeDeleteRecords);
+  static NAN_METHOD(NodeDescribeTopics);
 
   static NAN_METHOD(NodeConnect);
   static NAN_METHOD(NodeDisconnect);