KAFKA-16243: Make sure that we do not exceed max poll interval inside poll #15372
+41
−4
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
… poll The consumer keeps a poll timer which is used to ensure liveness of the application thread. The poll timer automatically updates while the `Consumer.poll(Duration)` method is blocked, while the newer consumer only updates the poll timer when a new call to `Consumer.poll(Duration)` is issued. This means that the kafka-console-consumer.sh tools, which uses a very long timeout by default, works differently with the new consumer, with the consumer proactively rejoining the group during long poll timeouts. We solve the problem by (a) repeatedly sending `PollApplicationEvents` to the background thread, not just on the first call of `poll` and (b) making sure that the application thread doesn't block for so long that it runs out of `max.poll.interval`. An integration test is added to make sure that we do not rejoin the group when a long poll timeout is used with a low `max.poll.interval`.
|
@cadonna could you please have a look? |
lianetm
reviewed
Feb 15, 2024
|
|
||
| val initialAssignedCalls = listener.callsToAssigned | ||
|
|
||
| consumer.poll(Duration.ofMillis(2000)); |
There was a problem hiding this comment.
nit: semicolon not needed here, same in the next 2 lines
|
thanks for the changes! lgtm. |
|
I like the last commit msg :) |
cadonna
approved these changes
Feb 19, 2024
|
|
||
| consumer.poll(Duration.ofMillis(2000)) | ||
|
|
||
| // Give enough time to rejoin |
There was a problem hiding this comment.
nit:
This comment is a bit confusing. What is it supposed to clarify?
| consumer.poll(Duration.ofMillis(500)) | ||
| consumer.poll(Duration.ofMillis(500)) | ||
|
|
||
| // Check that we did not rejoin |
There was a problem hiding this comment.
Do we need this comment? I think it would be better to delete it and to rename initialAssignedCalls to something more meaningful like callsToAssignedAfterFirstRebalance or callsToAssignedBeforePolls.
mjsax
reviewed
Feb 23, 2024
| ) { | ||
| return 0L; | ||
| } | ||
| return Math.min(pollTimer.remainingMs() / 2, heartbeatRequestState.nextHeartbeatMs(currentTimeMs)); |
There was a problem hiding this comment.
It's somewhat arbitrary. We need to make sure that the application thread doesn't block so long that the poll timer expires. We want to let it unblock sometime before the timer expires, and send an event to the background thread that it's still polling, and give the background thread time to process the event and reset the poll timer.
clolov
pushed a commit
to clolov/kafka
that referenced
this pull request
Apr 5, 2024
… poll (apache#15372) The consumer keeps a poll timer, which is used to ensure liveness of the application thread. The poll timer automatically updates while the Consumer.poll(Duration) method is blocked, while the newer consumer only updates the poll timer when a new call to Consumer.poll(Duration) is issued. This means that the kafka-console-consumer.sh tools, which uses a very long timeout by default, works differently with the new consumer, with the consumer proactively rejoining the group during long poll timeouts. This change solves the problem by (a) repeatedly sending PollApplicationEvents to the background thread, not just on the first call of poll and (b) making sure that the application thread doesn't block for so long that it runs out of max.poll.interval. An integration test is added to make sure that we do not rejoin the group when a long poll timeout is used with a low max.poll.interval. Reviewers: Lianet Magrans <[email protected]>, Andrew Schofield <[email protected]>, Bruno Cadonna <[email protected]>
Phuc-Hong-Tran
pushed a commit
to Phuc-Hong-Tran/kafka
that referenced
this pull request
Jun 6, 2024
… poll (apache#15372) The consumer keeps a poll timer, which is used to ensure liveness of the application thread. The poll timer automatically updates while the Consumer.poll(Duration) method is blocked, while the newer consumer only updates the poll timer when a new call to Consumer.poll(Duration) is issued. This means that the kafka-console-consumer.sh tools, which uses a very long timeout by default, works differently with the new consumer, with the consumer proactively rejoining the group during long poll timeouts. This change solves the problem by (a) repeatedly sending PollApplicationEvents to the background thread, not just on the first call of poll and (b) making sure that the application thread doesn't block for so long that it runs out of max.poll.interval. An integration test is added to make sure that we do not rejoin the group when a long poll timeout is used with a low max.poll.interval. Reviewers: Lianet Magrans <[email protected]>, Andrew Schofield <[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.
The consumer keeps a poll timer, which is used to ensure liveness of the application thread. The poll timer automatically updates while the
Consumer.poll(Duration)method is blocked, while the newer consumer only updates the poll timer when a new call toConsumer.poll(Duration)is issued. This means that the kafka-console-consumer.sh tools, which uses a very long timeout by default, works differently with the new consumer, with the consumer proactively rejoining the group during long poll timeouts.This change solves the problem by (a) repeatedly sending
PollApplicationEventsto the background thread, not just on the first call ofpolland (b) making sure that the application thread doesn't block for so long that it runs out ofmax.poll.interval.An integration test is added to make sure that we do not rejoin the group when a long poll timeout is used with a low
max.poll.interval.Committer Checklist (excluded from commit message)