You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "bbejeck (via GitHub)" <gi...@apache.org> on 2026/04/18 22:59:18 UTC
[PR] KAFKA-20456: Bound waitForFuture() timeout to less than max.poll.interval [kafka]
bbejeck opened a new pull request, #22092:
URL: https://github.com/apache/kafka/pull/22092
- `TaskManager.waitForFuture()` previously used a hardcoded 5-minute timeout when waiting for the `StateUpdater` to process a REMOVE action. When the `StateUpdater` is blocked (e.g., RocksDB write stall during
changelog restoration), the StreamThread blocks for the full timeout duration and cannot poll, exceeding `max.poll.interval.ms` and getting kicked from the consumer group. This triggers a rebalance cascade
that can lead to a crash loop.
- The timeout is now derived from the consumer's `max.poll.interval.ms` config (maxPollIntervalMs / 2), ensuring the `StreamThread `regains control and can poll before the consumer is removed from the group.
- This is a mitigation — the underlying task leak (when waitForFuture() times out and the task is silently dropped) is addressed in a follow-up PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20456: Bound waitForFuture() timeout to less than max.poll.interval [kafka]
Posted by "UladzislauBlok (via GitHub)" <gi...@apache.org>.
UladzislauBlok commented on code in PR #22092:
URL: https://github.com/apache/kafka/pull/22092#discussion_r3106741139
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java:
##########
@@ -4197,6 +4200,23 @@ private void runUntilTimeoutOrException(final Runnable action) {
}
}
+ @Test
+ public void shouldSetWaitForFutureTimeoutFromMaxPollIntervalMs() {
+ final Properties properties = configProps(false, false);
+ properties.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "20000");
+ final StreamsConfig config = new StreamsConfig(properties);
+ thread = createStreamThread(CLIENT_ID, config);
+
+ assertThat(thread.taskManager().waitForFutureTimeoutMs(), equalTo(10_000L));
Review Comment:
minor: shouldn't we use assertj?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -110,6 +110,7 @@ public class TaskManager {
private final StandbyTaskCreator standbyTaskCreator;
private final StateUpdater stateUpdater;
private final DefaultTaskManager schedulingTaskManager;
+ private final long waitForFutureTimeoutMs;
TaskManager(final Time time,
Review Comment:
```suggestion
private final DefaultTaskManager schedulingTaskManager;
private final long waitForFutureTimeoutMs;
TaskManager(final Time time,
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20456: Bound waitForFuture() timeout to less than max.poll.interval [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on PR #22092:
URL: https://github.com/apache/kafka/pull/22092#issuecomment-4284592313
> Don't you think max.poll.interval.ms / 2 is too strict? It can be something with percentage, e.g. max.poll.interval.ms * 0.9 with conversion back to long
I'm not sure. One thing, this PR part of another stacked one - https://github.com/apache/kafka/pull/22094 this fix isn't meant to go in by itself, so I think giving up early is ok as the task gets cleaned up. My concern would be not giving the `StreamThread` enough time to do other tasks before the embedded consumer can call `poll()` again. What would think about .75?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20456: Bound waitForFuture() timeout to less than max.poll.interval [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck closed pull request #22092: KAFKA-20456: Bound waitForFuture() timeout to less than max.poll.interval
URL: https://github.com/apache/kafka/pull/22092
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20456: Bound waitForFuture() timeout to less than max.poll.interval [kafka]
Posted by "UladzislauBlok (via GitHub)" <gi...@apache.org>.
UladzislauBlok commented on PR #22092:
URL: https://github.com/apache/kafka/pull/22092#issuecomment-4284603463
Hey
90% was just and example
75% works as well for me
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20456: Bound waitForFuture() timeout to less than max.poll.interval [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on code in PR #22092:
URL: https://github.com/apache/kafka/pull/22092#discussion_r3113894393
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java:
##########
@@ -4197,6 +4200,23 @@ private void runUntilTimeoutOrException(final Runnable action) {
}
}
+ @Test
+ public void shouldSetWaitForFutureTimeoutFromMaxPollIntervalMs() {
+ final Properties properties = configProps(false, false);
+ properties.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "20000");
+ final StreamsConfig config = new StreamsConfig(properties);
+ thread = createStreamThread(CLIENT_ID, config);
+
+ assertThat(thread.taskManager().waitForFutureTimeoutMs(), equalTo(10_000L));
Review Comment:
It's not a dependency in AK right now, we use Junit+Hamcrest, plus IMHO what's there now is clear enough.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org