[KIP-932] : Implement acknowledgement callback for Share Consumer#2253
Open
Kaushik Raina (k-raina) wants to merge 3 commits into
Open
[KIP-932] : Implement acknowledgement callback for Share Consumer#2253Kaushik Raina (k-raina) wants to merge 3 commits into
Kaushik Raina (k-raina) wants to merge 3 commits into
Conversation
Kaushik Raina (k-raina)
requested review from
a team and
Matthew Seal (MSeal)
as code owners
|
🎉 All Contributor License Agreements have been signed. Ready to merge. |
|
| return NULL; | ||
| } | ||
|
|
||
| if (!PyArg_ParseTupleAndKeywords(args, kwargs, "O", kws, &callback)) |
Member
Author
There was a problem hiding this comment.
Check for callback input as null
Pratyush Ranjan (PratRanj07)
requested changes
May 28, 2026
Contributor
Pratyush Ranjan (PratRanj07)
left a comment
Pratyush Ranjan (PratRanj07)
left a comment
There was a problem hiding this comment.
Thanks Kaushik Raina (@k-raina) for the PR! Check the tests on if we should add more. Some I can think of :
- Checking if calling Share consumer functions from the other thread should raise exception. Take a look at librdkafka's rd_kafka_acquire and Java's acquire
- Write some tests like registring a callback and calling poll with explicit mode, then registering a different callbackand then checking if they are being invoked correctly when commiting
Take a look and improve nomenclatures in tests
| * @returns The new Python dict, or NULL on allocation failure with an | ||
| * exception set. | ||
| */ | ||
| PyObject *c_share_partition_offsets_list_to_py( |
Contributor
There was a problem hiding this comment.
Suggested change
| PyObject *c_share_partition_offsets_list_to_py( | |
| PyObject *c_share_partition_offsets_list_to_py_dict( |
| crash: | ||
| CallState_fetch_exception(cs); | ||
| CallState_crash(cs); | ||
| rd_kafka_yield(NULL); |
Contributor
There was a problem hiding this comment.
Should we pass here NULL or rk?
Comment on lines
+1659
to
+1667
| const rd_kafka_share_partition_offsets_t *entry = | ||
| rd_kafka_share_partition_offsets_list_get(list, | ||
| partition_index); | ||
| const rd_kafka_topic_partition_t *rktpar = | ||
| rd_kafka_share_partition_offsets_partition(entry); | ||
| const int64_t *offsets = | ||
| rd_kafka_share_partition_offsets_offsets(entry); | ||
| size_t offsets_count = | ||
| rd_kafka_share_partition_offsets_offsets_cnt(entry); |
Contributor
There was a problem hiding this comment.
We should add NULL checks here
| * | ||
| * | ||
| * Common instance handle for both Producer and Consumer | ||
| * Common instance handle for Producer, Consumer, and ShareConsumer |
Contributor
There was a problem hiding this comment.
Suggested change
| * Common instance handle for Producer, Consumer, and ShareConsumer | |
| * Common instance handle for Producer, Consumer and ShareConsumer |
| batch = [] | ||
| deadline = time.time() + 20.0 | ||
| while time.time() < deadline and len(batch) < num_messages: | ||
| for m in sc.poll(timeout=0.5): |
Contributor
There was a problem hiding this comment.
Change naming
Comment on lines
+1142
to
+1143
| """Calling share-consumer APIs from inside the cb fails with _STATE — | ||
| librdkafka guards every entry point against reentrancy.""" |
Contributor
There was a problem hiding this comment.
Check with Java if this is the correct error. We should match Java semantics
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
2 participants
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.
Summary
(offsets, exception) -> NoneorNoneto clear.ShareConsumer_acknowledgement_commit_cbplus helperc_share_partition_offsets_list_to_py()that converts librdkafka'srd_kafka_share_partition_offsets_list_tintoDict[TopicPartition, frozenset[int]].longer piggybacks on the Consumer fields. GC traverse/clear updated accordingly.
ests/integration/share_consumer/test_share_consumer_ack.pycovering success, error,clear-callback, and lifecycle paths.