jiexray commented on code in PR #21822:
URL: https://github.com/apache/flink/pull/21822#discussion_r1134272846
##########
flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackend.java:
##########
@@ -455,12 +467,23 @@ private SnapshotResult<ChangelogStateBackendHandle>
buildSnapshotResult(
checkpointId,
changelogStateBackendStateCopy.materializationID,
persistedSizeOfThisCheckpoint);
- return SnapshotResult.withLocalState(
- jmHandle,
+ ChangelogStateBackendLocalHandle localHandle =
new ChangelogStateBackendLocalHandle(
changelogStateBackendStateCopy.getLocalMaterializedSnapshot(),
localDeltaCopy,
- jmHandle));
+ jmHandle);
+ // register local handle to localRegistry
+ for (ChangelogStateHandle handle : localDeltaCopy) {
+ if (handle instanceof ChangelogStateHandleStreamImpl) {
+ ((ChangelogStateHandleStreamImpl) handle)
+ .getHandlesAndOffsets()
+ .forEach(
+ tuple ->
+ localChangelogRegistry.register(
Review Comment:
Why do we need to register handle to LocalChangelogRegistry here? Now, one
handle may be registered twice. The first time is here (pre report snapshot
result to JM). The second time is the `confirm` (post receive completion
notification from JM).
I am not sure this will affect the `LocalChangelogRegistryImpl#prune()`. For
example, If `CL-File-A` is used by chk-1. chk-2 reuses `CL-file-A`, and
registers `CL-file-A` to LocalChangelogRegistry here. Some time later, chk-2 is
aborted, and `LocalChangelogRegistryImpl#prune()` will execute the following
logic to delete `CL-file-A`, thereby destroying chk-1.
```
public void prune(long checkpointID) {
Set<StreamStateHandle> handles =
handleToLastUsedCheckpointID.values().stream()
.filter(tuple -> tuple.f1 == checkpointID) // !HERE:
tuple.f1 is updated at buildSnapshotResult()
.map(tuple -> tuple.f0)
.collect(Collectors.toSet());
for (StreamStateHandle handle : handles) {
scheduleAsyncDelete(handle);
}
}
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
