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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
