ableegoldman commented on a change in pull request #9487:
URL: https://github.com/apache/kafka/pull/9487#discussion_r524824649
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
##########
@@ -559,18 +552,52 @@ void runLoop() {
}
} catch (final TaskCorruptedException e) {
log.warn("Detected the states of tasks " +
e.corruptedTaskWithChangelogs() + " are corrupted. " +
- "Will close the task as dirty and re-create and
bootstrap from scratch.", e);
+ "Will close the task as dirty and re-create and
bootstrap from scratch.", e);
try {
taskManager.handleCorruption(e.corruptedTaskWithChangelogs());
} catch (final TaskMigratedException taskMigrated) {
handleTaskMigrated(taskMigrated);
}
} catch (final TaskMigratedException e) {
handleTaskMigrated(e);
+ } catch (final UnsupportedVersionException e) {
Review comment:
Mm ok actually I think this should be fine. I was thinking of the
handler as just "swallowing" the exception, but in reality the user would still
let the current thread die and just spin up a new one in its place. And then
the new one would hit this UnsupportedVersionException and so on, until the
brokers are upgraded. So there shouldn't be any way to get into a bad state
----------------------------------------------------------------
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:
[email protected]
