[FLINK-37021][state/forst] Support reusing files in LEGACY recovery m…

…ode. (#26142)
apache · Feb 12, 2025 · e299645 · e299645
1 parent 4aacf33
commit e299645
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 5 deletions.
diff --git a/.../src/main/java/org/apache/flink/state/forst/datatransfer/DataTransferStrategyBuilder.java b/.../src/main/java/org/apache/flink/state/forst/datatransfer/DataTransferStrategyBuilder.java
@@ -125,7 +125,7 @@ public static DataTransferStrategy buildForRestore(
         if (forStFlinkFileSystem == null
                 || cpSharedFs == null
                 || !forStFlinkFileSystem.getUri().equals(cpSharedFs.getUri())
-                || recoveryClaimMode != RecoveryClaimMode.CLAIM) {
+                || recoveryClaimMode == RecoveryClaimMode.NO_CLAIM) {
             strategy =
                     forStFlinkFileSystem == null
                             ? new CopyDataTransferStrategy()

diff --git a/...rst/src/test/java/org/apache/flink/state/forst/datatransfer/DataTransferStrategyTest.java b/...rst/src/test/java/org/apache/flink/state/forst/datatransfer/DataTransferStrategyTest.java
@@ -109,8 +109,6 @@ static class DBFilesContainer {
 
         Map<String, Path> dbFilePaths = new HashMap<>();
 
-        FileOwnershipDecider fileOwnershipDecider;
-
         DBFilesContainer(Path dbLocalBase, Path dbRemoteBase) throws IOException {
             realFileSystem = LocalFileSystem.getLocalFileSystem();
 
@@ -133,8 +131,6 @@ static class DBFilesContainer {
                             4096);
             tmpResourcesRegistry = new CloseableRegistry();
             closeableRegistry = new CloseableRegistry();
-
-            this.fileOwnershipDecider = fileOwnershipDecider;
         }
 
         private void createDbFiles(List<String> fileNames) throws IOException {
@@ -480,6 +476,9 @@ void testBuildingStrategyAsExpected() throws IOException {
         testRestoreStrategyAsExpected(
                 forStFlinkFileSystem, RecoveryClaimMode.NO_CLAIM, CopyDataTransferStrategy.class);
 
+        testRestoreStrategyAsExpected(
+                forStFlinkFileSystem, RecoveryClaimMode.LEGACY, ReusableDataTransferStrategy.class);
+
         testRestoreStrategyAsExpected(
                 null, RecoveryClaimMode.CLAIM, CopyDataTransferStrategy.class);