Add ListGroups to admin client

Author: milindl

index.d.ts

@@ -333,6 +333,22 @@ export interface NewTopic {
     } | { [cfg: string]: string; };
 }
 
+export enum ConsumerGroupStates {
+    UNKNOWN = 0,
+    PREPARING_REBALANCE = 1,
+    COMPLETING_REBALANCE = 2,
+    STABLE = 3,
+    DEAD = 4,
+    EMPTY = 5,
+}
+
+export interface GroupOverview {
+    groupId: string;
+    protocolType: string;
+    isSimpleConsumerGroup: boolean;
+    state: ConsumerGroupStates;
+}
+
 export interface IAdminClient {
     createTopic(topic: NewTopic, cb?: (err: LibrdKafkaError) => void): void;
     createTopic(topic: NewTopic, timeout?: number, cb?: (err: LibrdKafkaError) => void): void;
@@ -343,6 +359,10 @@ 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;
 
+    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;
+
     disconnect(): void;
 }
 

lib/admin.js

@@ -2,14 +2,30 @@
  * confluent-kafka-javascript - Node.js wrapper  for RdKafka C/C++ library
  *
  * Copyright (c) 2016-2023 Blizzard Entertainment
+ *               2024 Confluent, Inc
  *
  * This software may be modified and distributed under the terms
  * of the MIT license.  See the LICENSE.txt file for details.
  */
 'use strict';
 
+/* TODO: Think of a way to fetch these from within librdkafka instead of this
+ * hardcoded list.
+ * New additions won't be automatically added to this list.
+ */
+const ConsumerGroupStates = Object.seal({
+  UNKNOWN: 0,
+  PREPARING_REBALANCE: 1,
+  COMPLETING_REBALANCE: 2,
+  STABLE: 3,
+  DEAD: 4,
+  EMPTY: 5,
+})
+
+
 module.exports = {
   create: createAdminClient,
+  ConsumerGroupStates,
 };
 
 var Client = require('./client');
