Added a delete records api

CHANGELOG.md

@@ -6,6 +6,8 @@ v2.4.1 is a maintenance release with the following fixes and enhancements:
 
  - Removed usage of `strcpy` to enhance security of the client (#1745)
  - Fixed invalid write in `OAUTHBEARER/OIDC` extensions copy (#1745)
+ - [KIP-107](https://cwiki.apache.org/confluence/display/KAFKA/KIP-107%3A+Add+deleteRecordsBefore%28%29+API+in+AdminClient) Added delete_records API (#1710)
+ - Fixed documentation for default value of operation_timeout in create_topics, delete_topics, create_partitions apis (#1710)
 
 confluent-kafka-python is based on librdkafka v2.4.1, see the
 [librdkafka release notes](https://github.com/confluentinc/librdkafka/releases/tag/v2.4.1)

examples/adminapi.py

@@ -808,24 +808,24 @@ def example_list_offsets(a, args):
                 f"Invalid number of arguments for list offsets, partition {partition_i}, expected 3," +
                 f" got {len(args) - i}")
         topic = args[i]
-        partition = int(args[i+1])
+        partition = int(args[i + 1])
         topic_partition = TopicPartition(topic, partition)
 
-        if "EARLIEST" == args[i+2]:
+        if "EARLIEST" == args[i + 2]:
             offset_spec = OffsetSpec.earliest()
 
-        elif "LATEST" == args[i+2]:
+        elif "LATEST" == args[i + 2]:
             offset_spec = OffsetSpec.latest()
 
-        elif "MAX_TIMESTAMP" == args[i+2]:
+        elif "MAX_TIMESTAMP" == args[i + 2]:
             offset_spec = OffsetSpec.max_timestamp()
 
-        elif "TIMESTAMP" == args[i+2]:
+        elif "TIMESTAMP" == args[i + 2]:
             if i + 4 > len(args):
                 raise ValueError(
                     f"Invalid number of arguments for list offsets, partition {partition_i}, expected 4" +
                     f", got {len(args) - i}")
-            offset_spec = OffsetSpec.for_timestamp(int(args[i+3]))
+            offset_spec = OffsetSpec.for_timestamp(int(args[i + 3]))
             i += 1
         else:
             raise ValueError("Invalid OffsetSpec, must be EARLIEST, LATEST, MAX_TIMESTAMP or TIMESTAMP")
@@ -845,6 +845,35 @@ def example_list_offsets(a, args):
                   .format(partition.topic, partition.partition, e))
 
 
+def example_delete_records(a, args):
+    if len(args) == 0:
+        raise ValueError(
+            "Invalid number of arguments for delete_records, expected at least 3 " +
+            "(Usage: delete_records <topic1> <partition1> <offset1> [<topic2> <partition2> <offset2>...])")
+    if len(args) % 3 != 0:
+        raise ValueError(
+            "Invalid number of arguments for delete_records " +
+            "(Usage: delete_records <topic1> <partition1> <offset1> [<topic2> <partition2> <offset2>...])")
+
+    topic_partition_offset = [
+        TopicPartition(topic, int(partition), int(offset))
+        for topic, partition, offset in zip(args[::3], args[1::3], args[2::3])
+    ]
+
+    futmap = a.delete_records(topic_partition_offset)
+    for partition, fut in futmap.items():
+        try:
+            result = fut.result()
+            print(
+                f"All records deleted before offset {partition.offset} in topic {partition.topic}" +
+                f" partition {partition.partition}. The minimum offset in this partition" +
+                f" is now {result.low_watermark}")
+        except KafkaException as e:
+            print(
+                f"Error deleting records in topic {partition.topic} partition {partition.partition}" +
+                f" before offset {partition.offset}: {e}")
+
+
 if __name__ == '__main__':
     if len(sys.argv) < 3:
         sys.stderr.write('Usage: %s <bootstrap-brokers> <operation> <args..>\n\n' % sys.argv[0])
@@ -883,7 +912,7 @@ def example_list_offsets(a, args):
                          ' <password2> <salt2> DELETE <user3> <mechanism3> ..]\n')
         sys.stderr.write(' list_offsets <isolation_level> <topic1> <partition1> <offset_spec1> ' +
                          '[<topic2> <partition2> <offset_spec2> ..]\n')
-
+        sys.stderr.write(' delete_records <topic1> <partition1> <offset1> [<topic2> <partition2> <offset2>...]\n')
         sys.exit(1)
 
     broker = sys.argv[1]
@@ -913,7 +942,8 @@ def example_list_offsets(a, args):
               'alter_consumer_group_offsets': example_alter_consumer_group_offsets,
               'describe_user_scram_credentials': example_describe_user_scram_credentials,
               'alter_user_scram_credentials': example_alter_user_scram_credentials,
-              'list_offsets': example_list_offsets}
+              'list_offsets': example_list_offsets,
+              'delete_records': example_delete_records}
 
     if operation not in opsmap:
         sys.stderr.write('Unknown operation: %s\n' % operation)

