Add AdminAPI for describeGroups

Author: milindl

index.d.ts

@@ -72,6 +72,7 @@ export interface WatermarkOffsets{
 export interface TopicPartition {
     topic: string;
     partition: number;
+    error?: LibrdKafkaError;
 }
 
 export interface TopicPartitionOffset extends TopicPartition{
@@ -349,6 +350,56 @@ export interface GroupOverview {
     state: ConsumerGroupStates;
 }
 
+export enum AclOperationTypes {
+    UNKNOWN = 0,
+    ANY = 1,
+    ALL = 2,
+    READ = 3,
+    WRITE = 4,
+    CREATE = 5,
+    DELETE = 6,
+    ALTER = 7,
+    DESCRIBE = 8,
+    CLUSTER_ACTION = 9,
+    DESCRIBE_CONFIGS = 10,
+    ALTER_CONFIGS = 11,
+    IDEMPOTENT_WRITE = 12,
+}
+
+export type MemberDescription = {
+    clientHost: string
+    clientId: string
+    memberId: string
+    memberAssignment: Buffer
+    memberMetadata: Buffer
+    groupInstanceId?: string,
+    assignment: TopicPartition[]
+}
+
+export type Node = {
+    id: number
+    host: string
+    port: number
+    rack?: string
+}
+
+export type GroupDescription = {
+    groupId: string
+    error?: LibrdKafkaError
+    members: MemberDescription[]
+    protocol: string
+    isSimpleConsumerGroup: boolean;
+    protocolType: string
+    partitionAssignor: string
+    state: ConsumerGroupStates
+    coordinator: Node
+    authorizedOperations?: AclOperationTypes[]
+}
+
+export type GroupDescriptions = {
+    groups: GroupDescription[],
+}
+
 export interface IAdminClient {
     createTopic(topic: NewTopic, cb?: (err: LibrdKafkaError) => void): void;
     createTopic(topic: NewTopic, timeout?: number, cb?: (err: LibrdKafkaError) => void): void;
@@ -360,9 +411,14 @@ export interface IAdminClient {
     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[] },
+    listGroups(options?: { timeout?: number, matchConsumerGroupStates?: ConsumerGroupStates[] },
         cb?: (err: LibrdKafkaError, result: { groups: GroupOverview[], errors: LibrdKafkaError[] }) => any): void;
 
+    describeGroups(groupIds: string[], cb?: (err: LibrdKafkaError, result: GroupDescriptions) => any): void;
+    describeGroups(groupIds: string[],
+        options?: { timeout?: number, includeAuthorizedOperations?: boolean },
+        cb?: (err: LibrdKafkaError, result: GroupDescriptions) => any): void;
+
     disconnect(): void;
 }
 

lib/admin.js

@@ -20,12 +20,28 @@ const ConsumerGroupStates = Object.seal({
   STABLE: 3,
   DEAD: 4,
   EMPTY: 5,
-})
+});
 
+const AclOperationTypes = Object.seal({
+  UNKNOWN: 0,
+  ANY: 1,
+  ALL: 2,
+  READ: 3,
+  WRITE: 4,
+  CREATE: 5,
+  DELETE: 6,
+  ALTER: 7,
+  DESCRIBE: 8,
+  CLUSTER_ACTION: 9,
+  DESCRIBE_CONFIGS: 10,
+  ALTER_CONFIGS: 11,
+  IDEMPOTENT_WRITE: 12,
+});
 
 module.exports = {
   create: createAdminClient,
   ConsumerGroupStates,
+  AclOperationTypes,
 };
 
 var Client = require('./client');
@@ -267,4 +283,45 @@ AdminClient.prototype.listGroups = function (options, cb) {
       cb(null, groups);
     }
   });
-}
+}
+
+/**
+ * Describe consumer groups.
+ * @param {string[]} groups - The names of the groups to describe.
+ * @param {any?} options
+ * @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 group by the calling client (default: false).
+ * @param {function} cb - The callback to be executed when finished.
+ *
+ * Valid ways to call this function:
+ * describeGroups(groups, cb)
+ * describeGroups(groups, options, cb)
+ */
+AdminClient.prototype.describeGroups = function (groups, options, cb) {
+  if (!this._isConnected) {
+    throw new Error('Client is disconnected');
+  }
+
+  if (typeof options === 'function') {
+    cb = options;
+    options = {};
+  }
+
+  if (!options) {
+    options = {};
+  }
+
+  this._client.describeGroups(groups, options, function (err, descriptions) {
+    if (err) {
+      if (cb) {
+        cb(LibrdKafkaError.create(err));
+      }
+      return;
+    }
+
+    if (cb) {
+      cb(null, descriptions);
+    }
+  });
+}

lib/kafkajs/_admin.js

@@ -106,7 +106,7 @@ class Admin {
 
     this.#state = AdminState.CONNECTING;
 
-     const config = this.#config();
+    const config = this.#config();
 
     return new Promise((resolve, reject) => {
       try {
@@ -258,6 +258,35 @@ class Admin {
     });
   }
 
+  /**
+   * Describe consumer groups.
+   *
+   * @param {string[]} groups - The names of the groups to describe.
+   * @param {object?} options
+   * @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 group by the calling client (default: false).
+   * @returns {Promise<import("../../types/kafkajs").GroupDescriptions>}
+   */
+  async describeGroups(groups, 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.describeGroups(groups, options, (err, descriptions) => {
+        if (err) {
+          reject(createKafkaJsErrorFromLibRdKafkaError(err));
+        } else {
+          resolve(descriptions);
+        }
+      });
+    });
+  }
 }
 
