KAFKA-15842: Correct handling of KafkaConsumer.committed for new consumer #14859
Conversation
This file contains 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
Merge commits from trunk
|
Closing and reopening PR to retry failed build. |
|
https://issues.apache.org/jira/browse/KAFKA-15932 opened for flaky test. |
kirktrue
suggested changes
Nov 30, 2023
There was a problem hiding this comment.
Thanks @AndrewJSchofield!
Thanks for the fix and the associated clean up. One minor nit-ty change requested.
| @@ -48,7 +48,7 @@ | |||
| <suppress id="dontUseSystemExit" | |||
| files="Exit.java"/> | |||
| <suppress checks="ClassFanOutComplexity" | |||
| files="(AbstractFetch|Sender|SenderTest|ConsumerCoordinator|KafkaConsumer|PrototypeAsyncConsumer|KafkaProducer|Utils|TransactionManager|TransactionManagerTest|KafkaAdminClient|NetworkClient|Admin|KafkaRaftClient|KafkaRaftClientTest|RaftClientTestContext).java"/> | |||
There was a problem hiding this comment.
Thank you for this clean up!
| @@ -663,6 +662,10 @@ public Map<TopicPartition, OffsetAndMetadata> committed(final Set<TopicPartition | |||
| time.timer(timeout)); | |||
| committedOffsets.forEach(this::updateLastSeenEpochIfNewer); | |||
| return committedOffsets; | |||
| } catch (TimeoutException e) { | |||
| throw new TimeoutException("Timeout of " + timeout.toMillis() + "ms expired before the last " + | |||
| "committed offset for partitions " + partitions + " could be determined. Try tuning default.api.timeout.ms " + | |||
There was a problem hiding this comment.
nit: please replace the literal string default.api.timeout.ms with the appropriate constant from ConsumerConfig 🥺
There was a problem hiding this comment.
While I agree with @kirktrue , let's merge this PR and fix this on the side in a follow-up PR.
Comment on lines
-656
to
+655
| return new HashMap<>(); | ||
| return Collections.emptyMap(); |
There was a problem hiding this comment.
Does this make a difference, or just a stylistic/pet peeve/optimiz/sation thing?
There was a problem hiding this comment.
I made it strictly identical to the previous implementation.
Comment on lines
-598
to
+599
| @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll")) | ||
| @MethodSource(Array("getTestQuorumAndGroupProtocolParametersGenericGroupProtocolOnly")) |
cadonna
approved these changes
Dec 1, 2023
ex172000
pushed a commit
to ex172000/kafka
that referenced
this pull request
Dec 15, 2023
…umer (apache#14859) This PR fixes some details of the interface to KafkaConsumer.committed which were different between the existing consumer and the new consumer. Adds a unit test that validates the behaviour is the same for both consumer implementations. Reviewers: Kirk True <[email protected]>, Bruno Cadonna <[email protected]>
yyu1993
pushed a commit
to yyu1993/kafka
that referenced
this pull request
Feb 15, 2024
…umer (apache#14859) This PR fixes some details of the interface to KafkaConsumer.committed which were different between the existing consumer and the new consumer. Adds a unit test that validates the behaviour is the same for both consumer implementations. Reviewers: Kirk True <[email protected]>, Bruno Cadonna <[email protected]>
AnatolyPopov
pushed a commit
to aiven/kafka
that referenced
this pull request
Feb 16, 2024
…umer (apache#14859) This PR fixes some details of the interface to KafkaConsumer.committed which were different between the existing consumer and the new consumer. Adds a unit test that validates the behaviour is the same for both consumer implementations. Reviewers: Kirk True <[email protected]>, Bruno Cadonna <[email protected]>
clolov
pushed a commit
to clolov/kafka
that referenced
this pull request
Apr 5, 2024
…umer (apache#14859) This PR fixes some details of the interface to KafkaConsumer.committed which were different between the existing consumer and the new consumer. Adds a unit test that validates the behaviour is the same for both consumer implementations. Reviewers: Kirk True <[email protected]>, Bruno Cadonna <[email protected]>
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.
This PR fixes some details of the interface to KafkaConsumer.committed which were different between the existing consumer and the new consumer.
Adds a unit test that validates the behaviour is the same for both consumer implementations.
Committer Checklist (excluded from commit message)