src/confluent_kafka/__init__.py

@@ -24,7 +24,8 @@
                      ConsumerGroupState,
                      TopicCollection,
                      TopicPartitionInfo,
-                     IsolationLevel)
+                     IsolationLevel,
+                     DeleteRecords)
 
 from .cimpl import (Producer,
                     Consumer,
@@ -49,7 +50,7 @@
            'SerializingProducer', 'TIMESTAMP_CREATE_TIME', 'TIMESTAMP_LOG_APPEND_TIME',
            'TIMESTAMP_NOT_AVAILABLE', 'TopicPartition', 'Node',
            'ConsumerGroupTopicPartitions', 'ConsumerGroupState', 'Uuid',
-           'IsolationLevel']
+           'IsolationLevel', 'DeletedRecords']
 
 __version__ = version()[0]
 

src/confluent_kafka/_model/__init__.py

@@ -149,3 +149,17 @@ def __lt__(self, other):
         if self.__class__ != other.__class__:
             return NotImplemented
         return self.value < other.value
+
+
+class DeleteRecords:
+    """
+    DeleteRecords
+    Result of a `AdminClient.delete_records` call associated to a partition.
+
+    Parameters
+    ----------
+    low_watermark: int
+        The "low watermark" for the topic partition on which the deletion was executed.
+    """
+    def __init__(self, low_watermark):
+        self.low_watermark = low_watermark

src/confluent_kafka/admin/__init__.py

@@ -534,6 +534,17 @@ def _check_list_offsets_request(topic_partition_offsets, kwargs):
             if not isinstance(kwargs['isolation_level'], _IsolationLevel):
                 raise TypeError("isolation_level argument should be an IsolationLevel")
 
