[KIP-932]: Implement api to acknowledge deserialization failure offsets - #5515
Open
Pratyush Ranjan (PratRanj07) wants to merge 1 commit into
Open
[KIP-932]: Implement api to acknowledge deserialization failure offsets#5515Pratyush Ranjan (PratRanj07) wants to merge 1 commit into
Pratyush Ranjan (PratRanj07) wants to merge 1 commit into
Conversation
|
🎉 All Contributor License Agreements have been signed. Ready to merge. |
Copilot started reviewing on behalf of
Pratyush Ranjan (PratRanj07)
June 20, 2026 23:40
View session
There was a problem hiding this comment.
Pull request overview
Adds a share-consumer bulk API intended for higher-level client libraries to release (RELEASE) a set of offsets that failed client-side deserialization, enabling broker redelivery without using the per-message acknowledge APIs (which are gated in implicit ack mode).
Changes:
- Introduces
rd_kafka_topic_partition_set_offsets()to attach an offsets array tord_kafka_topic_partition_tvia its private glue. - Implements
rd_kafka_share_acknowledge_deserialization_failure()to mark attached offsets as RELEASE (best-effort, with aggregated error reporting). - Adds integration tests covering explicit mode, implicit partial release behavior, and error surfacing.
Reviewed changes
Copilot reviewed 5 out of 5 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| tests/0172-share_consumer_acknowledge.c | Adds test cases validating bulk-release/redelivery semantics and error handling for the new API. |
| src/rdkafka.h | Declares the new public APIs and documents intended usage for client libraries. |
| src/rdkafka_share_acknowledgement.c | Implements the bulk deserialization-failure acknowledge/release operation. |
| src/rdkafka_partition.h | Extends rd_kafka_topic_partition_private_t to hold an optional offsets array. |
| src/rdkafka_partition.c | Implements offsets attach/clear, ensures deep-copy in partition copy/update paths, and frees offsets in private destroy. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+3488
to
+3490
| rd_kafka_error_t *rd_kafka_share_acknowledge_deserialization_failure( | ||
| rd_kafka_share_t *rkshare, | ||
| const rd_kafka_topic_partition_list_t *partitions); |
Comment on lines
+1065
to
+1067
| rd_kafka_error_t *rd_kafka_share_acknowledge_deserialization_failure( | ||
| rd_kafka_share_t *rkshare, | ||
| const rd_kafka_topic_partition_list_t *partitions) { |
Comment on lines
+1131
to
+1135
| err = rd_kafka_share_find_ack_entry( | ||
| rkshare, rktpar->topic, rktpar->partition, offset, | ||
| &entry, &idx); | ||
| if (err) | ||
| goto record_err; |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Implement api to acknowledge deserialization failure offsets