You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "gabriellefu (via GitHub)" <gi...@apache.org> on 2026/05/13 22:20:34 UTC
[PR] [KAFKA-16876] TaskManager.handleRevocation doesn't handle errors thrown from task.prepareCommit [kafka]
gabriellefu opened a new pull request, #22282:
URL: https://github.com/apache/kafka/pull/22282
1. Fix TaskManager.handleRevocation to always suspend revoked tasks, even when prepareCommit throws (e.g. TaskMigratedException from producer.send during cache flush). Previously the exception propagated uncaught, skipping the suspend loop entirely. This left tasks in RUNNING state, which caused a downstream IllegalStateException when handleAssignment tried to close them.
2. Wrap prepare/commit/postCommit in try-finally so the suspend loop and task unlock are guaranteed to execute regardless of where an exception occurs.
3. Preserve all exceptions via addSuppressed instead of silently dropping later exceptions. The first exception remains the primary thrown exception for backward compatibility, but subsequent exceptions (e.g. the IllegalStateException from closing an unsuspended task) are now attached as suppressed exceptions instead of lost.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16876: TaskManager.handleRevocation doesn't handle errors thrown from task.prepareCommit [kafka]
Posted by "gabriellefu (via GitHub)" <gi...@apache.org>.
gabriellefu commented on code in PR #22282:
URL: https://github.com/apache/kafka/pull/22282#discussion_r3299424197
##########
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:
added the check, thanks!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16876: TaskManager.handleRevocation doesn't handle errors thrown from task.prepareCommit [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #22282:
URL: https://github.com/apache/kafka/pull/22282#discussion_r3247219830
##########
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:
Please replace with parametrised message.
##########
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);
Review Comment:
ditto
##########
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);
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);
Review Comment:
ditto
##########
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);
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);
Review Comment:
ditto
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16876: TaskManager.handleRevocation doesn't handle errors thrown from task.prepareCommit [kafka]
Posted by "gabriellefu (via GitHub)" <gi...@apache.org>.
gabriellefu commented on code in PR #22282:
URL: https://github.com/apache/kafka/pull/22282#discussion_r3299447181
##########
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:
I have changed the name since I think this pr is mostly focusing on prepareCommit throws, suspend still run case.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16876: TaskManager.handleRevocation doesn't handle errors thrown from task.prepareCommit [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16876: TaskManager.handleRevocation doesn't handle errors thrown from task.prepareCommit [kafka]
Posted by "gabriellefu (via GitHub)" <gi...@apache.org>.
gabriellefu commented on code in PR #22282:
URL: https://github.com/apache/kafka/pull/22282#discussion_r3299404386
##########
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:
added a try/catch block here
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16876: TaskManager.handleRevocation doesn't handle errors thrown from task.prepareCommit [kafka]
Posted by "gabriellefu (via GitHub)" <gi...@apache.org>.
gabriellefu commented on code in PR #22282:
URL: https://github.com/apache/kafka/pull/22282#discussion_r3299447181
##########
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:
I have changed the name since I think this pr is mostly focusing on prepareCommit throws, suspend still run.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16876: TaskManager.handleRevocation doesn't handle errors thrown from task.prepareCommit [kafka]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #22282:
URL: https://github.com/apache/kafka/pull/22282#discussion_r3287966984
##########
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
Review Comment:
Typo in comment: "RUNNING tate" should be "RUNNING state".
##########
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 {
Review Comment:
The try/finally that guarantees task suspension and directory unlock only starts after revokeTasksInStateUpdater(...). If revokeTasksInStateUpdater (or anything before the try block after maybeLockTasks) throws, revokedActiveTasks may still not be suspended and lockedTaskIds may remain locked. Consider expanding the try/finally to begin immediately after maybeLockTasks (or at least include revokeTasksInStateUpdater) so suspend/unlock is guaranteed for all failures during revocation.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16876: TaskManager.handleRevocation doesn't handle errors thrown from task.prepareCommit [kafka]
Posted by "gabriellefu (via GitHub)" <gi...@apache.org>.
gabriellefu commented on code in PR #22282:
URL: https://github.com/apache/kafka/pull/22282#discussion_r3299385622
##########
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:
I have change the name to tasksToSkipPostCommit to avoid misleading
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16876: TaskManager.handleRevocation doesn't handle errors thrown from task.prepareCommit [kafka]
Posted by "gabriellefu (via GitHub)" <gi...@apache.org>.
gabriellefu commented on code in PR #22282:
URL: https://github.com/apache/kafka/pull/22282#discussion_r3299384599
##########
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:
I have added the commitNeededActiveTasks to the log too
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16876: TaskManager.handleRevocation doesn't handle errors thrown from task.prepareCommit [kafka]
Posted by "gabriellefu (via GitHub)" <gi...@apache.org>.
gabriellefu commented on code in PR #22282:
URL: https://github.com/apache/kafka/pull/22282#discussion_r3299435323
##########
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:
I added a thrown on the taskid, thanks
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16876: TaskManager.handleRevocation doesn't handle errors thrown from task.prepareCommit [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #22282:
URL: https://github.com/apache/kafka/pull/22282
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org