Added a delete records api #1710
Conversation
|
|
src/confluent_kafka/src/Admin.c
Outdated
| 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 = c_parts_to_py(c_delete_records_res_list); |
There was a problem hiding this comment.
I'm not sure the responses are returned in the same order as the requests, so that it's possible to use _make_futmap_result_from_list
Also we should use a DeletedRecords object as in Java, instead of a TopicPartition because that could be extended in future and in that case we cannot apply the same changes.
There was a problem hiding this comment.
We are converting it to the futmap in the end keyed at TopicPartition so there is no issue in that I think.
We can create DeletedRecords class as Java instead of directly using offset in the response.
There was a problem hiding this comment.
Created a new class DeleteRecords containing only offset and made changes according to that. Please review.
|
|
||
| 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. |
There was a problem hiding this comment.
| before which all the records should be deleted. | |
| 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. |
There was a problem hiding this comment.
Add DeletedRecords to index.rst after ConsumerGroupDescription
- :ref:`DeletedRecords <pythonclient_deleted_records>`
and
.. _pythonclient_deleted_records:
**************
DeletedRecords
**************
.. autoclass:: confluent_kafka.admin.DeletedRecords
:members:
| in the specified Topic and Partitions. | ||
|
|
||
| :param list(TopicPartitions) topic_partition_offsets: A list of | ||
| TopicPartition objects having `offset` field set to the offset |
There was a problem hiding this comment.
| TopicPartition objects having `offset` field set to the offset | |
| :class:`.TopicPartition` objects having `offset` field set to the offset |
| 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 |
There was a problem hiding this comment.
| controlling how long the delete_records request will block | |
| controlling how long the `delete_records` request will block |
| in the cluster. A value of 0 returns immediately. | ||
| Default: `socket.timeout.ms/1000.0` | ||
|
|
||
| :returns: A dict of futures keyed by the TopicPartition. |
There was a problem hiding this comment.
| :returns: A dict of futures keyed by the TopicPartition. | |
| :returns: A dict of futures keyed by the :class:`.TopicPartition`. |
| Default: `socket.timeout.ms/1000.0` | ||
|
|
||
| :returns: A dict of futures keyed by the TopicPartition. | ||
| The future result() method returns DeletedRecords |
There was a problem hiding this comment.
| The future result() method returns DeletedRecords | |
| The future result() method returns :class:`.DeletedRecords` |
|
|
||
| :returns: A dict of futures keyed by the TopicPartition. | ||
| The future result() method returns DeletedRecords | ||
| or raises KafkaException |
There was a problem hiding this comment.
| or raises KafkaException | |
| or raises :class:`.KafkaException` |
| def delete_records(self, topic_partition_offsets, **kwargs): | ||
| """ | ||
| Deletes all the records before the specified offsets (not including), | ||
| in the specified Topic and Partitions. |
There was a problem hiding this comment.
Written this way they seem like classes
| in the specified Topic and Partitions. | |
| in the specified topics and partitions. |
Added a delete records api and created integration tests and unit tests for the same