This is an automated email from the ASF dual-hosted git repository.
lianetm pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 8c0ca4ae35e KAFKA-18812: Improve consumer API errors upon background
thread failure (#22035)
8c0ca4ae35e is described below
commit 8c0ca4ae35e6cbcd3326ab43d4a0f28787d273f3
Author: nileshkumar3 <[email protected]>
AuthorDate: Tue Jul 7 08:53:54 2026 -0500
KAFKA-18812: Improve consumer API errors upon background thread failure
(#22035)
This PR makes two improvements:
**1. Fail-fast liveness check in `ApplicationEventHandler.add()`:**
Before enqueuing an event,
we now verify the background thread is still alive via
`Thread.isAlive()`. If the thread has
terminated, a `KafkaException` is thrown immediately with an actionable
message directing the
user to check earlier logs for the root cause. This prevents the
application thread from
silently enqueuing events that will never be processed, or blocking
indefinitely on `future.get()`.
**2. Event processing safety net in
`ConsumerNetworkThread.processApplicationEvents()`:** When
an exception happens in `ApplicationEventProcessor.process()`, the
outer catch block now completes
the event's `CompletableFuture` exceptionally with the actual error.
Previously, the error was
only logged and the future was left dangling — the caller would
eventually receive a generic
`TimeoutException` from the reaper instead of the real cause of the
failure.
Reviewers - @lianetm @kirktrue
Reviewers: Lianet Magrans <[email protected]>
---
.../consumer/internals/ConsumerNetworkThread.java | 5 ++++-
.../internals/events/ApplicationEventHandler.java | 22 +++++++++++++++++++
.../internals/ApplicationEventHandlerTest.java | 25 ++++++++++++++++++++++
.../internals/ConsumerNetworkThreadTest.java | 21 ++++++++++++++++++
4 files changed, 72 insertions(+), 1 deletion(-)
diff --git
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java
index b66857f2201..12bff6fa9ee 100644
---
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java
+++
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java
@@ -267,7 +267,10 @@ public class ConsumerNetworkThread extends KafkaThread
implements Closeable {
}
applicationEventProcessor.process(event);
} catch (Throwable t) {
- log.warn("Error processing event {}", t.getMessage(), t);
+ log.error("Error processing event {}", t.getMessage(), t);
+ if (event instanceof CompletableEvent) {
+ ((CompletableEvent<?>)
event).future().completeExceptionally(t);
+ }
}
}
asyncConsumerMetrics.recordApplicationEventQueueProcessingTime(time.milliseconds()
- startMs);
diff --git
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventHandler.java
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventHandler.java
index 7b6a7de773b..2f929d039f5 100644
---
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventHandler.java
+++
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventHandler.java
@@ -21,6 +21,7 @@ import
org.apache.kafka.clients.consumer.internals.ConsumerUtils;
import org.apache.kafka.clients.consumer.internals.NetworkClientDelegate;
import org.apache.kafka.clients.consumer.internals.RequestManagers;
import
org.apache.kafka.clients.consumer.internals.metrics.AsyncConsumerMetrics;
+import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.errors.InterruptException;
import org.apache.kafka.common.internals.IdempotentCloser;
import org.apache.kafka.common.utils.Time;
@@ -90,9 +91,11 @@ public class ApplicationEventHandler implements Closeable {
* to alert the network I/O thread that it has something to process.
*
* @param event An {@link ApplicationEvent} created by the application
thread
+ * @throws KafkaException if the consumer background thread is no longer
alive
*/
public void add(final ApplicationEvent event) {
Objects.requireNonNull(event, "ApplicationEvent provided to add must
be non-null");
+ ensureNetworkThreadAlive();
event.setEnqueuedMs(time.milliseconds());
// Record the updated queue size before actually adding the event to
the queue
// to avoid race conditions (the background thread is continuously
removing from this queue)
@@ -155,4 +158,23 @@ public class ApplicationEventHandler implements Closeable {
() -> log.warn("The application event handler was already
closed")
);
}
+
+ /**
+ * Best-effort check that the consumer network thread is still alive. If
the thread has
+ * already terminated (due to a failure or shutdown), it will never
process any events from
+ * the queue. Rather than blocking indefinitely or timing out with a
misleading error, this
+ * fails fast with a clear error message.
+ *
+ * <p>Note: this is inherently racy — the thread could die between this
check and the
+ * subsequent {@code applicationEventQueue.add()}. That narrow window is
acceptable because
+ * any subsequent call to {@code add()} will detect the dead thread
immediately.
+ *
+ * @throws KafkaException if the background thread is not alive
+ */
+ private void ensureNetworkThreadAlive() {
+ if (networkThread == null || !networkThread.isAlive()) {
+ throw new KafkaException(
+ "The consumer background thread is not running and cannot
process requests.");
+ }
+ }
}
diff --git
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ApplicationEventHandlerTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ApplicationEventHandlerTest.java
index 8ff686cf588..732dae84dd8 100644
---
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ApplicationEventHandlerTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ApplicationEventHandlerTest.java
@@ -34,12 +34,14 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.MethodSource;
+import java.time.Duration;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.function.Supplier;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
@@ -105,6 +107,29 @@ public class ApplicationEventHandlerTest {
assertInitializeResourcesError(InterruptException.class, () ->
networkClientDelegate);
}
+ @Test
+ public void testAddThrowsWhenBackgroundThreadDead() {
+ try (Metrics metrics = new Metrics();
+ AsyncConsumerMetrics asyncConsumerMetrics = spy(new
AsyncConsumerMetrics(metrics, "test-group"));
+ ApplicationEventHandler handler = new ApplicationEventHandler(
+ new LogContext(),
+ time,
+ initializationTimeoutMs,
+ applicationEventsQueue,
+ applicationEventReaper,
+ () -> applicationEventProcessor,
+ () -> networkClientDelegate,
+ () -> requestManagers,
+ asyncConsumerMetrics
+ )) {
+ handler.close(Duration.ZERO);
+
+ KafkaException error = assertThrows(KafkaException.class,
+ () -> handler.add(new AsyncPollEvent(time.milliseconds() + 10,
time.milliseconds())));
+ assertTrue(error.getMessage().contains("background thread is not
running"));
+ }
+ }
+
private <T extends Throwable> T assertInitializeResourcesError(Class<T>
exceptionClass,
Supplier<NetworkClientDelegate> networkClientDelegateSupplier) {
try (Metrics metrics = new Metrics();
diff --git
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThreadTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThreadTest.java
index bb80d5ac381..d027a790dda 100644
---
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThreadTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThreadTest.java
@@ -20,6 +20,7 @@ import
org.apache.kafka.clients.consumer.internals.events.ApplicationEvent;
import
org.apache.kafka.clients.consumer.internals.events.ApplicationEventProcessor;
import org.apache.kafka.clients.consumer.internals.events.AsyncPollEvent;
import
org.apache.kafka.clients.consumer.internals.events.CompletableEventReaper;
+import org.apache.kafka.clients.consumer.internals.events.PausePartitionsEvent;
import
org.apache.kafka.clients.consumer.internals.metrics.AsyncConsumerMetrics;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.metrics.Metrics;
@@ -37,6 +38,7 @@ import org.junit.jupiter.params.provider.MethodSource;
import org.junit.jupiter.params.provider.ValueSource;
import java.time.Duration;
+import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
@@ -52,6 +54,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -316,6 +319,24 @@ public class ConsumerNetworkThreadTest {
testInitializeResourcesError(networkClientDelegateSupplier,
requestManagersSupplier);
}
+ @Test
+ public void testProcessEventFailureCompletesFutureExceptionally() {
+ RuntimeException processingError = new RuntimeException("Simulated
processing failure");
+
doThrow(processingError).when(applicationEventProcessor).process(any(ApplicationEvent.class));
+
+ PausePartitionsEvent event = new
PausePartitionsEvent(Collections.emptyList(), time.milliseconds() + 1000);
+ event.setEnqueuedMs(time.milliseconds());
+ applicationEventQueue.add(event);
+
+ consumerNetworkThread.runOnce();
+
+ assertTrue(event.future().isDone(), "Event future should be completed
after processing failure");
+ assertTrue(event.future().isCompletedExceptionally(), "Event future
should be completed exceptionally");
+
+ KafkaException thrown = assertThrows(KafkaException.class, () ->
ConsumerUtils.getResult(event.future()));
+ assertEquals(processingError, thrown.getCause());
+ }
+
/**
* Tests that when an error occurs during {@link
ConsumerNetworkThread#initializeResources()} that the
* logic in {@link ConsumerNetworkThread#cleanup()} will not throw errors
when closing.
