lucasbru commented on code in PR #22282:
URL: https://github.com/apache/kafka/pull/22282#discussion_r3287756043
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1057,81 +1057,91 @@ void handleRevocation(final Collection<TopicPartition>
revokedPartitions) {
"have been cleaned up by the handleAssignment
callback.", remainingRevokedPartitions);
}
- if (revokedTasksNeedCommit) {
- prepareCommitAndAddOffsetsToMap(revokedActiveTasks,
consumedOffsetsPerTask);
- // if we need to commit any revoking task then we just commit all
of those needed committing together
- prepareCommitAndAddOffsetsToMap(commitNeededActiveTasks,
consumedOffsetsPerTask);
- }
-
- // even if commit failed, we should still continue and complete
suspending those tasks, so we would capture
- // any exception and rethrow it at the end. some exceptions may be
handled immediately and then swallowed,
- // as such we just need to skip those dirty tasks in the checkpoint
+ // even if prepare, commit, or postCommit failed, we must still
suspend revoked tasks and unlock,
+ // so we use try-finally to guarantee that. Exceptions are captured
and rethrown at the end.
final Set<Task> dirtyTasks = new
TreeSet<>(Comparator.comparing(Task::id));
+ boolean prepareCommitSucceeded = false;
try {
if (revokedTasksNeedCommit) {
- // in handleRevocation we must call
commitOffsetsOrTransaction() directly rather than
- // commitAndFillInConsumedOffsetsAndMetadataPerTaskMap() to
make sure we don't skip the
- // offset commit because we are in a rebalance
-
taskExecutor.commitOffsetsOrTransaction(consumedOffsetsPerTask);
- }
- } catch (final TaskCorruptedException e) {
- log.warn("Some tasks were corrupted when trying to commit offsets,
these will be cleaned and revived: {}",
- e.corruptedTasks());
-
- // If we hit a TaskCorruptedException it must be EOS, just handle
the cleanup for those corrupted tasks right here
- dirtyTasks.addAll(tasks.initializedTasks(e.corruptedTasks()));
- closeDirtyAndRevive(dirtyTasks, true);
- } catch (final TimeoutException e) {
- log.warn("Timed out while trying to commit all tasks during
revocation, these will be cleaned and revived");
-
- // If we hit a TimeoutException it must be ALOS, just close dirty
and revive without wiping the state
- dirtyTasks.addAll(consumedOffsetsPerTask.keySet());
- closeDirtyAndRevive(dirtyTasks, false);
- } catch (final RuntimeException e) {
- log.error("Exception caught while committing those revoked tasks "
+ revokedActiveTasks, e);
- firstException.compareAndSet(null, e);
- dirtyTasks.addAll(consumedOffsetsPerTask.keySet());
- }
-
- // we enforce checkpointing upon suspending a task: if it is resumed
later we just proceed normally, if it is
- // going to be closed we would checkpoint by then
- for (final Task task : revokedActiveTasks) {
- if (!dirtyTasks.contains(task)) {
try {
- task.postCommit(true);
+ prepareCommitAndAddOffsetsToMap(revokedActiveTasks,
consumedOffsetsPerTask);
+ // if we need to commit any revoking task then we just
commit all of those needed committing together
+ prepareCommitAndAddOffsetsToMap(commitNeededActiveTasks,
consumedOffsetsPerTask);
+ prepareCommitSucceeded = true;
} catch (final RuntimeException e) {
- log.error("Exception caught while post-committing task " +
task.id(), e);
- maybeSetFirstException(false, maybeWrapTaskException(e,
task.id()), firstException);
+ log.error("Exception caught while preparing to commit
revoked tasks {}", revokedActiveTasks, e);
+ maybeSetFirstException(false, e, firstException);
+ dirtyTasks.addAll(revokedActiveTasks);
+ dirtyTasks.addAll(commitNeededActiveTasks);
Review Comment:
Why add commitNeededActiveTasks to dirtyTasks here? They're non-revoked, the
suspend loop in the finally won't touch them, and the dirtyTasks check below
only gates the postCommit skip. If the intent is just "don't postCommit since
we never committed", a short comment would help - the name dirtyTasks made me
look for a close-dirty path that doesn't exist.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -2048,7 +2058,9 @@ private void maybeSetFirstException(final boolean
ignoreTaskMigrated,
final RuntimeException exception,
final
AtomicReference<RuntimeException> firstException) {
if (!ignoreTaskMigrated || !(exception instanceof
TaskMigratedException)) {
- firstException.compareAndSet(null, exception);
+ if (!firstException.compareAndSet(null, exception)) {
+ firstException.get().addSuppressed(exception);
Review Comment:
addSuppressed is neat. I actually wasn't aware of it.
It throws IllegalArgumentException if you hand it the same instance that's
already firstException.get() (self-suppression). I don't see a path that hits
that today, but maybeWrapTaskException returns its argument unchanged when it's
already a StreamsException, so it's not hard to imagine future callers
stumbling into it. A quick `if (exception != firstException.get())` guard would
harden it cheaply.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1384,14 +1394,14 @@ void shutdown(final boolean clean) {
executeAndMaybeSwallow(
clean,
() -> closeAndCleanUpTasks(activeTasks, standbyTasks, clean),
- e -> firstException.compareAndSet(null, e),
+ e -> maybeSetFirstException(false, e, firstException),
Review Comment:
These three callers in shutdown() also got switched from compareAndSet to
maybeSetFirstException - that's a real behavior change (later exceptions now
get attached as suppressed instead of dropped). It's nice, but the PR
description only mentions handleRevocation. Worth calling out in the
description since shutdown failure visibility changes too. Same for the
tryCloseCleanActiveTasks ones lower down.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1057,81 +1057,91 @@ void handleRevocation(final Collection<TopicPartition>
revokedPartitions) {
"have been cleaned up by the handleAssignment
callback.", remainingRevokedPartitions);
}
- if (revokedTasksNeedCommit) {
- prepareCommitAndAddOffsetsToMap(revokedActiveTasks,
consumedOffsetsPerTask);
- // if we need to commit any revoking task then we just commit all
of those needed committing together
- prepareCommitAndAddOffsetsToMap(commitNeededActiveTasks,
consumedOffsetsPerTask);
- }
-
- // even if commit failed, we should still continue and complete
suspending those tasks, so we would capture
- // any exception and rethrow it at the end. some exceptions may be
handled immediately and then swallowed,
- // as such we just need to skip those dirty tasks in the checkpoint
+ // even if prepare, commit, or postCommit failed, we must still
suspend revoked tasks and unlock,
+ // so we use try-finally to guarantee that. Exceptions are captured
and rethrown at the end.
final Set<Task> dirtyTasks = new
TreeSet<>(Comparator.comparing(Task::id));
+ boolean prepareCommitSucceeded = false;
try {
if (revokedTasksNeedCommit) {
- // in handleRevocation we must call
commitOffsetsOrTransaction() directly rather than
- // commitAndFillInConsumedOffsetsAndMetadataPerTaskMap() to
make sure we don't skip the
- // offset commit because we are in a rebalance
-
taskExecutor.commitOffsetsOrTransaction(consumedOffsetsPerTask);
- }
- } catch (final TaskCorruptedException e) {
- log.warn("Some tasks were corrupted when trying to commit offsets,
these will be cleaned and revived: {}",
- e.corruptedTasks());
-
- // If we hit a TaskCorruptedException it must be EOS, just handle
the cleanup for those corrupted tasks right here
- dirtyTasks.addAll(tasks.initializedTasks(e.corruptedTasks()));
- closeDirtyAndRevive(dirtyTasks, true);
- } catch (final TimeoutException e) {
- log.warn("Timed out while trying to commit all tasks during
revocation, these will be cleaned and revived");
-
- // If we hit a TimeoutException it must be ALOS, just close dirty
and revive without wiping the state
- dirtyTasks.addAll(consumedOffsetsPerTask.keySet());
- closeDirtyAndRevive(dirtyTasks, false);
- } catch (final RuntimeException e) {
- log.error("Exception caught while committing those revoked tasks "
+ revokedActiveTasks, e);
- firstException.compareAndSet(null, e);
- dirtyTasks.addAll(consumedOffsetsPerTask.keySet());
- }
-
- // we enforce checkpointing upon suspending a task: if it is resumed
later we just proceed normally, if it is
- // going to be closed we would checkpoint by then
- for (final Task task : revokedActiveTasks) {
- if (!dirtyTasks.contains(task)) {
try {
- task.postCommit(true);
+ prepareCommitAndAddOffsetsToMap(revokedActiveTasks,
consumedOffsetsPerTask);
+ // if we need to commit any revoking task then we just
commit all of those needed committing together
+ prepareCommitAndAddOffsetsToMap(commitNeededActiveTasks,
consumedOffsetsPerTask);
+ prepareCommitSucceeded = true;
} catch (final RuntimeException e) {
- log.error("Exception caught while post-committing task " +
task.id(), e);
- maybeSetFirstException(false, maybeWrapTaskException(e,
task.id()), firstException);
+ log.error("Exception caught while preparing to commit
revoked tasks {}", revokedActiveTasks, e);
Review Comment:
Minor: this log fires for failures while preparing commitNeededActiveTasks
too, but the message only mentions "revoked tasks". Maybe just "...while
preparing to commit tasks" or include both sets.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2989,6 +2989,54 @@ public void shouldSuspendActiveTasksDuringRevocation() {
verify(task00).suspend();
}
+ @Test
+ public void shouldSuspendRevokedTasksWhenPrepareCommitThrows() {
+ final StreamTask task00 = statefulTask(taskId00,
taskId00ChangelogPartitions)
+ .withInputPartitions(taskId00Partitions)
+ .inState(State.RUNNING)
+ .build();
+
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+ when(tasks.allInitializedTasks()).thenReturn(Set.of(task00));
+
+ when(task00.commitNeeded()).thenReturn(true);
+ when(task00.prepareCommit(true)).thenThrow(new
TaskMigratedException("task migrated"));
+
+ final TaskManager taskManager =
setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, tasks);
+
+ final StreamsException thrown = assertThrows(StreamsException.class,
+ () -> taskManager.handleRevocation(taskId00Partitions));
+
+ assertInstanceOf(TaskMigratedException.class, thrown);
Review Comment:
Could also assert the thrown TaskMigratedException has its taskId set -
prepareCommitAndAddOffsetsToMap calls setTaskId on the way through, and this is
the only test currently exercising that branch for prepareCommit. Worth pinning
it down.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1057,81 +1057,91 @@ void handleRevocation(final Collection<TopicPartition>
revokedPartitions) {
"have been cleaned up by the handleAssignment
callback.", remainingRevokedPartitions);
}
- if (revokedTasksNeedCommit) {
- prepareCommitAndAddOffsetsToMap(revokedActiveTasks,
consumedOffsetsPerTask);
- // if we need to commit any revoking task then we just commit all
of those needed committing together
- prepareCommitAndAddOffsetsToMap(commitNeededActiveTasks,
consumedOffsetsPerTask);
- }
-
- // even if commit failed, we should still continue and complete
suspending those tasks, so we would capture
- // any exception and rethrow it at the end. some exceptions may be
handled immediately and then swallowed,
- // as such we just need to skip those dirty tasks in the checkpoint
+ // even if prepare, commit, or postCommit failed, we must still
suspend revoked tasks and unlock,
+ // so we use try-finally to guarantee that. Exceptions are captured
and rethrown at the end.
final Set<Task> dirtyTasks = new
TreeSet<>(Comparator.comparing(Task::id));
+ boolean prepareCommitSucceeded = false;
try {
if (revokedTasksNeedCommit) {
- // in handleRevocation we must call
commitOffsetsOrTransaction() directly rather than
- // commitAndFillInConsumedOffsetsAndMetadataPerTaskMap() to
make sure we don't skip the
- // offset commit because we are in a rebalance
-
taskExecutor.commitOffsetsOrTransaction(consumedOffsetsPerTask);
- }
- } catch (final TaskCorruptedException e) {
- log.warn("Some tasks were corrupted when trying to commit offsets,
these will be cleaned and revived: {}",
- e.corruptedTasks());
-
- // If we hit a TaskCorruptedException it must be EOS, just handle
the cleanup for those corrupted tasks right here
- dirtyTasks.addAll(tasks.initializedTasks(e.corruptedTasks()));
- closeDirtyAndRevive(dirtyTasks, true);
- } catch (final TimeoutException e) {
- log.warn("Timed out while trying to commit all tasks during
revocation, these will be cleaned and revived");
-
- // If we hit a TimeoutException it must be ALOS, just close dirty
and revive without wiping the state
- dirtyTasks.addAll(consumedOffsetsPerTask.keySet());
- closeDirtyAndRevive(dirtyTasks, false);
- } catch (final RuntimeException e) {
- log.error("Exception caught while committing those revoked tasks "
+ revokedActiveTasks, e);
- firstException.compareAndSet(null, e);
- dirtyTasks.addAll(consumedOffsetsPerTask.keySet());
- }
-
- // we enforce checkpointing upon suspending a task: if it is resumed
later we just proceed normally, if it is
- // going to be closed we would checkpoint by then
- for (final Task task : revokedActiveTasks) {
- if (!dirtyTasks.contains(task)) {
try {
- task.postCommit(true);
+ prepareCommitAndAddOffsetsToMap(revokedActiveTasks,
consumedOffsetsPerTask);
+ // if we need to commit any revoking task then we just
commit all of those needed committing together
+ prepareCommitAndAddOffsetsToMap(commitNeededActiveTasks,
consumedOffsetsPerTask);
+ prepareCommitSucceeded = true;
} catch (final RuntimeException e) {
- log.error("Exception caught while post-committing task " +
task.id(), e);
- maybeSetFirstException(false, maybeWrapTaskException(e,
task.id()), firstException);
+ log.error("Exception caught while preparing to commit
revoked tasks {}", revokedActiveTasks, e);
+ maybeSetFirstException(false, e, firstException);
+ dirtyTasks.addAll(revokedActiveTasks);
+ dirtyTasks.addAll(commitNeededActiveTasks);
}
}
- }
- if (revokedTasksNeedCommit) {
- for (final Task task : commitNeededActiveTasks) {
+ try {
+ if (revokedTasksNeedCommit && prepareCommitSucceeded) {
+ // in handleRevocation we must call
commitOffsetsOrTransaction() directly rather than
+ // commitAndFillInConsumedOffsetsAndMetadataPerTaskMap()
to make sure we don't skip the
+ // offset commit because we are in a rebalance
+
taskExecutor.commitOffsetsOrTransaction(consumedOffsetsPerTask);
+ }
+ } catch (final TaskCorruptedException e) {
+ log.warn("Some tasks were corrupted when trying to commit
offsets, these will be cleaned and revived: {}",
+ e.corruptedTasks());
+
+ // If we hit a TaskCorruptedException it must be EOS, just
handle the cleanup for those corrupted tasks right here
+ dirtyTasks.addAll(tasks.initializedTasks(e.corruptedTasks()));
+ closeDirtyAndRevive(dirtyTasks, true);
+ } catch (final TimeoutException e) {
+ log.warn("Timed out while trying to commit all tasks during
revocation, these will be cleaned and revived");
+
+ // If we hit a TimeoutException it must be ALOS, just close
dirty and revive without wiping the state
+ dirtyTasks.addAll(consumedOffsetsPerTask.keySet());
+ closeDirtyAndRevive(dirtyTasks, false);
+ } catch (final RuntimeException e) {
+ log.error("Exception caught while committing those revoked
tasks {}", revokedActiveTasks, e);
+ maybeSetFirstException(false, e, firstException);
+ dirtyTasks.addAll(consumedOffsetsPerTask.keySet());
+ }
+
+ // we enforce checkpointing upon suspending a task: if it is
resumed later we just proceed normally, if it is
+ // going to be closed we would checkpoint by then
+ for (final Task task : revokedActiveTasks) {
if (!dirtyTasks.contains(task)) {
try {
- // for non-revoking active tasks, we should not
enforce checkpoint
- // since if it is EOS enabled, no checkpoint should be
written while
- // the task is in RUNNING tate
- task.postCommit(false);
+ task.postCommit(true);
} catch (final RuntimeException e) {
- log.error("Exception caught while post-committing task
" + task.id(), e);
+ log.error("Exception caught while post-committing task
{}", task.id(), e);
maybeSetFirstException(false,
maybeWrapTaskException(e, task.id()), firstException);
}
}
}
- }
- for (final Task task : revokedActiveTasks) {
- try {
- task.suspend();
- } catch (final RuntimeException e) {
- log.error("Caught the following exception while trying to
suspend revoked task " + task.id(), e);
- maybeSetFirstException(false, maybeWrapTaskException(e,
task.id()), firstException);
+ if (revokedTasksNeedCommit) {
+ for (final Task task : commitNeededActiveTasks) {
+ if (!dirtyTasks.contains(task)) {
+ try {
+ // for non-revoking active tasks, we should not
enforce checkpoint
+ // since if it is EOS enabled, no checkpoint
should be written while
+ // the task is in RUNNING tate
+ task.postCommit(false);
+ } catch (final RuntimeException e) {
+ log.error("Exception caught while post-committing
task {}", task.id(), e);
+ maybeSetFirstException(false,
maybeWrapTaskException(e, task.id()), firstException);
+ }
+ }
+ }
+ }
+ } finally {
+ for (final Task task : revokedActiveTasks) {
+ try {
+ task.suspend();
+ } catch (final RuntimeException e) {
+ log.error("Caught the following exception while trying to
suspend revoked task {}", task.id(), e);
+ maybeSetFirstException(false, maybeWrapTaskException(e,
task.id()), firstException);
+ }
}
- }
- maybeUnlockTasks(lockedTaskIds);
+ maybeUnlockTasks(lockedTaskIds);
Review Comment:
If maybeUnlockTasks throws here, it'll propagate out of the finally and the
`throw firstException.get()` below never runs - whatever revocation failure we
just captured gets dropped. Same behavior as before the refactor, but now that
we're explicitly going for a guaranteed-cleanup finally, might be worth
wrapping this in try/catch and routing through maybeSetFirstException too.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2989,6 +2989,54 @@ public void shouldSuspendActiveTasksDuringRevocation() {
verify(task00).suspend();
}
+ @Test
+ public void shouldSuspendRevokedTasksWhenPrepareCommitThrows() {
+ final StreamTask task00 = statefulTask(taskId00,
taskId00ChangelogPartitions)
+ .withInputPartitions(taskId00Partitions)
+ .inState(State.RUNNING)
+ .build();
+
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+ when(tasks.allInitializedTasks()).thenReturn(Set.of(task00));
+
+ when(task00.commitNeeded()).thenReturn(true);
+ when(task00.prepareCommit(true)).thenThrow(new
TaskMigratedException("task migrated"));
+
+ final TaskManager taskManager =
setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, tasks);
+
+ final StreamsException thrown = assertThrows(StreamsException.class,
+ () -> taskManager.handleRevocation(taskId00Partitions));
+
+ assertInstanceOf(TaskMigratedException.class, thrown);
+
+ verify(task00).suspend();
+ verify(task00, never()).postCommit(anyBoolean());
+ }
+
+ @Test
+ public void
shouldAttachSuppressedExceptionsWhenMultiplePhasesFailDuringRevocation() {
Review Comment:
Name says "MultiplePhases" but you're only failing prepareCommit and suspend
(two). Could either trim the name or extend it to fail commit/postCommit too so
the suppression chain is genuinely exercised on >2 failures.
--
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]
