Add AdminAPI for listTopics

Author: milindl

CHANGELOG.md

@@ -1,3 +1,12 @@
+# confluent-kafka-javascript vNext
+
+v0.1.12-devel is a pre-production, early-access release.
+
+## Features
+
+1. Add support for `listTopics` in the Admin API.
+
+
 # confluent-kafka-javascript v0.1.11-devel
 
 v0.1.11-devel is a pre-production, early-access release.

index.d.ts

@@ -416,6 +416,9 @@ export interface IAdminClient {
     createPartitions(topic: string, desiredPartitions: number, cb?: (err: LibrdKafkaError) => void): void;
     createPartitions(topic: string, desiredPartitions: number, timeout?: number, cb?: (err: LibrdKafkaError) => void): void;
 
+    listTopics(cb?: (err: LibrdKafkaError, topics: string[]) => any): void;
+    listTopics(options?: { timeout?: number }, cb?: (err: LibrdKafkaError, topics: string[]) => any): void;
+
     listGroups(cb?: (err: LibrdKafkaError, result: { groups: GroupOverview[], errors: LibrdKafkaError[] }) => any): void;
     listGroups(options?: { timeout?: number, matchConsumerGroupStates?: ConsumerGroupStates[] },
         cb?: (err: LibrdKafkaError, result: { groups: GroupOverview[], errors: LibrdKafkaError[] }) => any): void;

lib/admin.js

@@ -365,3 +365,59 @@ AdminClient.prototype.deleteGroups = function (groups, options, cb) {
     }
   });
 }
+
+/**
+ * List topics.
+ *
+ * @param {any?} options
+ * @param {number?} options.timeout - The request timeout in milliseconds.
+ *                                    May be unset (default: 5000)
+ * @param {function} cb - The callback to be executed when finished.
+ *
+ * Valid ways to call this function:
+ * listTopics(cb)
+ * listTopics(options, cb)
+ */
+AdminClient.prototype.listTopics = function (options, cb) {
+  if (!this._isConnected) {
+    throw new Error('Client is disconnected');
+  }
+
+  if (typeof options === 'function') {
+    cb = options;
+    options = {};
+  }
+
+  if (!options) {
+    options = {};
+  }
+
+  // Always set allTopics to true, since we need a list.
+  options.allTopics = true;
+  if (!Object.hasOwn(options, 'timeout')) {
+    options.timeout = 5000;
+  }
+
+  // This definitely isn't the fastest way to list topics as
+  // this makes a pretty large metadata request. But for the sake
+  // of AdminAPI, this is okay.
+  this._client.getMetadata(options, function (err, metadata) {
+    if (err) {
+      if (cb) {
+        cb(LibrdKafkaError.create(err));
+      }
+      return;
+    }
+
+    const topics = []
+    if (metadata.topics) {
+      for (const topic of metadata.topics) {
+        topics.push(topic.name);
+      }
+    }
+
+    if (cb) {
+      cb(null, topics);
+    }
+  });
+}

lib/kafkajs/_admin.js

@@ -307,6 +307,30 @@ class Admin {
       });
     });
   }
+
+  /**
+   * List topics.
+   *
+   * @param {any?} options
+   * @param {number?} options.timeout - The request timeout in milliseconds.
+   *                                    May be unset (default: 5000)
+   * @returns {Promise<string[]>}
+   */
+  async listTopics(options = {}) {
+    if (this.#state !== AdminState.CONNECTED) {
+      throw new error.KafkaJSError("Admin client is not connected.", { code: error.ErrorCodes.ERR__STATE });
+    }
+
+    return new Promise((resolve, reject) => {
+      this.#internalClient.listTopics(options, (err, topics) => {
+        if (err) {
+          reject(createKafkaJsErrorFromLibRdKafkaError(err));
+        } else {
+          resolve(topics);
+        }
+      });
+    });
+  }
 }
 
 module.exports = {

src/admin.cc

@@ -95,6 +95,7 @@ void AdminClient::Init(v8::Local<v8::Object> exports) {
   Nan::SetPrototypeMethod(tpl, "connect", NodeConnect);
   Nan::SetPrototypeMethod(tpl, "disconnect", NodeDisconnect);
   Nan::SetPrototypeMethod(tpl, "setSaslCredentials", NodeSetSaslCredentials);
+  Nan::SetPrototypeMethod(tpl, "getMetadata", NodeGetMetadata);
 
   constructor.Reset(
     (tpl->GetFunction(Nan::GetCurrentContext())).ToLocalChecked());

test/promisified/admin/list_topics.spec.js

@@ -0,0 +1,44 @@
+jest.setTimeout(30000);
+
+const {
+    secureRandom,
+    createTopic,
+    waitFor,
+    createAdmin,
+} = require('../testhelpers');
+const { ConsumerGroupStates, ErrorCodes } = require('../../../lib').KafkaJS;
+
+describe('Admin > listTopics', () => {
+    let topicNames, admin;
+
+    beforeEach(async () => {
+        topicNames = [`test-topic-${secureRandom()}`, `test-topic-${secureRandom()}`];
+
+        await createTopic({ topic: topicNames[0], });
+        await createTopic({ topic: topicNames[1] });
+
+        admin = createAdmin({});
+    });
+
+    afterEach(async () => {
+        admin && (await admin.disconnect());
+    });
+
+    it('should timeout', async () => {
+        await admin.connect();
+
+        await expect(admin.listTopics({ timeout: 1 })).rejects.toHaveProperty(
+            'code',
+            ErrorCodes.ERR__TIMED_OUT
+        );
+    });
+
+    it('should list consumer topics', async () => {
+        await admin.connect();
+        const listTopicsResult = await admin.listTopics();
+        expect(listTopicsResult).toEqual(
+            expect.arrayContaining(topicNames)
+        );
+    });
+});
+

types/kafkajs.d.ts

@@ -541,6 +541,7 @@ export type Admin = {
     topics: ITopicConfig[]
   }): Promise<boolean>
   deleteTopics(options: { topics: string[]; timeout?: number }): Promise<void>
+  listTopics(options?: { timeout?: number }): Promise<string[]>
   listGroups(options?: {
     timeout?: number,
     matchConsumerGroupStates?: ConsumerGroupStates[]