You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "chickenchickenlove (via GitHub)" <gi...@apache.org> on 2026/09/13 02:59:15 UTC
[PR] KAFKA-21084: Avoid waking the fetch buffer when requeuing paused fetches [kafka]
chickenchickenlove opened a new pull request, #23443:
URL: https://github.com/apache/kafka/pull/23443
`AsyncKafkaConsumer.poll(Duration)` calls `FetchCollector.collectFetch()` through `pollForFetches()`.
`collectFetch()` preserves fetches from paused partitions by requeuing them through `FetchBuffer.addAll()`. However, `addAll()` also calls `wokenup.set(true)` and `blockingCondition.signalAll()`.
When only paused fetches remain, collection returns an empty result, and `pollForFetches()` calls `fetchBuffer.awaitWakeup(pollTimer)`. Since `wokenup.compareAndSet(true, false)` succeeds, the method returns immediately without calling `Condition.await()`.
Subsequent collections requeue the same paused fetches and regenerate the wakeup signal. This can cause the application thread to repeatedly execute the internal `poll()` loop until the timeout expires, consuming CPU without new responses or consumption progress.
This change introduces `requeue()`, which preserves the existing wakeup state, and uses it to restore paused fetches without generating unnecessary wakeup signals.
--
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-21084: Avoid waking the fetch buffer when requeuing paused fetches [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on PR #23443:
URL: https://github.com/apache/kafka/pull/23443#issuecomment-6146761330
@lianetm
I'm so sorry for the delayed commit 🙇♂️
I've addressed them based on your comments!
When you get a chance, please take another look!
--
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-21084: Avoid waking the fetch buffer when requeuing paused fetches [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #23443:
URL: https://github.com/apache/kafka/pull/23443#discussion_r4049434795
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchCollector.java:
##########
@@ -140,7 +140,7 @@ public Fetch<K, V> collectFetch(final FetchBuffer fetchBuffer) {
} finally {
// add any polled completed fetches for paused partitions back to the completed fetches queue to be
// re-evaluated in the next poll
- fetchBuffer.addAll(pausedCompletedFetches);
Review Comment:
after removing this seems that the `addAll` is only used from add(List.of()), so should we better remove it too?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchBuffer.java:
##########
@@ -113,6 +113,18 @@ void addAll(Collection<CompletedFetch> completedFetches) {
}
}
+ void requeue(Collection<CompletedFetch> completedFetches) {
Review Comment:
what about renaming this to `addNoWakeUp` or similar? (to align with the existing add and make the behaviour difference clear with the name itself). Otherwise I would say we need to add a java doc here to explain that this is just adding without wake up
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchBufferTest.java:
##########
@@ -186,6 +190,57 @@ public void testWakeup() throws Exception {
}
}
+ @Test
+ public void testRequeueDoesNotWakeUpBuffer() throws Exception {
+ try (FetchBuffer fetchBuffer = new FetchBuffer(logContext)) {
+ CompletedFetch completedFetch = completedFetch(topicAPartition0);
+ fetchBuffer.requeue(List.of(completedFetch));
+
+ FutureTask<Void> awaitWakeup = new FutureTask<>(() -> {
+ fetchBuffer.awaitWakeup(time.timer(Duration.ofMinutes(1)));
+ return null;
+ });
+ Thread waitingThread = new Thread(awaitWakeup);
+ waitingThread.start();
+ try {
+ TestUtils.waitForCondition(
+ () -> waitingThread.getState() == Thread.State.TIMED_WAITING,
+ "Thread did not start waiting on the fetch buffer"
Review Comment:
nit: this message is kind of confusing with the "start waiting". The message in the `assertFalse` below is the helpful one, but that one wouldn't show if we introduce a regression and wake up on requeue. SHould we hint here about the unexpected wakeup when the thread should have been waiting?
--
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-21084: Avoid waking the fetch buffer when requeuing paused fetches [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #23443:
URL: https://github.com/apache/kafka/pull/23443#discussion_r4259296674
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchCollector.java:
##########
@@ -140,7 +140,7 @@ public Fetch<K, V> collectFetch(final FetchBuffer fetchBuffer) {
} finally {
// add any polled completed fetches for paused partitions back to the completed fetches queue to be
// re-evaluated in the next poll
- fetchBuffer.addAll(pausedCompletedFetches);
Review Comment:
You are right.
I removed `addAll()`!
--
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-21084: Avoid waking the fetch buffer when requeuing paused fetches [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #23443:
URL: https://github.com/apache/kafka/pull/23443#discussion_r4259305284
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchBuffer.java:
##########
@@ -113,6 +113,18 @@ void addAll(Collection<CompletedFetch> completedFetches) {
}
}
+ void requeue(Collection<CompletedFetch> completedFetches) {
Review Comment:
You are right!
Changed name from `requeue` to `addAllNoWakeup`.
--
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