-module.exports = { Admin, ConsumerGroupStates: RdKafka.AdminClient.ConsumerGroupStates }
+module.exports = {
+  Admin,
+  ConsumerGroupStates: RdKafka.AdminClient.ConsumerGroupStates,
+  AclOperationTypes: RdKafka.AdminClient.AclOperationTypes
+}

lib/kafkajs/_kafka.js

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

src/admin.cc

@@ -89,6 +89,7 @@ void AdminClient::Init(v8::Local<v8::Object> exports) {
 
   // Consumer group related operations
   Nan::SetPrototypeMethod(tpl, "listGroups", NodeListGroups);
+  Nan::SetPrototypeMethod(tpl, "describeGroups", NodeDescribeGroups);
 
   Nan::SetPrototypeMethod(tpl, "connect", NodeConnect);
   Nan::SetPrototypeMethod(tpl, "disconnect", NodeDisconnect);
@@ -455,18 +456,88 @@ Baton AdminClient::ListGroups(
     }
 
     // Create queue just for this operation.
-    rd_kafka_queue_t *topic_rkqu = rd_kafka_queue_new(m_client->c_ptr());
+    rd_kafka_queue_t *rkqu = rd_kafka_queue_new(m_client->c_ptr());
 
-    rd_kafka_ListConsumerGroups(m_client->c_ptr(), options, topic_rkqu);
+    rd_kafka_ListConsumerGroups(m_client->c_ptr(), 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(
-        topic_rkqu, RD_KAFKA_EVENT_LISTCONSUMERGROUPS_RESULT, timeout_ms);
+        rkqu, RD_KAFKA_EVENT_LISTCONSUMERGROUPS_RESULT, timeout_ms);
 
     // Destroy the queue since we are done with it.
-    rd_kafka_queue_destroy(topic_rkqu);
+    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);
+  }
+}
+
+Baton AdminClient::DescribeGroups(std::vector<std::string> &groups,
+                                  bool include_authorized_operations,
+                                  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 describing groups
+    rd_kafka_AdminOptions_t *options = rd_kafka_AdminOptions_new(
+        m_client->c_ptr(), RD_KAFKA_ADMIN_OP_DESCRIBECONSUMERGROUPS);
+
+    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);
+      }
+    }
+
+    // Create queue just for this operation.
+    rd_kafka_queue_t *rkqu = rd_kafka_queue_new(m_client->c_ptr());
+
+    // Construct a char** to pass to librdkafka. Avoid too many allocations.
+    std::vector<const char *> c_groups(groups.size());
+    for (size_t i = 0; i < groups.size(); i++) {
+      c_groups[i] = groups[i].c_str();
+    }
+
+    rd_kafka_DescribeConsumerGroups(m_client->c_ptr(), &c_groups[0],
+                                    groups.size(), 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_DESCRIBECONSUMERGROUPS_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);
@@ -696,10 +767,10 @@ NAN_METHOD(AdminClient::NodeListGroups) {
 
   // 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 =
+  v8::Local<v8::String> match_consumer_group_states_key =
       Nan::New("matchConsumerGroupStates").ToLocalChecked();
   bool is_match_states_set =
-      Nan::Has(config, matchConsumerGroupStatesKey).FromMaybe(false);
+      Nan::Has(config, match_consumer_group_states_key).FromMaybe(false);
   v8::Local<v8::Array> match_states_array;
 
   if (is_match_states_set) {
@@ -713,4 +784,51 @@ NAN_METHOD(AdminClient::NodeListGroups) {
       callback, client, is_match_states_set, match_states, timeout_ms));
 }
 
+/**
+ * Describe Consumer Groups.
+ */
+NAN_METHOD(AdminClient::NodeDescribeGroups) {
+  Nan::HandleScope scope;
+
+  if (info.Length() < 3 || !info[2]->IsFunction()) {
+    // Just throw an exception
+    return Nan::ThrowError("Need to specify a callback");
+  }
+
+  if (!info[0]->IsArray()) {
+    return Nan::ThrowError("Must provide group name array");
+  }
+
+  if (!info[1]->IsObject()) {
+    return Nan::ThrowError("Must provide options object");
+  }
+
+  // Get list of group names to describe.
+  v8::Local<v8::Array> group_names = info[0].As<v8::Array>();
+  if (group_names->Length() == 0) {
+    return Nan::ThrowError("Must provide at least one group name");
+  }
+  std::vector<std::string> group_names_vector =
+      v8ArrayToStringVector(group_names);
+
+  v8::Local<v8::Object> config = info[1].As<v8::Object>();
+
+  // Get the timeout - default 5000.
+  int timeout_ms = GetParameter<int64_t>(config, "timeout", 5000);
+
+  // Get whether to include authorized operations - default false.
+  bool include_authorized_operations =
+      GetParameter<bool>(config, "includeAuthorizedOperations", false);
+
+  // Create the final callback object
+  v8::Local<v8::Function> cb = info[2].As<v8::Function>();
+  Nan::Callback *callback = new Nan::Callback(cb);
+  AdminClient *client = ObjectWrap::Unwrap<AdminClient>(info.This());
+
+  // Queue the work.
+  Nan::AsyncQueueWorker(new Workers::AdminClientDescribeGroups(
+      callback, client, group_names_vector, include_authorized_operations,
+      timeout_ms));
+}
+
 }  // namespace NodeKafka