Add AdminAPI for deleteGroups

Author: milindl

index.d.ts

@@ -400,6 +400,12 @@ export type GroupDescriptions = {
     groups: GroupDescription[],
 }
 
+export type DeleteGroupsResult = {
+    groupId: string
+    errorCode?: number
+    error?: LibrdKafkaError
+}
+
 export interface IAdminClient {
     createTopic(topic: NewTopic, cb?: (err: LibrdKafkaError) => void): void;
     createTopic(topic: NewTopic, timeout?: number, cb?: (err: LibrdKafkaError) => void): void;
@@ -419,6 +425,11 @@ export interface IAdminClient {
         options?: { timeout?: number, includeAuthorizedOperations?: boolean },
         cb?: (err: LibrdKafkaError, result: GroupDescriptions) => any): void;
 
+    deleteGroups(groupIds: string[], cb?: (err: LibrdKafkaError, result: DeleteGroupsResult[]) => any): void;
+    deleteGroups(groupIds: string[],
+        options?: { timeout?: number },
+        cb?: (err: LibrdKafkaError, result: DeleteGroupsResult[]) => any): void;
+
     disconnect(): void;
 }
 

lib/admin.js

@@ -325,3 +325,43 @@ AdminClient.prototype.describeGroups = function (groups, options, cb) {
     }
   });
 }
+
+/**
+ * Delete consumer groups.
+ * @param {string[]} groups - The names of the groups to delete.
+ * @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:
+ * deleteGroups(groups, cb)
+ * deleteGroups(groups, options, cb)
+ */
+AdminClient.prototype.deleteGroups = 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.deleteGroups(groups, options, function (err, reports) {
+    if (err) {
+      if (cb) {
+        cb(LibrdKafkaError.create(err));
+      }
+      return;
+    }
+
+    if (cb) {
+      cb(null, reports);
+    }
+  });
+}

lib/kafkajs/_admin.js

@@ -283,6 +283,30 @@ class Admin {
       });
     });
   }
+
+  /**
+   * Delete consumer groups.
+   * @param {string[]} groups - The names of the groups to delete.
+   * @param {any?} options
+   * @param {number?} options.timeout - The request timeout in milliseconds.
+   *                                    May be unset (default: 5000)
+   * @returns {Promise<import("../../types/kafkajs").DeleteGroupsResult[]>}
+   */
+  async deleteGroups(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.deleteGroups(groups, options, (err, reports) => {
+        if (err) {
+          reject(createKafkaJsErrorFromLibRdKafkaError(err));
+        } else {
+          resolve(reports);
+        }
+      });
+    });
+  }
 }
 
 module.exports = {

src/admin.cc

@@ -90,6 +90,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, "deleteGroups", NodeDeleteGroups);
 
   Nan::SetPrototypeMethod(tpl, "connect", NodeConnect);
   Nan::SetPrototypeMethod(tpl, "disconnect", NodeDisconnect);
@@ -446,6 +447,13 @@ Baton AdminClient::ListGroups(
     rd_kafka_AdminOptions_t *options = rd_kafka_AdminOptions_new(
         m_client->c_ptr(), RD_KAFKA_ADMIN_OP_LISTCONSUMERGROUPS);
 
+    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);
+    }
+
     if (is_match_states_set) {
       rd_kafka_error_t *error =
           rd_kafka_AdminOptions_set_match_consumer_group_states(
@@ -509,6 +517,13 @@ Baton AdminClient::DescribeGroups(std::vector<std::string> &groups,
     rd_kafka_AdminOptions_t *options = rd_kafka_AdminOptions_new(
         m_client->c_ptr(), RD_KAFKA_ADMIN_OP_DESCRIBECONSUMERGROUPS);
 
+    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);
+    }
+
     if (include_authorized_operations) {
       rd_kafka_error_t *error =
           rd_kafka_AdminOptions_set_include_authorized_operations(
@@ -561,6 +576,67 @@ Baton AdminClient::DescribeGroups(std::vector<std::string> &groups,
   }
 }
 
+Baton AdminClient::DeleteGroups(rd_kafka_DeleteGroup_t **group_list,
+                                size_t group_cnt, 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 deleting groups
+    rd_kafka_AdminOptions_t *options = rd_kafka_AdminOptions_new(
+        m_client->c_ptr(), RD_KAFKA_ADMIN_OP_DELETEGROUPS);
+
+    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_DeleteGroups(m_client->c_ptr(), group_list, group_cnt, 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_DELETEGROUPS_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();
@@ -831,4 +907,54 @@ NAN_METHOD(AdminClient::NodeDescribeGroups) {
       timeout_ms));
 }
 
+/**
+ * Delete Consumer Groups.
+ */
+NAN_METHOD(AdminClient::NodeDeleteGroups) {
+  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 delete, and convert it into an
+  // rd_kafka_DeleteGroup_t array.
+  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);
+
+  // The ownership of this array is transferred to the worker.
+  rd_kafka_DeleteGroup_t **group_list = static_cast<rd_kafka_DeleteGroup_t **>(
+      malloc(sizeof(rd_kafka_DeleteGroup_t *) * group_names_vector.size()));
+  for (size_t i = 0; i < group_names_vector.size(); i++) {
+    group_list[i] = rd_kafka_DeleteGroup_new(group_names_vector[i].c_str());
+  }
+
+  v8::Local<v8::Object> config = info[1].As<v8::Object>();
+
+  // Get the timeout - default 5000.
+  int timeout_ms = GetParameter<int64_t>(config, "timeout", 5000);
+
+  // 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::AdminClientDeleteGroups(
+      callback, client, group_list, group_names_vector.size(), timeout_ms));
+}
+
 }  // namespace NodeKafka

