Added a delete records api

CHANGELOG.md

@@ -1,14 +1,17 @@
 # Confluent's Python client for Apache Kafka
 
-## v2.4.1
+## v2.5.0
 
-v2.4.1 is a maintenance release with the following fixes and enhancements:
+v2.5.0 is a feature release with the following features, fixes and enhancements:
 
+ - [KIP-107](https://cwiki.apache.org/confluence/display/KAFKA/KIP-107%3A+Add+deleteRecordsBefore%28%29+API+in+AdminClient) Added delete_records API (#1710)
  - Removed usage of `strcpy` to enhance security of the client (#1745)
  - Fixed invalid write in `OAUTHBEARER/OIDC` extensions copy (#1745)
+ - 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)
+
+confluent-kafka-python is based on librdkafka v2.5.0, see the
+[librdkafka release notes](https://github.com/confluentinc/librdkafka/releases/tag/v2.5.0)
 for a complete list of changes, enhancements, fixes and upgrade considerations.
 
 

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,
+                     DeletedRecords)
 
 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 DeletedRecords:
+    """
+    DeletedRecords
+    Represents information about deleted records.
+
+    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,8 @@ 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: `socket.timeout.ms/1000.0`
         :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 +588,8 @@ 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: `socket.timeout.ms/1000.0`
         :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 +627,8 @@ 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: `socket.timeout.ms/1000.0`
         :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 +1218,38 @@ 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, **kwargs):
+        """
+        Deletes all the records before the specified offsets (not including),
+        in the specified Topic and Partitions.
+
+        param list(TopicPartitions) topic_partition_offsets: A list of
+                TopicPartition objects having `offset` field set to the offset
+                before which all the records should be deleted.
+        :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: `socket.timeout.ms/1000.0`
+
+        :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)
+
+        f, futmap = AdminClient._make_futures_v2(
+            topic_partition_offsets, _TopicPartition, AdminClient._make_futmap_result)
+
+        super(AdminClient, self).delete_records(topic_partition_offsets, f, **kwargs)
+        return futmap