Added a delete records api

CHANGELOG.md

@@ -1,18 +1,20 @@
 # 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)
  - Added an example to show the usage of the custom logger with `AdminClient` 
  - 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)
  - Fixed an issue related to import error of `TopicCollection` and `TopicPartitionInfo` classes when importing through other module like mypy.
  - Fixed a segfault when `commit` or `store_offsets` consumer method is called incorrectly with errored Message object
  - Fixed `logger` not working when provided as an argument to `AdminClient` instead of a configuration property
 
-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,39 @@ 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_offsets = [
+        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_offsets)
+    for partition, fut in futmap.items():
+        try:
+            result = fut.result()
+            if partition.offset == -1:
+                print(f"All records deleted in topic {partition.topic} partition {partition.partition}." +
+                      f"The minimum offset in this partition is now {result.low_watermark}")
+            else:
+                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 +916,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 +946,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/admin/__init__.py

@@ -54,6 +54,8 @@
 from ._listoffsets import (OffsetSpec,  # noqa: F401
                            ListOffsetsResultInfo)
 
+from ._records import DeletedRecords  # noqa: F401
+
 from .._model import TopicCollection as _TopicCollection
 
 from ..cimpl import (KafkaException,  # noqa: F401
@@ -535,6 +537,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.
@@ -544,7 +557,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`
@@ -577,7 +591,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`
@@ -615,7 +630,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`
@@ -1205,3 +1221,40 @@ 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.
+               `offset` can be set to :py:const:`OFFSET_END` (-1) to delete all records
+               in the partition.
+        :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

src/confluent_kafka/admin/_records.py

@@ -0,0 +1,27 @@
+# Copyright 2024 Confluent Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+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