+    @staticmethod
+    def _check_delete_records(request):
+        if not isinstance(request, list):
+            raise TypeError(f"Expected Request to be a list, got '{type(request).__name__}' ")
+        for req in request:
+            if not isinstance(req, _TopicPartition):
+                raise TypeError("Element of the request list must be of type 'TopicPartition'" +
+                                f" got '{type(req).__name__}' ")
+            if req.partition < 0:
+                raise ValueError("'partition' cannot be negative")
+
     def create_topics(self, new_topics, **kwargs):
         """
         Create one or more new topics.
@@ -543,7 +554,7 @@ def create_topics(self, new_topics, **kwargs):
         :param float operation_timeout: The operation timeout in seconds,
                   controlling how long the CreateTopics request will block
                   on the broker waiting for the topic creation to propagate
-                  in the cluster. A value of 0 returns immediately. Default: 0
+                  in the cluster. A value of 0 returns immediately. Default: 60
         :param float request_timeout: The overall request timeout in seconds,
                   including broker lookup, request transmission, operation time
                   on broker, and response. Default: `socket.timeout.ms*1000.0`
@@ -576,7 +587,7 @@ def delete_topics(self, topics, **kwargs):
         :param float operation_timeout: The operation timeout in seconds,
                   controlling how long the DeleteTopics request will block
                   on the broker waiting for the topic deletion to propagate
-                  in the cluster. A value of 0 returns immediately. Default: 0
+                  in the cluster. A value of 0 returns immediately. Default: 60
         :param float request_timeout: The overall request timeout in seconds,
                   including broker lookup, request transmission, operation time
                   on broker, and response. Default: `socket.timeout.ms*1000.0`
@@ -614,7 +625,7 @@ def create_partitions(self, new_partitions, **kwargs):
         :param float operation_timeout: The operation timeout in seconds,
                   controlling how long the CreatePartitions request will block
                   on the broker waiting for the partition creation to propagate
-                  in the cluster. A value of 0 returns immediately. Default: 0
+                  in the cluster. A value of 0 returns immediately. Default: 60
         :param float request_timeout: The overall request timeout in seconds,
                   including broker lookup, request transmission, operation time
                   on broker, and response. Default: `socket.timeout.ms*1000.0`
@@ -1204,3 +1215,36 @@ def list_offsets(self, topic_partition_offsets, **kwargs):
 
         super(AdminClient, self).list_offsets(topic_partition_offsets_list, f, **kwargs)
         return futmap
+
+    def delete_records(self, topic_partition_offsets_list, **kwargs):
+        """
+        Deletes all the records before the specified offset,
+        in the specified Topic and Partition.
+
+        :param list(TopicPartitions) topic_partitions_offset_list: A list of TopicPartition objects
+        consisting of the Topic Partition and Offsets on which we have to perform the deleteion.
+        :param float request_timeout: The overall request timeout in seconds,
+                    including broker lookup, request transmission, operation time
+                    on broker, and response. Default: `socket.timeout.ms*1000.0`
+        :param float operation_timeout: The operation timeout in seconds,
+                  controlling how long the delete_records request will block
+                  on the broker waiting for the record deletion to propagate
+                  in the cluster. A value of 0 returns immediately. Default: 60
+
+        :returns: A dict of futures keyed by the TopicPartition.
+                    The future result() method returns DeletedRecords
+                    or raises KafkaException
+
+        :rtype: dict[TopicPartition, future]
+
+        :raises KafkaException: Operation failed locally or on broker.
+        :raises TypeError: Invalid input type.
+        :raises ValueError: Invalid input value.
+        """
+        AdminClient._check_delete_records(topic_partition_offsets_list)
+
+        f, futmap = AdminClient._make_futures_v2(
+            topic_partition_offsets_list, _TopicPartition, AdminClient._make_futmap_result)
+
+        super(AdminClient, self).delete_records(topic_partition_offsets_list, f, **kwargs)
+        return futmap

src/confluent_kafka/src/Admin.c

@@ -2963,6 +2963,103 @@ const char Admin_list_offsets_doc[] = PyDoc_STR(
         "  This method should not be used directly, use confluent_kafka.AdminClient.list_offsets()\n");
 
 
+/**
+ * @brief Delete records 
+ */
+PyObject* Admin_delete_records (Handle *self,PyObject *args,PyObject *kwargs){
+        PyObject *topic_partition_offset = NULL, *future;
+        int del_record_cnt = 1;
+        rd_kafka_DeleteRecords_t **c_obj = NULL;
+        struct Admin_options options = Admin_options_INITIALIZER;
+        rd_kafka_AdminOptions_t *c_options = NULL;
+        rd_kafka_topic_partition_list_t *c_topic_partition_offset = NULL;
+        CallState cs;
+        rd_kafka_queue_t *rkqu;
+
+        static char *kws[] = {"topic_partition_offset",
+                             "future",
+                             /* options */
+                             "request_timeout",
+                             "operation_timeout",
+                             NULL};
+
+        if (!PyArg_ParseTupleAndKeywords(args, kwargs, "OO|ff", kws,
+                                         &topic_partition_offset,
+                                         &future,
+                                         &options.request_timeout,
+                                         &options.operation_timeout)) {
+                goto err;
+        }
+
+        c_options = Admin_options_to_c(self, RD_KAFKA_ADMIN_OP_DELETERECORDS,
+                                       &options, future);
+        if (!c_options)  {
+                goto err; /* Exception raised by options_to_c() */
+        }
+
+        /* options_to_c() sets future as the opaque, which is used in the
+         * background_event_cb to set the results on the future as the
+         * admin operation is finished, so we need to keep our own refcount. */
+        Py_INCREF(future);
+
+        if (!PyList_Check(topic_partition_offset)) {
+                PyErr_SetString(PyExc_ValueError,
+                        "topic_partitions_offset must be a list");
+                goto err;
+        }
+        c_topic_partition_offset = py_to_c_parts(topic_partition_offset);
+
+        c_obj = malloc(sizeof(rd_kafka_DeleteRecords_t *) * del_record_cnt);
+        c_obj[0] = rd_kafka_DeleteRecords_new(c_topic_partition_offset);
+
+        /* Use librdkafka's background thread queue to automatically dispatch
+        * Admin_background_event_cb() when the admin operation is finished. */
+        rkqu = rd_kafka_queue_get_background(self->rk);
+
+        /*
+         * Call DeleteRecords
+         *
+         * We need to set up a CallState and release GIL here since
+         * the event_cb may be triggered immediately.
+         */
+        CallState_begin(self, &cs);
+        rd_kafka_DeleteRecords(self->rk, c_obj, del_record_cnt, c_options, rkqu);
+        CallState_end(self,&cs);
+
+        rd_kafka_queue_destroy(rkqu); /* drop reference from get_background */
+
+        rd_kafka_AdminOptions_destroy(c_options);
+        rd_kafka_DeleteRecords_destroy_array(c_obj, del_record_cnt);
+        free(c_obj);
+
+        rd_kafka_topic_partition_list_destroy(c_topic_partition_offset);     
+        Py_XDECREF(topic_partition_offset);
+
+        Py_RETURN_NONE;
+err: 
+        if (c_obj) {
+                rd_kafka_DeleteRecords_destroy_array(c_obj, del_record_cnt);
+                free(c_obj);
+        }
+        if (c_options) {
+                rd_kafka_AdminOptions_destroy(c_options);
+                Py_DECREF(future);
+        }
+        if(c_topic_partition_offset) {
+                rd_kafka_topic_partition_list_destroy(c_topic_partition_offset);
+        }
+        Py_XDECREF(topic_partition_offset);
+        return NULL;
+
+}
+
+const char Admin_delete_records_doc[] = PyDoc_STR(
+        ".. py:function:: delete_records(topic_partitions, future, [request_timeout, operation_timeout])\n"
+        "\n"
+        "  Delete all the records for the particular topic partition before the specified offset provided in the request.\n"
+        "\n"
+        "  This method should not be used directly, use confluent_kafka.AdminClient.delete_records()\n");
+
 /**
  * @brief Call rd_kafka_poll() and keep track of crashing callbacks.
  * @returns -1 if callback crashed (or poll() failed), else the number
@@ -3124,6 +3221,10 @@ static PyMethodDef Admin_methods[] = {
         { "list_offsets", (PyCFunction)Admin_list_offsets, METH_VARARGS|METH_KEYWORDS,
            Admin_list_offsets_doc
         },
+
+        { "delete_records", (PyCFunction)Admin_delete_records, METH_VARARGS|METH_KEYWORDS,
+           Admin_delete_records_doc
+        },
 
         { NULL }
 };
@@ -4369,6 +4470,52 @@ static PyObject *Admin_c_ListOffsetsResultInfos_to_py (const rd_kafka_ListOffset
         return NULL;
 }
 
+static PyObject *Admin_c_DeletedRecords_to_py (const rd_kafka_topic_partition_list_t *c_topic_partitions) {
+
+    size_t c_topic_partition_cnt = c_topic_partitions->cnt;
+    PyObject *result = NULL;
+    PyObject *DeleteRecords_type = NULL;
+    size_t i;
+
+    DeleteRecords_type = cfl_PyObject_lookup("confluent_kafka", "DeleteRecords");
+
+    if(!DeleteRecords_type){
+        cfl_PyErr_Format(RD_KAFKA_RESP_ERR__INVALID_ARG, "Unable to load DeleteRecords type");
+        goto raise;
+    }
+
+    result = PyDict_New();
+    for(i=0; i<c_topic_partition_cnt; i++){
+        PyObject *value = NULL;
+        rd_kafka_topic_partition_t *c_topic_partition = &c_topic_partitions->elems[i];
+        if (c_topic_partitions->elems[i].err) {
+            value = KafkaError_new_or_None(c_topic_partitions->elems[i].err, rd_kafka_err2str(c_topic_partitions->elems[i].err));
+        } else {
+            PyObject *args = NULL;
+            PyObject *kwargs = NULL;
+            kwargs = PyDict_New();
+            cfl_PyDict_SetLong(kwargs, "low_watermark", c_topic_partitions->elems[i].offset);
+            args = PyTuple_New(0);
+            value = PyObject_Call(DeleteRecords_type, args, kwargs);
+            Py_DECREF(args);
+            Py_DECREF(kwargs);
+            if (value == NULL)
+                goto raise;
+        }
+        PyObject *key = c_part_to_py(c_topic_partition);
+        PyDict_SetItem(result, key, value);
+        Py_DECREF(key);
+        Py_DECREF(value);
+    }
+
+    Py_DECREF(DeleteRecords_type);
+    return result;
+raise:
+    Py_XDECREF(result);
+    Py_XDECREF(DeleteRecords_type);
+    return NULL;
+}
+
 /**
  * @brief Event callback triggered from librdkafka's background thread
  *        when Admin API results are ready.
@@ -4712,6 +4859,15 @@ static void Admin_background_event_cb (rd_kafka_t *rk, rd_kafka_event_t *rkev,
                 break;
         }
 
+        case RD_KAFKA_EVENT_DELETERECORDS_RESULT:
+        {
+                const rd_kafka_DeleteRecords_result_t *c_delete_records_res = rd_kafka_event_DeleteRecords_result(rkev);
+                const rd_kafka_topic_partition_list_t *c_delete_records_res_list = rd_kafka_DeleteRecords_result_offsets(c_delete_records_res);
+
+                result = Admin_c_DeletedRecords_to_py(c_delete_records_res_list);
+                break;
+        }
+
         default:
                 Py_DECREF(error); /* Py_None */
                 error = KafkaError_new0(RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE,