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 2020/05/14 17:44:46 UTC
[GitHub] [kafka] abbccdda opened a new pull request #8667: KAFKA-9994: Handle task migrated inside corruption path
abbccdda opened a new pull request #8667:
URL: https://github.com/apache/kafka/pull/8667
The TaskMigratedException should always be non-fatal and caught within the run loop. Adding try-catch for corrupted path as well is necessary.
### 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] abbccdda commented on a change in pull request #8667: KAFKA-9994: Handle task migrated inside corruption path
Posted by GitBox <gi...@apache.org>.
abbccdda commented on a change in pull request #8667:
URL: https://github.com/apache/kafka/pull/8667#discussion_r425366150
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
##########
@@ -555,28 +555,35 @@ 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);
-
- taskManager.commit(
- taskManager.tasks()
- .values()
- .stream()
- .filter(t -> t.state() == Task.State.RUNNING || t.state() == Task.State.RESTORING)
- .filter(t -> !e.corruptedTaskWithChangelogs().containsKey(t.id()))
- .collect(Collectors.toSet())
- );
- taskManager.handleCorruption(e.corruptedTaskWithChangelogs());
+ try {
+ taskManager.commit(
+ taskManager.tasks()
+ .values()
+ .stream()
+ .filter(t -> t.state() == Task.State.RUNNING || t.state() == Task.State.RESTORING)
+ .filter(t -> !e.corruptedTaskWithChangelogs().containsKey(t.id()))
+ .collect(Collectors.toSet())
+ );
+ taskManager.handleCorruption(e.corruptedTaskWithChangelogs());
+ } catch (final TaskMigratedException taskMigrated) {
+ handleTaskMigrated(taskMigrated);
Review comment:
Good point, I was also thinking whether we should do the corruption logic no matter what. But if we hit a TaskMigrated, the `taskManager.handleLostAll` will wipe out all the task states dirty, which seems like a super-set of jobs for handleCorruption. If we failed the commit, maybe we should just skip the corruption logic?
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] vvcephei merged pull request #8667: KAFKA-9994: Handle task migrated inside corruption path
Posted by GitBox <gi...@apache.org>.
vvcephei merged pull request #8667:
URL: https://github.com/apache/kafka/pull/8667
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on a change in pull request #8667: KAFKA-9994: Handle task migrated inside corruption path
Posted by GitBox <gi...@apache.org>.
vvcephei commented on a change in pull request #8667:
URL: https://github.com/apache/kafka/pull/8667#discussion_r425386296
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
##########
@@ -555,28 +555,35 @@ 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);
-
- taskManager.commit(
- taskManager.tasks()
- .values()
- .stream()
- .filter(t -> t.state() == Task.State.RUNNING || t.state() == Task.State.RESTORING)
- .filter(t -> !e.corruptedTaskWithChangelogs().containsKey(t.id()))
- .collect(Collectors.toSet())
- );
- taskManager.handleCorruption(e.corruptedTaskWithChangelogs());
+ try {
+ taskManager.commit(
+ taskManager.tasks()
+ .values()
+ .stream()
+ .filter(t -> t.state() == Task.State.RUNNING || t.state() == Task.State.RESTORING)
+ .filter(t -> !e.corruptedTaskWithChangelogs().containsKey(t.id()))
+ .collect(Collectors.toSet())
+ );
+ taskManager.handleCorruption(e.corruptedTaskWithChangelogs());
+ } catch (final TaskMigratedException taskMigrated) {
+ handleTaskMigrated(taskMigrated);
Review comment:
Sounds legit. 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on a change in pull request #8667: KAFKA-9994: Handle task migrated inside corruption path
Posted by GitBox <gi...@apache.org>.
vvcephei commented on a change in pull request #8667:
URL: https://github.com/apache/kafka/pull/8667#discussion_r425348145
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
##########
@@ -555,28 +555,35 @@ 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);
-
- taskManager.commit(
- taskManager.tasks()
- .values()
- .stream()
- .filter(t -> t.state() == Task.State.RUNNING || t.state() == Task.State.RESTORING)
- .filter(t -> !e.corruptedTaskWithChangelogs().containsKey(t.id()))
- .collect(Collectors.toSet())
- );
- taskManager.handleCorruption(e.corruptedTaskWithChangelogs());
+ try {
+ taskManager.commit(
+ taskManager.tasks()
+ .values()
+ .stream()
+ .filter(t -> t.state() == Task.State.RUNNING || t.state() == Task.State.RESTORING)
+ .filter(t -> !e.corruptedTaskWithChangelogs().containsKey(t.id()))
+ .collect(Collectors.toSet())
+ );
+ taskManager.handleCorruption(e.corruptedTaskWithChangelogs());
+ } catch (final TaskMigratedException taskMigrated) {
+ handleTaskMigrated(taskMigrated);
Review comment:
Should we also handle the corrupted tasks here (before this line), so that they can be already cleaned up before the next round? Or, alternatively, should we move `taskManager.handleCorruption(e.corruptedTaskWithChangelogs());` to before the attempted commit (it looks like it could be outside the try block as well).
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] mjsax commented on pull request #8667: KAFKA-9994: Handle task migrated inside corruption path
Posted by GitBox <gi...@apache.org>.
mjsax commented on pull request #8667:
URL: https://github.com/apache/kafka/pull/8667#issuecomment-628795711
Retest this please.
----------------------------------------------------------------
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:
users@infra.apache.org