@@ -89,7 +105,7 @@ function AdminClient(conf) {
  *
  * Unlike the other connect methods, this one is synchronous.
  */
-AdminClient.prototype.connect = function() {
+AdminClient.prototype.connect = function () {
   LibrdKafkaError.wrap(this._client.connect(), true);
   this._isConnected = true;
 };
@@ -100,7 +116,7 @@ AdminClient.prototype.connect = function() {
  * This is a synchronous method, but all it does is clean up
  * some memory and shut some threads down
  */
-AdminClient.prototype.disconnect = function() {
+AdminClient.prototype.disconnect = function () {
   LibrdKafkaError.wrap(this._client.disconnect(), true);
   this._isConnected = false;
 };
@@ -112,7 +128,7 @@ AdminClient.prototype.disconnect = function() {
  * @param {number} timeout - Number of milliseconds to wait while trying to create the topic.
  * @param {function} cb - The callback to be executed when finished
  */
-AdminClient.prototype.createTopic = function(topic, timeout, cb) {
+AdminClient.prototype.createTopic = function (topic, timeout, cb) {
   if (!this._isConnected) {
     throw new Error('Client is disconnected');
   }
@@ -126,7 +142,7 @@ AdminClient.prototype.createTopic = function(topic, timeout, cb) {
     timeout = 5000;
   }
 
-  this._client.createTopic(topic, timeout, function(err) {
+  this._client.createTopic(topic, timeout, function (err) {
     if (err) {
       if (cb) {
         cb(LibrdKafkaError.create(err));
@@ -147,7 +163,7 @@ AdminClient.prototype.createTopic = function(topic, timeout, cb) {
  * @param {number} timeout - Number of milliseconds to wait while trying to delete the topic.
  * @param {function} cb - The callback to be executed when finished
  */
-AdminClient.prototype.deleteTopic = function(topic, timeout, cb) {
+AdminClient.prototype.deleteTopic = function (topic, timeout, cb) {
   if (!this._isConnected) {
     throw new Error('Client is disconnected');
   }
@@ -161,7 +177,7 @@ AdminClient.prototype.deleteTopic = function(topic, timeout, cb) {
     timeout = 5000;
   }
 
-  this._client.deleteTopic(topic, timeout, function(err) {
+  this._client.deleteTopic(topic, timeout, function (err) {
     if (err) {
       if (cb) {
         cb(LibrdKafkaError.create(err));
@@ -184,7 +200,7 @@ AdminClient.prototype.deleteTopic = function(topic, timeout, cb) {
  * @param {number} timeout - Number of milliseconds to wait while trying to create the partitions.
  * @param {function} cb - The callback to be executed when finished
  */
-AdminClient.prototype.createPartitions = function(topic, totalPartitions, timeout, cb) {
+AdminClient.prototype.createPartitions = function (topic, totalPartitions, timeout, cb) {
   if (!this._isConnected) {
     throw new Error('Client is disconnected');
   }
@@ -198,7 +214,7 @@ AdminClient.prototype.createPartitions = function(topic, totalPartitions, timeou
     timeout = 5000;
   }
 
-  this._client.createPartitions(topic, totalPartitions, timeout, function(err) {
+  this._client.createPartitions(topic, totalPartitions, timeout, function (err) {
     if (err) {
       if (cb) {
         cb(LibrdKafkaError.create(err));
@@ -211,3 +227,44 @@ AdminClient.prototype.createPartitions = function(topic, totalPartitions, timeou
     }
   });
 };
+
+/**
+ * List consumer groups.
+ * @param {any} options
+ * @param {number?} options.timeout - The request timeout in milliseconds.
+ *                                    May be unset (default: 5000)
+ * @param {import("../").ConsumerGroupStates[]?} options.matchConsumerGroupStates -
+ *        A list of consumer group states to match. May be unset, fetches all states (default: unset).
+ * @param {function} cb - The callback to be executed when finished.
+ *
+ * Valid ways to call this function:
+ * listGroups(cb)
+ * listGroups(options, cb)
+ */
+AdminClient.prototype.listGroups = function (options, cb) {
+  if (!this._isConnected) {
+    throw new Error('Client is disconnected');
+  }
+
+  if (typeof options === 'function') {
+    cb = options;
+    options = {};
+  }
+
+  if (!options) {
+    options = {};
+  }
+
+  this._client.listGroups(options, function (err, groups) {
+    if (err) {
+      if (cb) {
+        cb(LibrdKafkaError.create(err));
+      }
+      return;
+    }
+
+    if (cb) {
+      cb(null, groups);
+    }
+  });
+}

lib/kafkajs/_admin.js

@@ -231,6 +231,33 @@ class Admin {
     );
   }
 
+  /**
+   * List consumer groups.
+   *
+   * @param {object?} options
+   * @param {number?} options.timeout - The request timeout in milliseconds.
+   *                                    May be unset (default: 5000)
+   * @param {import("../../types/kafkajs").ConsumerGroupStates[]?} options.matchConsumerGroupStates -
+   *        A list of consumer group states to match. May be unset, fetches all states (default: unset).
+   * @returns {Promise<{ groups: import("../../types/kafkajs").GroupOverview[], errors: import("../../types/kafkajs").LibrdKafkaError[] }>}
+   *          Resolves with the list of consumer groups, rejects on error.
+   */
+  async listGroups(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.listGroups(options, (err, groups) => {
+        if (err) {
+          reject(createKafkaJsErrorFromLibRdKafkaError(err));
+        } else {
+          resolve(groups);
+        }
+      });
+    });
+  }
+
 }
 
-module.exports = { Admin }
+module.exports = { Admin, ConsumerGroupStates: RdKafka.AdminClient.ConsumerGroupStates }

lib/kafkajs/_kafka.js

@@ -1,6 +1,6 @@
 const { Producer, CompressionTypes } = require('./_producer');
 const { Consumer, PartitionAssigners } = require('./_consumer');
-const { Admin } = require('./_admin');
+const { Admin, ConsumerGroupStates } = require('./_admin');
 const error = require('./_error');
 const { logLevel, checkIfKafkaJsKeysPresent, CompatibilityErrorMessages } = require('./_common');
 
@@ -80,4 +80,10 @@ class Kafka {
   }
 }
 
-module.exports = { Kafka, ...error, logLevel, PartitionAssigners, PartitionAssignors: PartitionAssigners, CompressionTypes };
+module.exports = {
+  Kafka,
+  ...error, logLevel,
+  PartitionAssigners,
+  PartitionAssignors: PartitionAssigners,
+  CompressionTypes,
+  ConsumerGroupStates };

src/admin.cc

@@ -87,6 +87,9 @@ void AdminClient::Init(v8::Local<v8::Object> exports) {
   Nan::SetPrototypeMethod(tpl, "deleteTopic", NodeDeleteTopic);
   Nan::SetPrototypeMethod(tpl, "createPartitions", NodeCreatePartitions);
 
+  // Consumer group related operations
+  Nan::SetPrototypeMethod(tpl, "listGroups", NodeListGroups);
+
   Nan::SetPrototypeMethod(tpl, "connect", NodeConnect);
   Nan::SetPrototypeMethod(tpl, "disconnect", NodeDisconnect);
   Nan::SetPrototypeMethod(tpl, "setSaslCredentials", NodeSetSaslCredentials);
@@ -179,6 +182,9 @@ rd_kafka_event_t* PollForEvent(
     rd_kafka_event_type(event_response) != event_type &&
     attempts > 0);
 
+  // TODO: change this function so a type mismatch leads to an INVALID_TYPE
+  // error rather than a null event. A null event is treated as a timeout, which
+  // isn't true all the time.
   // If this isn't the type of response we want, or if we do not have a response
   // type, bail out with a null
   if (event_response == NULL ||
@@ -421,6 +427,69 @@ Baton AdminClient::CreatePartitions(
   }
 }
 
+Baton AdminClient::ListGroups(
+    bool is_match_states_set,
+    std::vector<rd_kafka_consumer_group_state_t> &match_states, int timeout_ms,
+    /* out */ 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 listing groups
+    rd_kafka_AdminOptions_t *options = rd_kafka_AdminOptions_new(
+        m_client->c_ptr(), RD_KAFKA_ADMIN_OP_LISTCONSUMERGROUPS);
+
+    if (is_match_states_set) {
+      rd_kafka_error_t *error =
+          rd_kafka_AdminOptions_set_match_consumer_group_states(
+              options, &match_states[0], match_states.size());
+      if (error) {
+        return Baton::BatonFromErrorAndDestroy(error);
+      }
+    }
+
+    // Create queue just for this operation.
+    rd_kafka_queue_t *topic_rkqu = rd_kafka_queue_new(m_client->c_ptr());
+
+    rd_kafka_ListConsumerGroups(m_client->c_ptr(), options, topic_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(
+        topic_rkqu, RD_KAFKA_EVENT_LISTCONSUMERGROUPS_RESULT, timeout_ms);
+
+    // Destroy the queue since we are done with it.
+    rd_kafka_queue_destroy(topic_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();
@@ -600,4 +669,48 @@ NAN_METHOD(AdminClient::NodeCreatePartitions) {
   return info.GetReturnValue().Set(Nan::Null());
 }
 
+/**
+ * List Consumer Groups.
+ */
+NAN_METHOD(AdminClient::NodeListGroups) {
+  Nan::HandleScope scope;
+
+  if (info.Length() < 2 || !info[1]->IsFunction()) {
+    // Just throw an exception
+    return Nan::ThrowError("Need to specify a callback");
+  }
+
+  if (!info[0]->IsObject()) {
+    return Nan::ThrowError("Must provide options object");
+  }
+
+  v8::Local<v8::Object> config = info[0].As<v8::Object>();
+
+  // Create the final callback object
+  v8::Local<v8::Function> cb = info[1].As<v8::Function>();
+  Nan::Callback *callback = new Nan::Callback(cb);
+  AdminClient *client = ObjectWrap::Unwrap<AdminClient>(info.This());
+
+  // Get the timeout - default 5000.
+  int timeout_ms = GetParameter<int64_t>(config, "timeout", 5000);
+
+  // Get the match states, or not if they are unset.
+  std::vector<rd_kafka_consumer_group_state_t> match_states;
+  v8::Local<v8::String> matchConsumerGroupStatesKey =
+      Nan::New("matchConsumerGroupStates").ToLocalChecked();
+  bool is_match_states_set =
+      Nan::Has(config, matchConsumerGroupStatesKey).FromMaybe(false);
+  v8::Local<v8::Array> match_states_array;
+
+  if (is_match_states_set) {
+    match_states_array = GetParameter<v8::Local<v8::Array>>(
+        config, "matchConsumerGroupStates", match_states_array);
+    match_states = Conversion::Admin::FromV8GroupStateArray(match_states_array);
+  }
+
+  // Queue the work.
+  Nan::AsyncQueueWorker(new Workers::AdminClientListGroups(
+      callback, client, is_match_states_set, match_states, timeout_ms));
+}
+
 }  // namespace NodeKafka

src/admin.h

@@ -51,6 +51,10 @@ class AdminClient : public Connection {
   Baton CreatePartitions(rd_kafka_NewPartitions_t* topic, int timeout_ms);
   // Baton AlterConfig(rd_kafka_NewTopic_t* topic, int timeout_ms);
   // Baton DescribeConfig(rd_kafka_NewTopic_t* topic, int timeout_ms);
+  Baton ListGroups(bool is_match_states_set,
+                   std::vector<rd_kafka_consumer_group_state_t>& match_states,
+                   int timeout_ms,
+                   rd_kafka_event_t** event_response);
 
  protected:
   static Nan::Persistent<v8::Function> constructor;
@@ -68,6 +72,9 @@ class AdminClient : public Connection {
   static NAN_METHOD(NodeDeleteTopic);
   static NAN_METHOD(NodeCreatePartitions);
 
+  // Consumer group operations
+  static NAN_METHOD(NodeListGroups);
+
   static NAN_METHOD(NodeConnect);
   static NAN_METHOD(NodeDisconnect);
 };