Repository navigation
Conversation
nack() on a batch seeked only the partition of the first record, so the other partitions' records were not redelivered and a later successful batch committed past them. Both the aiokafka and the confluent message now seek each partition in the batch to the first offset the batch holds there. Fixes ag2ai#3280.
|
|
Collaborator
|
Hi. I have to close your Pull Request because someone else submitted #3281 before you, and I merged it. |
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.
Description
KafkaAckableMessage.nack()seeked only the partition ofraw_message[0]. A batch can span partitions, as in the issue's[(0, 1), (0, 2), (1, 0), (1, 1), (2, 0), (2, 1)]. WithAckPolicy.NACK_ON_ERROR, only partition 0 was rewound, and the next successful batch'scommit()committed past the records of partitions 1 and 2 that were never processed.nack()now collects the lowest offset the batch holds in each partition and seeks every one of them. A single message works as before: one seek to its own offset. The confluentKafkaMessage.nack()had the sameraw_message[0]pattern, so it gets the same change: oneseek(topic, partition, offset)per partition in the batch. Records without a topic, partition or offset are still skipped.Fixes #3280
Type of change
Checklist
ruff check,ruff format --check(0.14.14) andmypy(1.19.1) are clean on the changed files. I didn't run the fulljust lint.just test-coverage: not run in full, see below.New tests in
tests/brokers/kafka/test_message_nack.pyandtests/brokers/confluent/test_message_nack.pybuild the message with a stub consumer that records itsseek()calls:Without the change, the two batch tests fail (one seek, for partition 0 only), and the single-message tests pass. I had no Kafka broker, so I didn't rerun the issue's real-broker scenario. In my container, the
test_test_client.pysuites for both backends gave 167 passed and 10 failed. The first failure isKafkaConnectionError: Unable to bootstrap from localhost:9092(a test that wants a real broker). I didn't compare the other nine againstmain.