src/admin.h

@@ -59,6 +59,8 @@ class AdminClient : public Connection {
   Baton DescribeGroups(std::vector<std::string>& groups,
                        bool include_authorized_operations, int timeout_ms,
                        rd_kafka_event_t** event_response);
+  Baton DeleteGroups(rd_kafka_DeleteGroup_t** group_list, size_t group_cnt,
+                     int timeout_ms, rd_kafka_event_t** event_response);
 
  protected:
   static Nan::Persistent<v8::Function> constructor;
@@ -79,6 +81,7 @@ class AdminClient : public Connection {
   // Consumer group operations
   static NAN_METHOD(NodeListGroups);
   static NAN_METHOD(NodeDescribeGroups);
+  static NAN_METHOD(NodeDeleteGroups);
 
   static NAN_METHOD(NodeConnect);
   static NAN_METHOD(NodeDisconnect);

src/common.cc

@@ -961,6 +961,51 @@ v8::Local<v8::Object> FromDescribeConsumerGroupsResult(
   return returnObject;
 }
 
+/**
+ * @brief Converts a rd_kafka_DeleteGroups_result_t* into a v8 array.
+*/
+v8::Local<v8::Array> FromDeleteGroupsResult(
+    const rd_kafka_DeleteGroups_result_t* result) {
+  /* Return object type:
+    [{
+      groupId: string
+      errorCode?: number
+      error?: LibrdKafkaError
+    }]
+  */
+  v8::Local<v8::Array> returnArray = Nan::New<v8::Array>();
+  size_t result_cnt;
+  const rd_kafka_group_result_t** results =
+      rd_kafka_DeleteGroups_result_groups(result, &result_cnt);
+
+  for (size_t i = 0; i < result_cnt; i++) {
+    const rd_kafka_group_result_t* group_result = results[i];
+    v8::Local<v8::Object> group_object = Nan::New<v8::Object>();
+
+    Nan::Set(group_object, Nan::New("groupId").ToLocalChecked(),
+             Nan::New<v8::String>(rd_kafka_group_result_name(group_result))
+                 .ToLocalChecked());
+
+    const rd_kafka_error_t* error = rd_kafka_group_result_error(group_result);
+    if (!error) {
+      Nan::Set(group_object, Nan::New("errorCode").ToLocalChecked(),
+               Nan::New<v8::Number>(RD_KAFKA_RESP_ERR_NO_ERROR));
+    } else {
+      RdKafka::ErrorCode code =
+          static_cast<RdKafka::ErrorCode>(rd_kafka_error_code(error));
+      const char* msg = rd_kafka_error_string(error);
+
+      Nan::Set(group_object, Nan::New("errorCode").ToLocalChecked(),
+               Nan::New<v8::Number>(code));
+      Nan::Set(group_object, Nan::New("error").ToLocalChecked(),
+               RdKafkaError(code, msg));
+    }
+    Nan::Set(returnArray, i, group_object);
+  }
+
+  return returnArray;
+}
+
 }  // namespace Admin
 
 }  // namespace Conversion

src/common.h

@@ -124,6 +124,10 @@ v8::Local<v8::Object> FromConsumerGroupDescription(
     const rd_kafka_ConsumerGroupDescription_t *desc);
 v8::Local<v8::Object> FromDescribeConsumerGroupsResult(
     const rd_kafka_DescribeConsumerGroups_result_t *);
+
+// DeleteGroups: Response
+v8::Local<v8::Array> FromDeleteGroupsResult(
+    const rd_kafka_DeleteGroups_result_t *);
 }  // namespace Admin
 
 namespace TopicPartition {