You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mj...@apache.org on 2026/08/12 01:02:46 UTC
(kafka) branch 4.4 updated: KAFKA-20721: Removed timeout on task removal from state updater. (#23093)
This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch 4.4
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.4 by this push:
new 43caa4ce412 KAFKA-20721: Removed timeout on task removal from state updater. (#23093)
43caa4ce412 is described below
commit 43caa4ce412e7f7b2958b3c4b9cc485bb560d463
Author: Nikita Shupletsov <ni...@shupletsov.ca>
AuthorDate: Tue Aug 11 17:56:29 2026 -0700
KAFKA-20721: Removed timeout on task removal from state updater. (#23093)
- Added logic to catch and propagate to the stream thread fatal state
update errors.
- Improved the error handling logic in the state updater thread. Now
when the thread is failing it marks all tasks as failed.
- Added logic to fail task removal futures if the thread is dead.
Reviewers: Matthias J. Sax <ma...@confluent.io>
---
.../StateUpdaterFailureIntegrationTest.java | 86 +++++++++---
.../processor/internals/DefaultStateUpdater.java | 137 ++++++++++++++-----
.../streams/processor/internals/StateUpdater.java | 17 +++
.../streams/processor/internals/TaskManager.java | 34 +++--
.../internals/DefaultStateUpdaterTest.java | 146 +++++++++++++++++++--
.../processor/internals/TaskManagerTest.java | 47 ++++++-
6 files changed, 397 insertions(+), 70 deletions(-)
diff --git a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/StateUpdaterFailureIntegrationTest.java b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/StateUpdaterFailureIntegrationTest.java
index 16cb72e3a3b..a0dd4a85a20 100644
--- a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/StateUpdaterFailureIntegrationTest.java
+++ b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/StateUpdaterFailureIntegrationTest.java
@@ -17,6 +17,7 @@
package org.apache.kafka.streams.integration;
import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.streams.KafkaStreams;
@@ -24,6 +25,7 @@ import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.TopologyWrapper;
import org.apache.kafka.streams.errors.ProcessorStateException;
import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
+import org.apache.kafka.streams.processor.StateRestoreListener;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.StateStoreContext;
import org.apache.kafka.streams.state.KeyValueStore;
@@ -35,16 +37,22 @@ import org.apache.kafka.test.TestUtils;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Named;
import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import java.io.IOException;
import java.time.Duration;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Stream;
import static org.apache.kafka.streams.utils.TestUtils.safeUniqueTestName;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class StateUpdaterFailureIntegrationTest {
@@ -56,7 +64,6 @@ public class StateUpdaterFailureIntegrationTest {
private final EmbeddedKafkaCluster cluster = new EmbeddedKafkaCluster(NUM_BROKERS);
private Properties streamsConfiguration;
- private final MockTime mockTime = cluster.time;
private KafkaStreams streams;
@BeforeEach
@@ -85,20 +92,34 @@ public class StateUpdaterFailureIntegrationTest {
}
}
+ private static Stream<Arguments> failuresOfTasksInTheStateUpdater() {
+ return Stream.of(
+ Arguments.of(Named.of("exception", (Runnable) () -> {
+ throw new ProcessorStateException("restore end");
+ })),
+ Arguments.of(Named.of("error", (Runnable) () -> {
+ throw new Error("restore end");
+ }))
+ );
+ }
+
/**
* The conditions that we need to meet:
* <p><ul>
* <li>We have an unhandled task in {@link org.apache.kafka.streams.processor.internals.DefaultStateUpdater}</li>
* <li>StreamThread is not running, so {@link org.apache.kafka.streams.processor.internals.TaskManager#handleExceptionsFromStateUpdater} is not called anymore</li>
- * <li>The task throws exception in {@link org.apache.kafka.streams.processor.internals.Task#maybeCheckpoint()} while being processed by {@code DefaultStateUpdater}</li>
- * <li>{@link org.apache.kafka.streams.processor.internals.TaskManager#shutdownStateUpdater} tries to clean up all tasks that are left in the {@code DefaultStateUpdater}</li>
+ * <li>The task fails while being processed by {@code DefaultStateUpdater}, either with an exception that the {@code DefaultStateUpdater} handles or with an error that it does not handle and that stops its thread</li>
+ * <li>{@link org.apache.kafka.streams.processor.internals.TaskManager#shutdownStateUpdater} tries to clean up all tasks that are left in the {@code DefaultStateUpdater} and waits for the removals of those tasks</li>
* </ul><p>
* If all conditions are met, {@code TaskManager} needs to be able to handle the failed task from the {@code DefaultStateUpdater} correctly and not hang.
*/
- @Test
- public void correctlyHandleFlushErrorsDuringRebalance() throws Exception {
+ @ParameterizedTest
+ @MethodSource("failuresOfTasksInTheStateUpdater")
+ public void correctlyHandleFailuresDuringRebalance(final Runnable failure) throws Exception {
final AtomicInteger numberOfStoreInits = new AtomicInteger();
+ final AtomicInteger numberOfStoreCloses = new AtomicInteger();
final CountDownLatch pendingShutdownLatch = new CountDownLatch(1);
+ final AtomicBoolean failureInjected = new AtomicBoolean(false);
final StoreBuilder<KeyValueStore<Object, Object>> storeBuilder = new AbstractStoreBuilder<>("testStateStore", Serdes.Integer(), Serdes.ByteArray(), new MockTime()) {
@@ -113,17 +134,9 @@ public class StateUpdaterFailureIntegrationTest {
}
@Override
- public void flush() {
- // we want to throw the ProcessorStateException here only when the rebalance finished(we reassigned the 3 tasks from the removed thread to the existing thread)
- // we use waitForCondition to wait until the current state is PENDING_SHUTDOWN to make sure the Stream Thread will not handle the exception and we can get to in TaskManager#shutdownStateUpdater
- if (numberOfStoreInits.get() == 9) {
- try {
- pendingShutdownLatch.await();
- } catch (final InterruptedException e) {
- throw new RuntimeException(e);
- }
- throw new ProcessorStateException("flush");
- }
+ public void close() {
+ super.close();
+ numberOfStoreCloses.incrementAndGet();
}
};
}
@@ -134,7 +147,42 @@ public class StateUpdaterFailureIntegrationTest {
topology.addProcessor("my-processor", new MockApiProcessorSupplier<>(), "ingest");
topology.addStateStore(storeBuilder, "my-processor");
+ final StateRestoreListener stateRestoreListener = new StateRestoreListener() {
+
+ @Override
+ public void onRestoreStart(final TopicPartition topicPartition,
+ final String storeName,
+ final long startingOffset,
+ final long endingOffset) {
+ }
+
+ @Override
+ public void onBatchRestored(final TopicPartition topicPartition,
+ final String storeName,
+ final long batchEndOffset,
+ final long numRestored) {
+ }
+
+ @Override
+ public void onRestoreEnd(final TopicPartition topicPartition,
+ final String storeName,
+ final long totalRestored) {
+ // we want to fail here only when the rebalance finished(we reassigned the 3 tasks from the removed thread to the existing thread)
+ // we wait until the current state is PENDING_SHUTDOWN to make sure the Stream Thread will not handle the failure and we can get to in TaskManager#shutdownStateUpdater
+ if (numberOfStoreInits.get() == 9) {
+ try {
+ pendingShutdownLatch.await();
+ } catch (final InterruptedException interruptedException) {
+ throw new RuntimeException(interruptedException);
+ }
+ failureInjected.set(true);
+ failure.run();
+ }
+ }
+ };
+
streams = new KafkaStreams(topology, streamsConfiguration);
+ streams.setGlobalStateRestoreListener(stateRestoreListener);
streams.setStateListener((newState, oldState) -> {
if (newState == KafkaStreams.State.PENDING_SHUTDOWN) {
pendingShutdownLatch.countDown();
@@ -150,5 +198,9 @@ public class StateUpdaterFailureIntegrationTest {
TestUtils.waitForCondition(() -> numberOfStoreInits.get() == 9, "Streams never reinitialized the store enough times");
assertTrue(streams.close(Duration.ofSeconds(60)));
+
+ assertEquals(numberOfStoreInits.get(), numberOfStoreCloses.get(), "Not all state stores were closed");
+ // Verifying that the failure was actually injected to avoid situations when the test passes only because nothing was injected.
+ assertTrue(failureInjected.get(), "The failure was never injected into the state updater");
}
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java
index 68e86449bda..dd138eea3e7 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java
@@ -56,6 +56,7 @@ import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
@@ -79,6 +80,7 @@ import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetric
public class DefaultStateUpdater implements StateUpdater {
+ private static final String NOT_RUNNING_ERROR_MESSAGE = "The state updater is not running.";
private static final String BUG_ERROR_MESSAGE = "This indicates a bug. " +
"Please report at https://issues.apache.org/jira/projects/KAFKA/issues or to the dev-mailing list (https://kafka.apache.org/contact).";
@@ -167,19 +169,30 @@ public class DefaultStateUpdater implements StateUpdater {
while (isRunning.get()) {
runOnce();
}
- } catch (final RuntimeException anyOtherException) {
- handleRuntimeException(anyOtherException);
+ } catch (final Throwable anyOtherThrowable) {
+ handleFatalThrowable(anyOtherThrowable);
} finally {
- clearInputQueue();
- clearUpdatingAndPausedTasks();
+ // report the tasks before the pending actions, so that a task is already reported as failed when the
+ // caller of a failed action looks for it
+ failRemainingTasks();
+ failPendingActions();
updaterMetrics.clear();
log.info("State updater thread stopped");
}
}
- private void clearInputQueue() {
+ // Once this thread stopped, nobody will ever perform the actions that are still in the input queue. Hence, we
+ // report them as failed instead of silently dropping them, so that the stream thread does not wait for a
+ // result that will never come and does not lose track of the tasks in the input queue.
+ private void failPendingActions() {
tasksAndActionsLock.lock();
try {
+ // The flag is published under the lock that guards the input queue. This ensures that an action that is
+ // added concurrently is either failed here or immediately failed by add() or remove().
+ threadStopped.set(true);
+ for (final TaskAndAction taskAndAction : tasksAndActions) {
+ failAction(taskAndAction);
+ }
tasksAndActions.clear();
} finally {
tasksAndActionsLock.unlock();
@@ -223,7 +236,8 @@ public class DefaultStateUpdater implements StateUpdater {
private void performActionsOnTasks() {
tasksAndActionsLock.lock();
try {
- for (final TaskAndAction taskAndAction : tasksAndActions()) {
+ while (!tasksAndActions.isEmpty()) {
+ final TaskAndAction taskAndAction = tasksAndActions.peek();
final Action action = taskAndAction.action();
switch (action) {
case ADD:
@@ -235,6 +249,7 @@ public class DefaultStateUpdater implements StateUpdater {
default:
throw new IllegalStateException("Unknown action type " + action);
}
+ tasksAndActions.poll();
}
} finally {
tasksAndActionsLock.unlock();
@@ -338,9 +353,18 @@ public class DefaultStateUpdater implements StateUpdater {
}
- private void handleRuntimeException(final RuntimeException runtimeException) {
- log.error("An unexpected error occurred within the state updater thread: {}", String.valueOf(runtimeException));
- addToExceptionsAndFailedTasksThenClearUpdatingAndPausedTasks(runtimeException);
+ // Any throwable that escapes the update loop stops this thread. Even for a throwable that is not recoverable,
+ // like an error, we report the tasks of this thread as failed, so that the stream thread notices that this
+ // thread is gone instead of silently losing its tasks.
+ private void handleFatalThrowable(final Throwable throwable) {
+ log.error("An unexpected error occurred within the state updater thread.", throwable);
+ final RuntimeException exception = throwable instanceof RuntimeException
+ ? (RuntimeException) throwable
+ : new StreamsException("The state updater thread failed with a fatal error.", throwable);
+ // publish the exception before reporting the tasks as failed, so that the stream thread never sees a task
+ // that failed because this thread died without also seeing why this thread died
+ fatalException.set(exception);
+ addToExceptionsAndFailedTasksThenClearUpdatingAndPausedTasks(exception);
isRunning.set(false);
}
@@ -511,18 +535,36 @@ public class DefaultStateUpdater implements StateUpdater {
!isTopologyResumed.get();
}
- private void clearUpdatingAndPausedTasks() {
- updatingTasks.clear();
- pausedTasks.clear();
+ // Once this thread stopped, it cannot update or hand over any of the tasks it still owns. Hence, we report all
+ // of them as failed instead of silently dropping them, so that the stream thread closes them
+ private void failRemainingTasks() {
+ if (!updatingTasks.isEmpty() || !pausedTasks.isEmpty()) {
+ addToExceptionsAndFailedTasksThenClearUpdatingAndPausedTasks(
+ new StreamsException(NOT_RUNNING_ERROR_MESSAGE)
+ );
+ }
+ restoredActiveTasksLock.lock();
+ try {
+ exceptionsAndFailedTasksLock.lock();
+ try {
+ for (final StreamTask restoredTask : restoredActiveTasks) {
+ log.warn("Restored task {} was not handed over to the stream thread before the state updater "
+ + "stopped. The task is reported as failed.", restoredTask.id());
+ exceptionsAndFailedTasks.add(new ExceptionAndTask(
+ new StreamsException(NOT_RUNNING_ERROR_MESSAGE, restoredTask.id()),
+ restoredTask
+ ));
+ }
+ } finally {
+ exceptionsAndFailedTasksLock.unlock();
+ }
+ restoredActiveTasks.clear();
+ } finally {
+ restoredActiveTasksLock.unlock();
+ }
changelogReader.clear();
}
- private List<TaskAndAction> tasksAndActions() {
- final List<TaskAndAction> tasksAndActionsToProcess = new ArrayList<>(tasksAndActions);
- tasksAndActions.clear();
- return tasksAndActionsToProcess;
- }
-
private void addTask(final Task task) {
final TaskId taskId = task.id();
@@ -580,9 +622,6 @@ public class DefaultStateUpdater implements StateUpdater {
} catch (final StreamsException streamsException) {
handleStreamsExceptionWithTask(streamsException, taskId);
future.completeExceptionally(streamsException);
- } catch (final RuntimeException runtimeException) {
- handleRuntimeException(runtimeException);
- future.completeExceptionally(runtimeException);
}
}
@@ -866,6 +905,9 @@ public class DefaultStateUpdater implements StateUpdater {
private StateUpdaterThread stateUpdaterThread = null;
+ private final AtomicBoolean threadStopped = new AtomicBoolean(false);
+ private final AtomicReference<RuntimeException> fatalException = new AtomicReference<>();
+
public DefaultStateUpdater(final String name,
final StreamsMetricsImpl metrics,
final StreamsConfig config,
@@ -892,6 +934,10 @@ public class DefaultStateUpdater implements StateUpdater {
throw new IllegalStateException("State updater started with non-empty output queues. "
+ BUG_ERROR_MESSAGE);
}
+ if (threadStopped.get()) {
+ throw new IllegalStateException("State updater cannot be started again after it was shut down. "
+ + BUG_ERROR_MESSAGE);
+ }
stateUpdaterThread = new StateUpdaterThread(name, metrics, changelogReader);
stateUpdaterThread.start();
@@ -934,13 +980,7 @@ public class DefaultStateUpdater implements StateUpdater {
public void add(final Task task) {
verifyStateFor(task);
- tasksAndActionsLock.lock();
- try {
- tasksAndActions.add(TaskAndAction.createAddTask(task));
- tasksAndActionsCondition.signalAll();
- } finally {
- tasksAndActionsLock.unlock();
- }
+ submit(TaskAndAction.createAddTask(task));
}
private void verifyStateFor(final Task task) {
@@ -956,14 +996,44 @@ public class DefaultStateUpdater implements StateUpdater {
public CompletableFuture<RemovedTaskResult> remove(final TaskId taskId,
final StandbyUpdateListener.SuspendReason suspendReason) {
final CompletableFuture<RemovedTaskResult> future = new CompletableFuture<>();
+ submit(TaskAndAction.createRemoveTask(taskId, future, suspendReason));
+ return future;
+ }
+
+ private void submit(final TaskAndAction taskAndAction) {
tasksAndActionsLock.lock();
try {
- tasksAndActions.add(TaskAndAction.createRemoveTask(taskId, future, suspendReason));
- tasksAndActionsCondition.signalAll();
+ if (threadStopped.get()) {
+ // The state updater thread will never pick up this action, so we fail it right away instead of
+ // letting the caller wait for a result that will never come.
+ failAction(taskAndAction);
+ } else {
+ tasksAndActions.add(taskAndAction);
+ tasksAndActionsCondition.signalAll();
+ }
} finally {
tasksAndActionsLock.unlock();
}
- return future;
+ }
+
+ private void failAction(final TaskAndAction taskAndAction) {
+ final StreamsException exception = new StreamsException(NOT_RUNNING_ERROR_MESSAGE);
+ if (taskAndAction.action() == Action.REMOVE) {
+ log.warn("Task {} cannot be removed because the state updater thread is not running.",
+ taskAndAction.taskId());
+ taskAndAction.futureForRemove().completeExceptionally(exception);
+ } else {
+ final Task task = taskAndAction.task();
+ log.warn("Task {} cannot be updated because the state updater thread is not running. The task is "
+ + "reported as failed.", task.id());
+ final ExceptionAndTask exceptionAndTask = new ExceptionAndTask(exception, task);
+ exceptionsAndFailedTasksLock.lock();
+ try {
+ exceptionsAndFailedTasks.add(exceptionAndTask);
+ } finally {
+ exceptionsAndFailedTasksLock.unlock();
+ }
+ }
}
@Override
@@ -1033,6 +1103,11 @@ public class DefaultStateUpdater implements StateUpdater {
}
}
+ @Override
+ public Optional<RuntimeException> fatalException() {
+ return Optional.ofNullable(fatalException.get());
+ }
+
public Set<StandbyTask> updatingStandbyTasks() {
return stateUpdaterThread != null
? Set.copyOf(stateUpdaterThread.updatingStandbyTasks())
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateUpdater.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateUpdater.java
index 37ccdea199a..2456395b056 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateUpdater.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateUpdater.java
@@ -135,6 +135,9 @@ public interface StateUpdater {
*
* This method does not block until the task is added to the state updater.
*
+ * If the state updater is not able to update tasks anymore, the task is immediately reported as failed and can be
+ * retrieved with {@link StateUpdater#drainExceptionsAndFailedTasks()}.
+ *
* @param task task to add
*/
void add(final Task task);
@@ -146,6 +149,9 @@ public interface StateUpdater {
* which processing can be blocked. The task to remove is removed from the updating tasks, paused tasks,
* restored tasks, or failed tasks.
*
+ * If the state updater is not able to remove tasks anymore, the returned future is completed exceptionally
+ * instead of never being completed.
+ *
* @param taskId ID of the task to remove
* @param suspendReason the reason for suspending standby update, passed through to the changelog reader
*/
@@ -186,6 +192,17 @@ public interface StateUpdater {
*/
boolean hasExceptionsAndFailedTasks();
+ /**
+ * Gets the exception that made the state updater die.
+ *
+ * A state updater that died cannot update any task anymore, neither the tasks it owned when it died nor tasks that
+ * are added afterwards. Hence, the application cannot continue and must be failed with this exception.
+ *
+ * @return the exception that made the state updater die, or {@link Optional#empty()} if the state updater is
+ * running or was shut down regularly
+ */
+ Optional<RuntimeException> fatalException();
+
/**
* Gets all tasks that are managed by the state updater.
*
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
index e32051c16da..8f9c681fbc6 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
@@ -60,6 +60,7 @@ import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.CompletableFuture;
@@ -79,6 +80,7 @@ public class TaskManager {
private static final String BUG_ERROR_MESSAGE = "This indicates a bug. " +
"Please report at https://issues.apache.org/jira/projects/KAFKA/issues or to the dev-mailing list (https://kafka.apache.org/contact).";
private static final String INTERRUPTED_ERROR_MESSAGE = "Thread got interrupted. " + BUG_ERROR_MESSAGE;
+ private static final long REMOVAL_LOG_INTERVAL_MINUTES = 1L;
// initialize the task list
// activeTasks needs to be concurrent as it can be accessed
@@ -716,14 +718,22 @@ public class TaskManager {
private StateUpdater.RemovedTaskResult waitForFuture(final TaskId taskId,
final CompletableFuture<StateUpdater.RemovedTaskResult> future) {
- final StateUpdater.RemovedTaskResult removedTaskResult;
try {
- removedTaskResult = future.get(5, TimeUnit.MINUTES);
- if (removedTaskResult == null) {
- throw new IllegalStateException("Task " + taskId + " was not found in the state updater. "
- + BUG_ERROR_MESSAGE);
+ long minutesWaited = 0L;
+ while (true) {
+ try {
+ final StateUpdater.RemovedTaskResult removedTaskResult =
+ future.get(REMOVAL_LOG_INTERVAL_MINUTES, TimeUnit.MINUTES);
+ if (removedTaskResult == null) {
+ throw new IllegalStateException("Task " + taskId + " was not found in the state updater. "
+ + BUG_ERROR_MESSAGE);
+ }
+ return removedTaskResult;
+ } catch (final java.util.concurrent.TimeoutException retryTimeout) {
+ minutesWaited += REMOVAL_LOG_INTERVAL_MINUTES;
+ log.warn("Waiting for the removal of task {} from the state updater for {} minute(s).", taskId, minutesWaited);
+ }
}
- return removedTaskResult;
} catch (final ExecutionException executionException) {
log.warn("An exception happened when removing task {} from the state updater. The task was added to the " +
"failed task in the state updater: ",
@@ -733,10 +743,6 @@ public class TaskManager {
Thread.currentThread().interrupt();
log.error(INTERRUPTED_ERROR_MESSAGE, shouldNotHappen);
throw new IllegalStateException(INTERRUPTED_ERROR_MESSAGE, shouldNotHappen);
- } catch (final java.util.concurrent.TimeoutException timeoutException) {
- log.warn("The state updater wasn't able to remove task {} in time. The state updater thread may be dead. "
- + BUG_ERROR_MESSAGE, taskId, timeoutException);
- return null;
}
}
@@ -858,6 +864,7 @@ public class TaskManager {
public boolean checkStateUpdater(final long now,
final java.util.function.Consumer<Set<TopicPartition>> offsetResetter) {
+ maybeThrowFatalExceptionFromStateUpdater();
addTasksToStateUpdater();
if (stateUpdater.hasExceptionsAndFailedTasks()) {
handleExceptionsFromStateUpdater();
@@ -985,6 +992,13 @@ public class TaskManager {
}
}
+ private void maybeThrowFatalExceptionFromStateUpdater() {
+ final Optional<RuntimeException> fatalException = stateUpdater.fatalException();
+ if (fatalException.isPresent()) {
+ throw new StreamsException("The state updater died and cannot update tasks anymore.", fatalException.get());
+ }
+ }
+
public void handleExceptionsFromStateUpdater() {
final Map<TaskId, RuntimeException> taskExceptions = collectExceptionsAndFailedTasksFromStateUpdater();
maybeThrowTaskExceptions(taskExceptions);
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java
index 2f6db345195..4f74c5b8c6d 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java
@@ -47,6 +47,7 @@ import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
@@ -155,8 +156,8 @@ class DefaultStateUpdaterTest {
stateUpdater.shutdown(Duration.ofMinutes(1));
- verifyRestoredActiveTasks(statelessTask, restoredStatefulTask);
- verifyExceptionsAndFailedTasks(new ExceptionAndTask(taskCorruptedException, failedStatefulTask));
+ verifyRestoredActiveTasks();
+ verifyFailedTasks(StreamsException.class, statelessTask, restoredStatefulTask, failedStatefulTask, standbyTask);
verifyUpdatingTasks();
verifyPausedTasks();
verify(changelogReader).clear();
@@ -178,23 +179,24 @@ class DefaultStateUpdaterTest {
stateUpdater.shutdown(Duration.ofMinutes(1));
verifyRestoredActiveTasks();
- verifyExceptionsAndFailedTasks();
+ verifyFailedTasks(StreamsException.class, statefulTask, standbyTask);
verifyUpdatingTasks();
verifyPausedTasks();
verify(changelogReader).clear();
}
@Test
- public void shouldShutdownStateUpdaterAndRestart() {
+ public void shouldThrowIfStartedAfterShutdown() {
stateUpdater.start();
stateUpdater.shutdown(Duration.ofMinutes(1));
- stateUpdater.start();
-
- stateUpdater.shutdown(Duration.ofMinutes(1));
+ final IllegalStateException exception = assertThrows(IllegalStateException.class, stateUpdater::start);
- verify(changelogReader, times(2)).clear();
+ assertEquals("State updater cannot be started again after it was shut down."
+ + " This indicates a bug. Please report at https://issues.apache.org/jira/projects/KAFKA/issues or to the"
+ + " dev-mailing list (https://kafka.apache.org/contact).", exception.getMessage());
+ verify(changelogReader).clear();
}
@Test
@@ -945,6 +947,134 @@ class DefaultStateUpdaterTest {
assertEquals(shouldStillBeRunning, stateUpdater.isRunning());
}
+ @Test
+ public void shouldFailPendingRemovalsWhenStateUpdaterThreadDies() throws Exception {
+ final StreamTask taskWithFailingRemoval = statefulTask(TASK_0_0, Set.of(TOPIC_PARTITION_A_0)).inState(State.RESTORING).build();
+ final StreamTask otherTask = statefulTask(TASK_1_0, Set.of(TOPIC_PARTITION_B_0)).inState(State.RESTORING).build();
+ final RuntimeException fatalException = new RuntimeException("Something happened");
+ when(changelogReader.completedChangelogs()).thenReturn(Collections.emptySet());
+ when(changelogReader.allChangelogsCompleted()).thenReturn(false);
+ final Collection<TopicPartition> changelogPartitions = taskWithFailingRemoval.changelogPartitions();
+ doThrow(fatalException).when(changelogReader)
+ .unregister(changelogPartitions, StandbyUpdateListener.SuspendReason.MIGRATED);
+ stateUpdater.start();
+ stateUpdater.add(taskWithFailingRemoval);
+ stateUpdater.add(otherTask);
+ verifyUpdatingTasks(taskWithFailingRemoval, otherTask);
+
+ final CompletableFuture<StateUpdater.RemovedTaskResult> futureOfFailingRemoval =
+ stateUpdater.remove(taskWithFailingRemoval.id(), StandbyUpdateListener.SuspendReason.MIGRATED);
+ final CompletableFuture<StateUpdater.RemovedTaskResult> futureOfOtherTask =
+ stateUpdater.remove(otherTask.id(), StandbyUpdateListener.SuspendReason.MIGRATED);
+
+ assertInstanceOf(
+ StreamsException.class,
+ assertThrows(ExecutionException.class, futureOfFailingRemoval::get).getCause()
+ );
+ assertInstanceOf(
+ StreamsException.class,
+ assertThrows(ExecutionException.class, futureOfOtherTask::get).getCause()
+ );
+ verifyExceptionsAndFailedTasks(
+ new ExceptionAndTask(fatalException, taskWithFailingRemoval),
+ new ExceptionAndTask(fatalException, otherTask)
+ );
+ // the exception that made the thread die is reported separately, so that the stream thread can fail the
+ // application even if the thread did not own any task
+ assertEquals(Optional.of(fatalException), stateUpdater.fatalException());
+ assertFalse(stateUpdater.isRunning());
+ }
+
+ @Test
+ public void shouldFailRemovalOfTaskWhenStateUpdaterThreadDiesWhilePerformingIt() throws Exception {
+ final StreamTask task = statefulTask(TASK_0_0, Set.of(TOPIC_PARTITION_A_0)).inState(State.RESTORING).build();
+ when(changelogReader.completedChangelogs()).thenReturn(Collections.emptySet());
+ when(changelogReader.allChangelogsCompleted()).thenReturn(false);
+ final Collection<TopicPartition> changelogPartitions = task.changelogPartitions();
+ doThrow(new Error("Something happened")).when(changelogReader)
+ .unregister(changelogPartitions, StandbyUpdateListener.SuspendReason.MIGRATED);
+ stateUpdater.start();
+ stateUpdater.add(task);
+ verifyUpdatingTasks(task);
+
+ final CompletableFuture<StateUpdater.RemovedTaskResult> future =
+ stateUpdater.remove(task.id(), StandbyUpdateListener.SuspendReason.MIGRATED);
+
+ assertInstanceOf(StreamsException.class, assertThrows(ExecutionException.class, future::get).getCause());
+ verifyFailedTasks(StreamsException.class, task);
+ assertFalse(stateUpdater.isRunning());
+ }
+
+ @Test
+ public void shouldFailRemovalOfTaskWhenStateUpdaterThreadIsNotRunningAnymore() throws Exception {
+ killStateUpdaterThread();
+
+ final CompletableFuture<StateUpdater.RemovedTaskResult> future =
+ stateUpdater.remove(TASK_0_1, StandbyUpdateListener.SuspendReason.MIGRATED);
+
+ assertInstanceOf(StreamsException.class, assertThrows(ExecutionException.class, future::get).getCause());
+ }
+
+ @Test
+ public void shouldFailRemovalOfTaskAfterStateUpdaterWasShutDown() {
+ stateUpdater.start();
+ stateUpdater.shutdown(Duration.ofMinutes(1));
+
+ final CompletableFuture<StateUpdater.RemovedTaskResult> future =
+ stateUpdater.remove(TASK_0_0, StandbyUpdateListener.SuspendReason.MIGRATED);
+
+ assertInstanceOf(StreamsException.class, assertThrows(ExecutionException.class, future::get).getCause());
+ }
+
+ @Test
+ public void shouldReportTaskAsFailedWhenStateUpdaterThreadIsNotRunningAnymore() throws Exception {
+ final StreamTask task = statefulTask(TASK_0_1, Set.of(TOPIC_PARTITION_B_0)).inState(State.RESTORING).build();
+ killStateUpdaterThread();
+
+ stateUpdater.add(task);
+
+ verifyFailedTasks(StreamsException.class, task);
+ verifyUpdatingTasks();
+ }
+
+ @Test
+ public void shouldReportFatalExceptionWhenStateUpdaterThreadDied() throws Exception {
+ final StreamTask task = statefulTask(TASK_0_0, Set.of(TOPIC_PARTITION_A_0)).inState(State.RESTORING).build();
+ final RuntimeException fatalException = new RuntimeException("Something happened");
+ doThrow(fatalException).when(changelogReader).restore(mkMap(mkEntry(TASK_0_0, task)));
+ assertEquals(Optional.empty(), stateUpdater.fatalException());
+ stateUpdater.start();
+ assertEquals(Optional.empty(), stateUpdater.fatalException());
+
+ stateUpdater.add(task);
+
+ verifyExceptionsAndFailedTasks(new ExceptionAndTask(fatalException, task));
+ assertEquals(Optional.of(fatalException), stateUpdater.fatalException());
+ }
+
+ @Test
+ public void shouldNotReportFatalExceptionWhenStateUpdaterWasShutDown() {
+ stateUpdater.start();
+
+ stateUpdater.shutdown(Duration.ofMinutes(1));
+
+ assertEquals(Optional.empty(), stateUpdater.fatalException());
+ }
+
+ private void killStateUpdaterThread() throws Exception {
+ final StreamTask task = statefulTask(TASK_0_0, Set.of(TOPIC_PARTITION_A_0)).inState(State.RESTORING).build();
+ final RuntimeException fatalException = new RuntimeException("Something happened");
+ doThrow(fatalException).when(changelogReader).restore(mkMap(mkEntry(TASK_0_0, task)));
+ stateUpdater.start();
+ stateUpdater.add(task);
+ verifyExceptionsAndFailedTasks(new ExceptionAndTask(fatalException, task));
+ // the changelog reader is cleared at the very end of the state updater thread, so this ensures that the thread
+ // stopped completely
+ verify(changelogReader, timeout(VERIFICATION_TIMEOUT)).clear();
+ assertFalse(stateUpdater.isRunning());
+ stateUpdater.drainExceptionsAndFailedTasks();
+ }
+
@Test
public void shouldRemovePausedTask() throws Exception {
final StreamTask statefulTask = statefulTask(TASK_0_0, Set.of(TOPIC_PARTITION_A_0)).inState(State.RESTORING).build();
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java
index 9540851f5cc..2548d2a7a12 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java
@@ -291,6 +291,20 @@ public class TaskManagerTest {
verify(schedulingTaskManager).unlockTasks(Set.of(taskId00, taskId01));
}
+ @Test
+ public void shouldFailStreamThreadIfStateUpdaterDied() {
+ final RuntimeException fatalException = new RuntimeException("KABOOM!");
+ when(stateUpdater.fatalException()).thenReturn(Optional.of(fatalException));
+
+ final StreamsException thrown = assertThrows(
+ StreamsException.class,
+ () -> taskManager.checkStateUpdater(time.milliseconds(), noOpResetter)
+ );
+
+ assertEquals("The state updater died and cannot update tasks anymore.", thrown.getMessage());
+ assertEquals(fatalException, thrown.getCause());
+ }
+
@Test
public void shouldNotRecreateActiveTasksThatAreAlreadyOwned() {
// Real-registry regression test reproducing the crash end-to-end on the active path (reached with
@@ -3565,9 +3579,8 @@ public class TaskManagerTest {
verify(stateUpdater).shutdown(Duration.ofMinutes(1L));
}
- @SuppressWarnings("unchecked")
@Test
- public void shouldCloseTasksIfStateUpdaterTimesOutOnRemove() throws Exception {
+ public void shouldCloseTasksIfStateUpdaterFailsRemoval() {
final StreamTask task00 = statefulTask(taskId00, taskId00ChangelogPartitions)
.inState(State.RUNNING)
.withInputPartitions(taskId00Partitions)
@@ -3578,15 +3591,41 @@ public class TaskManagerTest {
final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, tasks, false);
when(stateUpdater.tasks()).thenReturn(singleton(task00));
- final CompletableFuture<StateUpdater.RemovedTaskResult> future = mock(CompletableFuture.class);
+ final CompletableFuture<StateUpdater.RemovedTaskResult> future = new CompletableFuture<>();
when(stateUpdater.remove(eq(taskId00), eq(SuspendReason.MIGRATED))).thenReturn(future);
- when(future.get(anyLong(), any())).thenThrow(new java.util.concurrent.TimeoutException());
+ future.completeExceptionally(new StreamsException("The state updater thread died."));
taskManager.shutdown(true);
verify(task00).closeDirty();
}
+ @SuppressWarnings("unchecked")
+ @Test
+ public void shouldKeepWaitingForRemovalIfStateUpdaterDoesNotCompleteItWithinLogInterval() throws Exception {
+ final StreamTask task00 = statefulTask(taskId00, taskId00ChangelogPartitions)
+ .inState(State.RUNNING)
+ .withInputPartitions(taskId00Partitions)
+ .build();
+
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+
+ final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, tasks, false);
+
+ when(stateUpdater.tasks()).thenReturn(singleton(task00)).thenReturn(emptySet());
+ final CompletableFuture<StateUpdater.RemovedTaskResult> future = mock(CompletableFuture.class);
+ when(future.get(anyLong(), any()))
+ .thenThrow(new java.util.concurrent.TimeoutException())
+ .thenReturn(new StateUpdater.RemovedTaskResult(task00));
+ when(stateUpdater.remove(eq(taskId00), eq(SuspendReason.MIGRATED))).thenReturn(future);
+
+ taskManager.shutdown(true);
+
+ verify(future, times(2)).get(anyLong(), any());
+ verify(tasks).addTask(task00);
+ verify(task00, never()).closeDirty();
+ }
+
@Test
public void shouldPropagateSuspendExceptionWhenRevokingStandbyTask() {
final StandbyTask task00 = standbyTask(taskId00, taskId00ChangelogPartitions)