ableegoldman commented on a change in pull request #10407:
URL: https://github.com/apache/kafka/pull/10407#discussion_r602584404
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
##########
@@ -167,18 +167,7 @@ void handleCorruption(final Set<TaskId> corruptedTasks) {
}
}
- // Make sure to clean up any corrupted standby tasks in their entirety
before committing
- // since TaskMigrated can be thrown and the resulting handleLostAll
will only clean up active tasks
closeAndRevive(corruptedStandbyTasks);
-
- commit(tasks()
Review comment:
I think we do need to do the `commit` including the `abortTxn` since we
can now get a TaskCorrupted at any point, not just on RESTORING tasks which
don't have anything to abort. But since we're already in `handleCorrupted` we
may as well just catch the TaskCorrupted, tack on any new corrupted tasks, and
proceed with the closeAndRevive
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
