You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/08/16 02:42:48 UTC
[GitHub] [kafka] guozhangwang opened a new pull request, #12519: [WIP] KAFKA-10199: Handle exceptions from state updater
guozhangwang opened a new pull request, #12519:
URL: https://github.com/apache/kafka/pull/12519
1. In state updater, when handling task corrupted exception due to invalid restoring offset, first delete the affected partitions from the checkpoint before reporting it back to the stream thread. This is to mimic the same behavior in stream threads's StateManager#handleCorruption#closeDirtyAndRevive. It's cleaner to do so inside the restore thread, plus it enables us to optimize by only deleting those corrupted partitions, and not all.
2. In the state manager, handle the drained exceptions as follows (this is the same as handling all exceptions from `handleAssignment`): 1) Task-migrated, throw all the way to stream-thread as `handleTaskMigrated`, 2) any fatal Streams exception, throw all the way to stream-thread to trigger exception handler, 3) Task-corrupted, throw to the stream-thread as `handleCorruption`. Note that for 3), we would specially distinguish if the corrupted-tasks are already closed (when they are thrown from `handleAssignment` or not (when they are thrown from the state updater).
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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
[GitHub] [kafka] guozhangwang commented on pull request #12519: [WIP] KAFKA-10199: Handle exceptions from state updater
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on PR #12519:
URL: https://github.com/apache/kafka/pull/12519#issuecomment-1216080732
cc @cadonna would like to hear your comments before I continue adding unit tests.
--
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
[GitHub] [kafka] guozhangwang merged pull request #12519: KAFKA-10199: Handle exceptions from state updater
Posted by GitBox <gi...@apache.org>.
guozhangwang merged PR #12519:
URL: https://github.com/apache/kafka/pull/12519
--
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
[GitHub] [kafka] cadonna commented on a diff in pull request #12519: [WIP] KAFKA-10199: Handle exceptions from state updater
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12519:
URL: https://github.com/apache/kafka/pull/12519#discussion_r954839073
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -877,6 +877,8 @@ private void initializeAndRestorePhase() {
// transit to restore active is idempotent so we can call it multiple times
changelogReader.enforceRestoreActive();
+ taskManager.tryHandleExceptionsFromStateUpdater();
Review Comment:
Could we also handle the exceptions inside `tryToCompleteRestoration()`?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -667,6 +672,25 @@ private void addTasksToStateUpdater() {
}
}
+ public void tryHandleExceptionsFromStateUpdater() {
+ if (stateUpdater != null) {
+ final Map<TaskId, RuntimeException> taskExceptions = new LinkedHashMap<>();
+
+ for (final StateUpdater.ExceptionAndTasks exceptionAndTasks : stateUpdater.drainExceptionsAndFailedTasks()) {
+ final RuntimeException exception = exceptionAndTasks.exception();
+ final Set<Task> failedTasks = exceptionAndTasks.getTasks();
+
+ for (final Task failedTask : failedTasks) {
+ // need to add task back to the bookkeeping to be handled by the stream thread
+ tasks.addTask(failedTask);
+ taskExceptions.put(failedTask.id(), exception);
+ }
+ }
+
+ maybeThrowTaskExceptions(taskExceptions);
Review Comment:
Here we basically transform a mapping from exception -> set of tasks to a mapping task ID -> exception and then in `maybeThrowTaskExceptions()` we again transform the mapping task ID -> exception to a mapping exception -> set of tasks.
Can we not make this a bit more straightforward by skipping the transformation to task ID -> exception?
Is it really necessary to aggregate all `TaskCorruptedException` into a new one? I think by re-throwing a brand new `TaskCorruptedException` we lose the information where the actual exception happened in the stacktrace which might make debugging harder.
I see that with aggregating the exception we can revive all corrupted tasks all at once, but I am not sure if it is worth losing debugging information. One option could be to not drain the exceptions from the state updater, but handle one after the other. Another option could be to collect the `TaskCorruptedException` in the new exception and log all of them when the exception is caught.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -667,6 +672,25 @@ private void addTasksToStateUpdater() {
}
}
+ public void tryHandleExceptionsFromStateUpdater() {
+ if (stateUpdater != null) {
+ final Map<TaskId, RuntimeException> taskExceptions = new LinkedHashMap<>();
+
+ for (final StateUpdater.ExceptionAndTasks exceptionAndTasks : stateUpdater.drainExceptionsAndFailedTasks()) {
+ final RuntimeException exception = exceptionAndTasks.exception();
+ final Set<Task> failedTasks = exceptionAndTasks.getTasks();
+
+ for (final Task failedTask : failedTasks) {
+ // need to add task back to the bookkeeping to be handled by the stream thread
+ tasks.addTask(failedTask);
Review Comment:
Method `addTask()` does not add the partitions to the `activeTasksPerPartition` map in `Tasks`. We need to extend `addTask()` and consolidate it with `addNewActiveTask()`.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -227,34 +227,35 @@ boolean handleCorruption(final Set<TaskId> corruptedTasks) {
private void closeDirtyAndRevive(final Collection<Task> taskWithChangelogs, final boolean markAsCorrupted) {
for (final Task task : taskWithChangelogs) {
- final Collection<TopicPartition> corruptedPartitions = task.changelogPartitions();
+ if (task.state() != State.CLOSED) {
+ final Collection<TopicPartition> corruptedPartitions = task.changelogPartitions();
- // mark corrupted partitions to not be checkpointed, and then close the task as dirty
- if (markAsCorrupted) {
- task.markChangelogAsCorrupted(corruptedPartitions);
- }
+ // mark corrupted partitions to not be checkpointed, and then close the task as dirty
+ if (markAsCorrupted) {
+ task.markChangelogAsCorrupted(corruptedPartitions);
+ }
- try {
- // we do not need to take the returned offsets since we are not going to commit anyways;
- // this call is only used for active tasks to flush the cache before suspending and
- // closing the topology
- task.prepareCommit();
- } catch (final RuntimeException swallow) {
- log.error("Error flushing cache for corrupted task {} ", task.id(), swallow);
- }
+ try {
+ // we do not need to take the returned offsets since we are not going to commit anyways;
+ // this call is only used for active tasks to flush the cache before suspending and
+ // closing the topology
+ task.prepareCommit();
+ } catch (final RuntimeException swallow) {
+ log.error("Error flushing cache for corrupted task {} ", task.id(), swallow);
+ }
- try {
- task.suspend();
+ try {
+ task.suspend();
- // we need to enforce a checkpoint that removes the corrupted partitions
- if (markAsCorrupted) {
- task.postCommit(true);
+ // we need to enforce a checkpoint that removes the corrupted partitions
+ if (markAsCorrupted) {
+ task.postCommit(true);
Review Comment:
Aren't we removing the corrupted partitions twice if we use the state updater?
--
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
[GitHub] [kafka] guozhangwang commented on a diff in pull request #12519: KAFKA-10199: Handle exceptions from state updater
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on code in PR #12519:
URL: https://github.com/apache/kafka/pull/12519#discussion_r959853208
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TasksRegistry.java:
##########
@@ -57,10 +57,10 @@ public interface TasksRegistry {
void addNewActiveTasks(final Collection<Task> newTasks);
- void addNewActiveTask(final Task task);
-
void addNewStandbyTasks(final Collection<Task> newTasks);
Review Comment:
Yes I agree :) I'm actually thinking about some more aggressive change in the future, since as we complete the state updater, the task manager / task registry would then only handle running active tasks, and there's no restoring active and standby tasks ever in this class, we can further slimmer this class and even consider inlining it.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -877,6 +877,8 @@ private void initializeAndRestorePhase() {
// transit to restore active is idempotent so we can call it multiple times
changelogReader.enforceRestoreActive();
+ taskManager.tryHandleExceptionsFromStateUpdater();
Review Comment:
Ack, we can do that.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -229,34 +229,36 @@ boolean handleCorruption(final Set<TaskId> corruptedTasks) {
private void closeDirtyAndRevive(final Collection<Task> taskWithChangelogs, final boolean markAsCorrupted) {
for (final Task task : taskWithChangelogs) {
- final Collection<TopicPartition> corruptedPartitions = task.changelogPartitions();
+ if (task.state() != State.CLOSED) {
+ final Collection<TopicPartition> corruptedPartitions = task.changelogPartitions();
- // mark corrupted partitions to not be checkpointed, and then close the task as dirty
- if (markAsCorrupted) {
- task.markChangelogAsCorrupted(corruptedPartitions);
- }
+ // mark corrupted partitions to not be checkpointed, and then close the task as dirty
+ // TODO: this step should be removed as we complete migrating to state updater
+ if (markAsCorrupted && stateUpdater == null) {
Review Comment:
I have thought about that and put a JIRA ticket about exception handling for task corrupted after we've done state updater, and just to copy-paste my thoughts below:
```
Polling thread:
RecordCollector → Producer.send(callback)'s callback gets a retriable exception: We cannot simply retry sending as it would change the ordering appended to the topics. This can be thrown in the following case
Task.suspend → RecordCollector.flush() → checkException()
Task.closeClean → RecordCollector.closeClean() → checkException() (? Honestly I think it should never reach this since if there’s an error it should be found in the previous flush call?)
Normally, when RecordCollector.forward() → checkException()
```
We would handle it as fine-grained TaskEmitOrderCorruptedException. This exception does not need to mark any changelogs as corrupted.
(EOS) checkpoint not found when registering stores: This means we are unclear which snapshot the local state stores represent and hence cannot restore from a specific position.
We would throw it as fine-grained TaskStateCorruptedException.
As for the handling logic:
1) Mark the task as corrupted and hence not be scheduled for any further processing anymore.
2) At the beginning of each loop, check if there are any tasks marked as corrupted (either by itself, or sent back from the restore thread, see below). If there are any:
3) Commit for all other non-corrupted tasks first.
4) Revive those corrupted tasks by first close-dirty (which will wipe the state stores under EOS), and then convert to created mode.
```
Restore thread:
Restore consumer throws invalid offset exception. E.g. out-of-range, log truncated, no-offset-for-partition
We would throw it as fine-grained TaskChangelogCorruptedException.
```
As for handling: as we did here, we would mark changelog as corrupted and delete the topic partition entry in the checkpoint file inside the state updater, and then remove the task in the state updater and send the exception back to the polling thread to handle.
```
EOS:
When initializing the task, we found there's no checkpoint file.
```
Here I think we should just handle it directly inside the function to wipe out store and re-initialize, rather than throw an exception all the way up to stream thread to handle.
As a result of the above, the ONLY case that we should mark changelog as corrupted would be for this case.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -667,6 +672,25 @@ private void addTasksToStateUpdater() {
}
}
+ public void tryHandleExceptionsFromStateUpdater() {
+ if (stateUpdater != null) {
+ final Map<TaskId, RuntimeException> taskExceptions = new LinkedHashMap<>();
+
+ for (final StateUpdater.ExceptionAndTasks exceptionAndTasks : stateUpdater.drainExceptionsAndFailedTasks()) {
+ final RuntimeException exception = exceptionAndTasks.exception();
+ final Set<Task> failedTasks = exceptionAndTasks.getTasks();
+
+ for (final Task failedTask : failedTasks) {
+ // need to add task back to the bookkeeping to be handled by the stream thread
+ tasks.addTask(failedTask);
+ taskExceptions.put(failedTask.id(), exception);
+ }
+ }
+
+ maybeThrowTaskExceptions(taskExceptions);
Review Comment:
The motivation for `maybeThrowTaskExceptions` is that 1) if there's exceptions that would cause "all is lost", then we can directly throw that one since we are going to close-dirty all tasks; 2) otherwise, then there's only task-corrupted, then we should handle them at once. If we do not do 2) then we need to bookkeep the list of "yet handled corrupted exception" to let the thread throw and handle one by one, which would be quite complicated.. I agree that doing two transformations here is a bit awkward, we can instead just do the quoted logic 1/2 here inside the function directly (from state updater, we should either see fatal streams exception or task corrupted only), I'm doing this just for paranoid code-deduping.
On the lose log information, yeah I totally agree with you. After thinking about that a bit I think we can just log the stack trace as we group the task corrupted exception, wdyt? BTW in the future we would finer-categorize different causes of task-corrupted as different exceptions so this logic is likely going to be temporary only.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -872,14 +872,92 @@ public void shouldHandleMultipleRestoredTasks() {
taskManager.tryToCompleteRestoration(time.milliseconds(), noOpResetter);
- Mockito.verify(tasks).addNewActiveTask(taskToTransitToRunning);
+ Mockito.verify(tasks).addTask(taskToTransitToRunning);
Mockito.verify(stateUpdater).add(recycledStandbyTask);
Mockito.verify(stateUpdater).add(recycledStandbyTask);
Mockito.verify(taskToCloseClean).closeClean();
Mockito.verify(taskToCloseDirty).closeDirty();
Mockito.verify(taskToUpdateInputPartitions).updateInputPartitions(Mockito.eq(taskId05Partitions), isNull());
}
+ @Test
+ public void shouldRethrowStreamsExceptionFromStateUpdater() {
+ final StreamTask statefulTask = statefulTask(taskId00, taskId00ChangelogPartitions)
+ .inState(State.RESTORING)
+ .withInputPartitions(taskId00Partitions).build();
+ final StreamsException exception = new StreamsException("boom!");
+ final StateUpdater.ExceptionAndTasks exceptionAndTasks = new StateUpdater.ExceptionAndTasks(
+ Collections.singleton(statefulTask),
+ exception
+ );
+ when(stateUpdater.drainExceptionsAndFailedTasks()).thenReturn(Collections.singletonList(exceptionAndTasks));
+
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+ final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, tasks, true);
+
+ final StreamsException thrown = assertThrows(
+ StreamsException.class,
+ () -> taskManager.tryToCompleteRestoration(time.milliseconds(), noOpResetter)
+ );
+
+ assertEquals(exception, thrown);
+ assertEquals(statefulTask.id(), thrown.taskId().get());
+ }
+
+ @Test
+ public void shouldRethrowRuntimeExceptionFromStateUpdater() {
+ final StreamTask statefulTask = statefulTask(taskId00, taskId00ChangelogPartitions)
+ .inState(State.RESTORING)
+ .withInputPartitions(taskId00Partitions).build();
+ final RuntimeException exception = new RuntimeException("boom!");
+ final StateUpdater.ExceptionAndTasks exceptionAndTasks = new StateUpdater.ExceptionAndTasks(
+ Collections.singleton(statefulTask),
+ exception
+ );
+ when(stateUpdater.drainExceptionsAndFailedTasks()).thenReturn(Collections.singletonList(exceptionAndTasks));
+
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+ final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, tasks, true);
+
+ final StreamsException thrown = assertThrows(
+ StreamsException.class,
+ () -> taskManager.tryToCompleteRestoration(time.milliseconds(), noOpResetter)
+ );
+
+ assertEquals(exception, thrown.getCause());
Review Comment:
Yup, good idea!
I will add for those cases where we do not validate `equals(exception, thrown)` directly.
--
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
[GitHub] [kafka] cadonna commented on pull request #12519: KAFKA-10199: Handle exceptions from state updater
Posted by GitBox <gi...@apache.org>.
cadonna commented on PR #12519:
URL: https://github.com/apache/kafka/pull/12519#issuecomment-1234414581
I am sorry, one of my merged PRs has introduced conflicts 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
[GitHub] [kafka] cadonna commented on a diff in pull request #12519: KAFKA-10199: Handle exceptions from state updater
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12519:
URL: https://github.com/apache/kafka/pull/12519#discussion_r959304723
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -872,14 +872,92 @@ public void shouldHandleMultipleRestoredTasks() {
taskManager.tryToCompleteRestoration(time.milliseconds(), noOpResetter);
- Mockito.verify(tasks).addNewActiveTask(taskToTransitToRunning);
+ Mockito.verify(tasks).addTask(taskToTransitToRunning);
Mockito.verify(stateUpdater).add(recycledStandbyTask);
Mockito.verify(stateUpdater).add(recycledStandbyTask);
Mockito.verify(taskToCloseClean).closeClean();
Mockito.verify(taskToCloseDirty).closeDirty();
Mockito.verify(taskToUpdateInputPartitions).updateInputPartitions(Mockito.eq(taskId05Partitions), isNull());
}
+ @Test
+ public void shouldRethrowStreamsExceptionFromStateUpdater() {
+ final StreamTask statefulTask = statefulTask(taskId00, taskId00ChangelogPartitions)
+ .inState(State.RESTORING)
+ .withInputPartitions(taskId00Partitions).build();
+ final StreamsException exception = new StreamsException("boom!");
+ final StateUpdater.ExceptionAndTasks exceptionAndTasks = new StateUpdater.ExceptionAndTasks(
+ Collections.singleton(statefulTask),
+ exception
+ );
+ when(stateUpdater.drainExceptionsAndFailedTasks()).thenReturn(Collections.singletonList(exceptionAndTasks));
+
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+ final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, tasks, true);
+
+ final StreamsException thrown = assertThrows(
+ StreamsException.class,
+ () -> taskManager.tryToCompleteRestoration(time.milliseconds(), noOpResetter)
+ );
+
+ assertEquals(exception, thrown);
+ assertEquals(statefulTask.id(), thrown.taskId().get());
+ }
+
+ @Test
+ public void shouldRethrowRuntimeExceptionFromStateUpdater() {
+ final StreamTask statefulTask = statefulTask(taskId00, taskId00ChangelogPartitions)
+ .inState(State.RESTORING)
+ .withInputPartitions(taskId00Partitions).build();
+ final RuntimeException exception = new RuntimeException("boom!");
+ final StateUpdater.ExceptionAndTasks exceptionAndTasks = new StateUpdater.ExceptionAndTasks(
+ Collections.singleton(statefulTask),
+ exception
+ );
+ when(stateUpdater.drainExceptionsAndFailedTasks()).thenReturn(Collections.singletonList(exceptionAndTasks));
+
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+ final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, tasks, true);
+
+ final StreamsException thrown = assertThrows(
+ StreamsException.class,
+ () -> taskManager.tryToCompleteRestoration(time.milliseconds(), noOpResetter)
+ );
+
+ assertEquals(exception, thrown.getCause());
Review Comment:
I think here it would be good also to verify the exception message since it should be "First unexpected error for task ...", AFAIU.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -229,34 +229,36 @@ boolean handleCorruption(final Set<TaskId> corruptedTasks) {
private void closeDirtyAndRevive(final Collection<Task> taskWithChangelogs, final boolean markAsCorrupted) {
for (final Task task : taskWithChangelogs) {
- final Collection<TopicPartition> corruptedPartitions = task.changelogPartitions();
+ if (task.state() != State.CLOSED) {
+ final Collection<TopicPartition> corruptedPartitions = task.changelogPartitions();
- // mark corrupted partitions to not be checkpointed, and then close the task as dirty
- if (markAsCorrupted) {
- task.markChangelogAsCorrupted(corruptedPartitions);
- }
+ // mark corrupted partitions to not be checkpointed, and then close the task as dirty
+ // TODO: this step should be removed as we complete migrating to state updater
+ if (markAsCorrupted && stateUpdater == null) {
Review Comment:
Are you sure about this condition and the ToDo? A `TaskCorruptedException` can also be thrown outside the state updater when the task is not restoring.
Maybe we should remove the checkpointing due to corruption from the state updater since at the moment we mark all changelogs as corrupted also there? We can create a ticket for the improvement you are proposing of only marking as corrupted the actual corrupted changelogs.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TasksRegistry.java:
##########
@@ -57,10 +57,10 @@ public interface TasksRegistry {
void addNewActiveTasks(final Collection<Task> newTasks);
- void addNewActiveTask(final Task task);
-
void addNewStandbyTasks(final Collection<Task> newTasks);
Review Comment:
I think in future we can consolidate `addNewActiveTasks()` and `addNewStandbyTasks()` into `addTasks()`.
Just an idea, no action needed for this PR!
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -322,40 +324,46 @@ public void handleAssignment(final Map<TaskId, Set<TopicPartition>> activeTasks,
final Map<TaskId, RuntimeException> taskCloseExceptions = closeAndRecycleTasks(tasksToRecycle, tasksToCloseClean);
- throwTaskExceptions(taskCloseExceptions);
+ maybeThrowTaskExceptions(taskCloseExceptions);
createNewTasks(activeTasksToCreate, standbyTasksToCreate);
}
- private void throwTaskExceptions(final Map<TaskId, RuntimeException> taskExceptions) {
+ // if at least one of the exception is a task-migrated exception, then directly throw since it indicates all tasks are lost
+ // if at least one of the exception is a streams exception, then directly throw since it should be handled by thread's handler
+ // if at least one of the exception is a non-streams exception, then wrap and throw since it should be handled by thread's handler
+ // otherwise, all the exceptions are task-corrupted, then merge their tasks and throw a single one
+ // TODO: move task-corrupted and task-migrated out of the public errors package since they are internal errors and always be
+ // handled by Streams library itself
+ private void maybeThrowTaskExceptions(final Map<TaskId, RuntimeException> taskExceptions) {
if (!taskExceptions.isEmpty()) {
log.error("Get exceptions for the following tasks: {}", taskExceptions);
+ final TaskCorruptedException allTaskCorrupts = new TaskCorruptedException(new HashSet<>());
for (final Map.Entry<TaskId, RuntimeException> entry : taskExceptions.entrySet()) {
- if (!(entry.getValue() instanceof TaskMigratedException)) {
- final TaskId taskId = entry.getKey();
- final RuntimeException exception = entry.getValue();
- if (exception instanceof StreamsException) {
+ final TaskId taskId = entry.getKey();
+ final RuntimeException exception = entry.getValue();
+
+ if (exception instanceof StreamsException) {
+ if (exception instanceof TaskMigratedException) {
+ throw entry.getValue();
+ } else if (exception instanceof TaskCorruptedException) {
+ log.warn("Encounter corrupted task" + taskId + ", will group it with other corrupted tasks " +
+ "and handle together", exception);
+ allTaskCorrupts.corruptedTasks().add(taskId);
+ } else {
((StreamsException) exception).setTaskId(taskId);
throw exception;
- } else if (exception instanceof KafkaException) {
- throw new StreamsException(exception, taskId);
- } else {
- throw new StreamsException(
- "Unexpected failure to close " + taskExceptions.size() +
- " task(s) [" + taskExceptions.keySet() + "]. " +
- "First unexpected exception (for task " + taskId + ") follows.",
- exception,
- taskId
- );
+
Review Comment:
nit: remove line
--
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
[GitHub] [kafka] guozhangwang commented on pull request #12519: KAFKA-10199: Handle exceptions from state updater
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on PR #12519:
URL: https://github.com/apache/kafka/pull/12519#issuecomment-1236007271
Test failures are not relevant. I'm going to merge to trunk.
--
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