Delete Records Api implemented (#141)

* delete records api implemented

* requested changes

* changelog change

* minor changes

* minor changes

* minor changes

Author: PratRanj07

CHANGELOG.md

@@ -1,3 +1,12 @@
+# confluent-kafka-javascript v0.5.0
+
+v0.5.0 is a limited availability feature release. It is supported for all usage.
+
+## Enhancements
+
+1. Add support for an Admin API to delete records.(#141).
+
+
 # confluent-kafka-javascript v0.4.0
 
 v0.4.0 is a limited availability feature release. It is supported for all usage.

MIGRATION.md

@@ -333,6 +333,8 @@ The admin-client only has support for a limited subset of methods, with more to
   * The `deleteGroups` method is supported with an additional `timeout` option.
   * The `fetchOffsets` method is supported with additional `timeout` and 
   `requireStableOffsets` option but `resolveOffsets` option is not yet supported.
+  * The `deleteTopicRecords` method is supported with additional `timeout`
+  and `operationTimeout` option.
 
 ### Using the Schema Registry
 

examples/kafkajs/admin/delete-topic-records.js

@@ -0,0 +1,83 @@
+const { Kafka } = require('@confluentinc/kafka-javascript').KafkaJS;
+const { parseArgs } = require('node:util');
+
+async function deleteTopicRecords() {
+    const args = parseArgs({
+        allowPositionals: true,
+        options: {
+            'bootstrap-servers': {
+                type: 'string',
+                short: 'b',
+                default: 'localhost:9092',
+            },
+            'timeout': {
+                type: 'string',
+                short: 't',
+                default: '5000',
+            },
+            'operation-timeout': {
+                type: 'string',
+                short: 'o',
+                default: '60000',
+            },
+        },
+    });
+
+    const {
+        'bootstrap-servers': bootstrapServers,
+        timeout,
+        'operation-timeout': operationTimeout,
+    } = args.values;
+
+    const [topic, ...rest] = args.positionals;
+
+    if (!topic || rest.length % 2 !== 0) {
+        console.error("Usage: node deleteTopicRecords.js --bootstrap-servers <servers> --timeout <timeout> --operation-timeout <operation-timeout> <topic> <partition offset ...>");
+        process.exit(1);
+    }
+
+    const kafka = new Kafka({
+        kafkaJS: {
+            brokers: [bootstrapServers],
+        },
+    });
+
+    const admin = kafka.admin();
+    await admin.connect();
+
+    try {
+        // Parse partitions and offsets, ensuring pairs of partition and offset are provided
+        const partitionsInput = parsePartitionsAndOffsets(rest);
+
+        // Delete records for the specified topic and partitions
+        const result = await admin.deleteTopicRecords({
+            topic: topic,
+            partitions: partitionsInput,
+            timeout: Number(timeout),
+            operationTimeout: Number(operationTimeout),
+        });
+
+        console.log(`Records deleted for Topic "${topic}":`, JSON.stringify(result, null, 2));
+    } catch (err) {
+        console.error("Error deleting topic records:", err);
+    } finally {
+        await admin.disconnect();
+    }
+}
+
+// Helper function to parse partitions and offsets from arguments
+function parsePartitionsAndOffsets(args) {
+    const partitions = [];
+    for (let i = 0; i < args.length; i += 2) {
+        const partition = parseInt(args[i]);
+        const offset = args[i + 1];
+        if (isNaN(partition) || isNaN(parseInt(offset))) {
+            console.error("Partition and offset should be numbers and provided in pairs.");
+            process.exit(1);
+          }
+        partitions.push({ partition, offset });
+    }
+    return partitions;
+}
+
+deleteTopicRecords();

lib/admin.js

@@ -461,6 +461,9 @@ AdminClient.prototype.listConsumerGroupOffsets = function (listGroupOffsets, opt
     throw new Error('groupId must be provided');
   }
 
+  if(!options) {
+    options = {};
+  }
 
   if (!Object.hasOwn(options, 'timeout')) {
     options.timeout = 5000;
@@ -484,3 +487,50 @@ AdminClient.prototype.listConsumerGroupOffsets = function (listGroupOffsets, opt
   });
 };
 
+/**
+ * Deletes records (messages) in topic partitions older than the offsets provided.
+ * Provide Topic.OFFSET_END or -1 as offset to delete all records.
+ *
+ * @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.
+ *                                             May be unset (default: 60000)
+ * @param {number?} options.timeout - The request timeout in milliseconds.
+ *                                    May be unset (default: 5000)
+ *
+ * @param {function} cb - The callback to be executed when finished.
+ */
+AdminClient.prototype.deleteRecords = function (delRecords, 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, 'operationTimeout')) {
+    options.operationTimeout = 60000;
+  }
+
+  if (!Array.isArray(delRecords) || delRecords.length === 0) {
+    throw new Error('delRecords must be a non-empty array');
+  }
+
+  this._client.deleteRecords(delRecords, options, function (err, results) {
+    if (err) {
+      if (cb) {
+        cb(LibrdKafkaError.create(err));
+      }
+      return;
+    }
+
+    if (cb) {
+      cb(null, results);
+    }
+  });
+};

lib/kafkajs/_admin.js

@@ -534,6 +534,61 @@ class Admin {
       });
     });
   }
+
+  /**
+   * Deletes records (messages) in topic partitions older than the offsets provided.
+   * Provide -1 as offset to delete all records.
+   *
+   * @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.
+   *                                             May be unset (default: 60000)
+   * @param {number?} options.timeout - The request timeout in milliseconds.
+   *                                    May be unset (default: 5000)
+   *
+   * @returns {Promise<import('../../types/kafkajs').DeleteRecordsResult[]>>}
+   */
+  async deleteTopicRecords(options = {}) {
+    if (this.#state !== AdminState.CONNECTED) {
+      throw new error.KafkaJSError("Admin client is not connected.", { code: error.ErrorCodes.ERR__STATE });
+    }
+
+    if (!Object.hasOwn(options, 'topic') || !Object.hasOwn(options, 'partitions') || !Array.isArray(options.partitions)) {
+      throw new error.KafkaJSError("Options must include 'topic' and 'partitions', and 'partitions' must be an array.", { code: error.ErrorCodes.ERR__INVALID_ARG });
+    }
+
+    const { topic, partitions } = options;
+
+    // Create an array of TopicPartitionOffset objects
+    const topicPartitionOffsets = [];
+
+    for (const partition of partitions) {
+      if (partition.offset === null || partition.offset === undefined) {
+        throw new error.KafkaJSError("Each partition must have a valid offset.", { code: error.ErrorCodes.ERR__INVALID_ARG });
+      }
+
+      const offset = +partition.offset;
+      if (isNaN(offset)) {
+        throw new error.KafkaJSError("Offset must be a valid number.", { code: error.ErrorCodes.ERR__INVALID_ARG });
+      }
+
+      topicPartitionOffsets.push({
+        topic,
+        partition: partition.partition,
+        offset: offset,
+      });
+    }
+
+    return new Promise((resolve, reject) => {
+      this.#internalClient.deleteRecords(topicPartitionOffsets, options, (err, results) => {
+        if (err) {
+          reject(createKafkaJsErrorFromLibRdKafkaError(err));
+        } else {
+          resolve(results);
+        }
+      });
+    });
+  }
 }
 
 module.exports = {

src/admin.cc

@@ -115,6 +115,7 @@ void AdminClient::Init(v8::Local<v8::Object> exports) {
   Nan::SetPrototypeMethod(tpl, "createTopic", NodeCreateTopic);
   Nan::SetPrototypeMethod(tpl, "deleteTopic", NodeDeleteTopic);
   Nan::SetPrototypeMethod(tpl, "createPartitions", NodeCreatePartitions);
+  Nan::SetPrototypeMethod(tpl, "deleteRecords", NodeDeleteRecords);
 
   // Consumer group related operations
   Nan::SetPrototypeMethod(tpl, "listGroups", NodeListGroups);
@@ -743,6 +744,74 @@ Baton AdminClient::ListConsumerGroupOffsets(
   }
 }
 
+Baton AdminClient::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) {
+  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 records
+    rd_kafka_AdminOptions_t *options = rd_kafka_AdminOptions_new(
+        m_client->c_ptr(), RD_KAFKA_ADMIN_OP_DELETERECORDS);
+
+    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);
+    }
+
+    err = rd_kafka_AdminOptions_set_operation_timeout(
+        options, operation_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_DeleteRecords(m_client->c_ptr(), del_records,
+                           del_records_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_DELETERECORDS_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();
@@ -1147,10 +1216,7 @@ NAN_METHOD(AdminClient::NodeListConsumerGroupOffsets) {
   }
 
   // Now process the second argument: options (timeout and requireStableOffsets)
-  v8::Local<v8::Object> options = Nan::New<v8::Object>();
-  if (info.Length() > 2 && info[1]->IsObject()) {
-    options = info[1].As<v8::Object>();
-  }
+  v8::Local<v8::Object> options = info[1].As<v8::Object>();
 
   bool require_stable_offsets =
       GetParameter<bool>(options, "requireStableOffsets", false);
@@ -1167,4 +1233,69 @@ NAN_METHOD(AdminClient::NodeListConsumerGroupOffsets) {
       require_stable_offsets, timeout_ms));
 }
 
+/**
+ * Delete Records.
+ */
+NAN_METHOD(AdminClient::NodeDeleteRecords) {
+  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 array containg 'TopicPartitionOffset' objects");
+  }
+
+  if (!info[1]->IsObject()) {
+    return Nan::ThrowError("Must provide 'options' object");
+  }
+
+  // Get list of TopicPartitions to delete records from
+  // and convert it into rd_kafka_DeleteRecords_t array
+  v8::Local<v8::Array> delete_records_list = info[0].As<v8::Array>();
+
+  if (delete_records_list->Length() == 0) {
+    return Nan::ThrowError("Must provide at least one TopicPartitionOffset");
+  }
+
+  /**
+   * The ownership of this is taken by
+   * Workers::AdminClientDeleteRecords and freeing it is also handled
+   * by that class.
+   */
+  rd_kafka_DeleteRecords_t **delete_records =
+      static_cast<rd_kafka_DeleteRecords_t **>(
+          malloc(sizeof(rd_kafka_DeleteRecords_t *) * 1));
+
+  rd_kafka_topic_partition_list_t *partitions =
+      Conversion::TopicPartition::TopicPartitionv8ArrayToTopicPartitionList(
+          delete_records_list, true);
+  if (partitions == NULL) {
+    return Nan::ThrowError(
+        "Failed to convert objects in delete records list, provide proper "
+        "TopicPartitionOffset objects");
+  }
+  delete_records[0] = rd_kafka_DeleteRecords_new(partitions);
+
+  rd_kafka_topic_partition_list_destroy(partitions);
+
+  // Now process the second argument: options (timeout and operation_timeout)
+  v8::Local<v8::Object> options = info[1].As<v8::Object>();
+
+  int operation_timeout_ms =
+      GetParameter<int64_t>(options, "operation_timeout", 60000);
+  int timeout_ms = GetParameter<int64_t>(options, "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 worker to process the offset fetch request asynchronously
+  Nan::AsyncQueueWorker(new Workers::AdminClientDeleteRecords(
+      callback, client, delete_records, 1, operation_timeout_ms, timeout_ms));
+}
+
 }  // namespace NodeKafka