ljz2051 commented on code in PR #24640:
URL: https://github.com/apache/flink/pull/24640#discussion_r1579435004
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/filemerging/FileMergingSnapshotManagerBase.java:
##########
@@ -575,8 +576,51 @@ private void createManagedDirectory(Path managedPath) {
@Override
public void close() throws IOException {}
+ // ------------------------------------------------------------------------
+ // restore
+ // ------------------------------------------------------------------------
+
+ @Override
+ public void restoreStateHandles(
+ long checkpointId, SubtaskKey subtaskKey,
Stream<SegmentFileStateHandle> stateHandles) {
+
+ Set<LogicalFile> uploadedLogicalFiles;
+ synchronized (lock) {
+ uploadedLogicalFiles =
+ uploadedStates.computeIfAbsent(checkpointId, id -> new
HashSet<>());
+ }
+
+ stateHandles.forEach(
+ fileHandle -> {
+ PhysicalFile physicalFile =
+ new PhysicalFile(
+ null,
+ fileHandle.getFilePath(),
+ physicalFileDeleter,
Review Comment:
IIUC, In the case of the job failover, need `FileMergingSnapshotManager` to
delete the physical file; but in the case of the job rescaling, need jobManager
to delete the physical file dir.
I will correct this logic. But i meet a problem where the TaskManager side
has no way of knowing if the recovery is a rescaling situation (especially for
the operator state rescaling).
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
