Skip to content

KAFKA-10362: When resuming Streams active task with EOS, the checkpoint file is deleted #9247

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
guozhangwang merged 4 commits into from
Oct 7, 2020
Merged

KAFKA-10362: When resuming Streams active task with EOS, the checkpoint file is deleted #9247

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
2ba81ca
When resuming Streams active task with EOS, the checkpoint file is de…
Sep 2, 2020
aae073b
changed error message of IO exception
Sep 2, 2020
70a7b2c
Merge branch 'trunk' into KAFKA-10362
Sep 12, 2020
489b034
unit test cases added
Sep 16, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions ...ams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -662,4 +662,10 @@ public TopicPartition registeredChangelogPartitionFor(final String storeName) {
public String changelogFor(final String storeName) {
return storeToChangelogTopic.get(storeName);
}

public void deleteCheckPointFile() throws IOException {
Comment thread
DOJI45 marked this conversation as resolved.
Outdated
if (eosEnabled) {
checkpointFile.delete();
}
}
}
9 changes: 9 additions & 0 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -332,6 +332,15 @@ public void resume() {
case SUSPENDED:
// just transit the state without any logical changes: suspended and restoring states
// are not actually any different for inner modules

// Deleting checkpoint file before transition to RESTORING state (KAFKA-10362)
try {
stateMgr.deleteCheckPointFile();
log.debug("Deleted check point file");
Comment thread
DOJI45 marked this conversation as resolved.
Outdated
} catch (final IOException ioe) {
log.error("Encountered error while deleting the checkpoint file due to this exception", ioe);
}

transitionTo(State.RESTORING);
log.info("Resumed to restoring state");

Expand Down