Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-3885: Fix sleep loop for retry batch

Browse Source 
Fixes: #3885
PR link: https://github.com/spring-projects/spring-kafka/pull/3885

The `ListenerUtils.conditionalSleep()` has a loop.
And if the whole interval is longer than consumer poll interval, the consumer is going to be dropped from the group.

* Add a new `ListenerUtils.conditionalSleepWithPoll()` with a `consumer.poll(Duration.ZERO)` in between sleep loop iterations.

Signed-off-by: Sanghyeok An <ojt90902@naver.com>

Signed-off-by: Artem Bilan <artem.bilan@broadcom.com>
(cherry picked from commit dd30085f0b)
This commit is contained in:
ChickenchickenLove
2025-05-07 23:13:51 +09:00
committed by Spring Builds
parent 36e16a4c60
commit 291b4f0467
2 changed files with 43 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlingUtils.java
View File

@@ -50,6 +50,7 @@ import org.springframework.util.backoff.BackOffExecution;
* @author Andrii Pelesh
* @author Antonio Tomac
* @author Wang Zhiyang
* @author Sanghyeok An
*
* @since 2.8
*
@@ -112,13 +113,18 @@ public final class ErrorHandlingUtils {
throw new KafkaException("Woken up during retry", logLevel, we);
}
try {
ListenerUtils.conditionalSleep(
ListenerUtils.conditionalSleepWithPoll(
() -> container.isRunning() &&
!container.isPauseRequested() &&
records.partitions().stream().noneMatch(container::isPartitionPauseRequested),
nextBackOff
nextBackOff,
consumer
);
}
catch (WakeupException we) {
seeker.handleBatch(thrownException, records, consumer, container, NO_OP);
throw new KafkaException("Woken up during retry", logLevel, we);
}
catch (InterruptedException e1) {
Thread.currentThread().interrupt();
seeker.handleBatch(thrownException, records, consumer, container, NO_OP);

36
spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerUtils.java
View File

@@ -16,9 +16,11 @@
package org.springframework.kafka.listener;
import java.time.Duration;
import java.util.Map;
import java.util.function.Supplier;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.springframework.util.Assert;
@@ -32,6 +34,7 @@ import org.springframework.util.backoff.BackOffExecution;
* @author Francois Rosiere
* @author Antonio Tomac
* @author Wang Zhiyang
* @author Sanghyeok An
* @since 2.0
*
*/
@@ -181,6 +184,38 @@ public final class ListenerUtils {
while (System.currentTimeMillis() < timeout);
}
/**
* Sleep for the desired timeout, as long as shouldSleepCondition supplies true.
* This method requires that the consumer is paused; otherwise, ConsumerRecord may be lost.
* Periodically calls {@code Consumer.poll(Duration.ZERO)} to prevent a paused consumer from being rebalanced.
* @param shouldSleepCondition to.
* @param interval the timeout.
* @param consumer the kafka consumer to call poll().
* @throws InterruptedException if the thread is interrupted.
*/
public static void conditionalSleepWithPoll(Supplier<Boolean> shouldSleepCondition,
long interval,
Consumer<?, ?> consumer) throws InterruptedException {
boolean isFirst = true;
long timeout = System.currentTimeMillis() + interval;
long sleepInterval = interval > SMALL_INTERVAL_THRESHOLD ? DEFAULT_SLEEP_INTERVAL : SMALL_SLEEP_INTERVAL;
do {
Thread.sleep(sleepInterval);
if (!shouldSleepCondition.get()) {
break;
}
if (isFirst) {
isFirst = false;
}
else {
// To prevent consumer group rebalancing during retry backoff.
consumer.poll(Duration.ZERO);
}
}
while (System.currentTimeMillis() < timeout);
}
/**
* Create a new {@link OffsetAndMetadata} using the given container and offset.
* @param container a container.
@@ -198,4 +233,3 @@ public final class ListenerUtils {
return new OffsetAndMetadata(offset);
}
}