[FLINK-38336][state/forst] Avoid data copy during failover for ForSt statebackend #27042

AlexYinHan · 2025-09-25T12:54:24Z

What is the purpose of the change

This PR fixes FLINK-38336, by allowing ForSt statebackend to reuse the restored files in failover scenario.

Brief change log

Reorganize the local/remote path of ForSt into a ForStPathContainer
Determine whether we are in a failover scenario by comparing the DB path and the file paths stored in StateHandles
Use ReusableDataTransferStrategy if we are in a failover scenario

Verifying this change

This change added tests and can be verified as follows:

DataTransferStrategyTest#testBuildingStrategyAsExpected

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (no)

flinkbot · 2025-09-25T13:00:15Z

CI report:

a8460bc Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

…statebackend

Zakelly · 2025-09-29T06:14:07Z

...s/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStStateBackend.java

-                                .getSharedStateDirectory();
+                FsCheckpointStorageAccess fsCheckpointStorageAccess =
+                        (FsCheckpointStorageAccess) env.getCheckpointStorageAccess();
+                remoteJobPath = fsCheckpointStorageAccess.getCheckpointsDirectory();


We can only share state from the shared directory, so what exactly the remoteJobPath means?

Zakelly · 2025-09-29T06:22:09Z

...rst/src/main/java/org/apache/flink/state/forst/datatransfer/DataTransferStrategyBuilder.java

                            : new CopyDataTransferStrategy(forStFlinkFileSystem);
            LOG.info(
-                    "Build DataTransferStrategy for Restore: {}, forStFlinkFileSystem: {}, cpSharedFs:{}, recoveryClaimMode:{}",
+                    "Build DataTransferStrategy for Restore: {}, forStFlinkFileSystem: {}, cpSharedFs:{}, isDbUnderSameJobPathFromRestore:{}, recoveryClaimMode:{}",


I'd suggest we make this log as debug level one, as well the one in buildForSnapshot

davidradl · 2025-10-03T11:41:10Z

.../flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStPathContainer.java

+            @Nullable Path remoteBasePath) {
+        this.localJobPath = localJobPath;
+        this.localBasePath = localBasePath;
+        this.localForStPath = localBasePath != null ? new Path(localBasePath, DB_DIR_STRING) : null;


we have localJobPath, localBasePath and localForStPath which is derived from localBasePath.

I was expecting localJobpath to be a subfolder of localBasePath, is this true? Is this the bean to validate that?

I see localBasePath can be null, in this case localForStPath is set to null, but localJobPath can have a value. What does this mean? The code indicates that it is possible to run without Forst but have a . Have I understood this correctly?

davidradl · 2025-10-03T11:43:31Z

.../flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStPathContainer.java

+
+import javax.annotation.Nullable;
+
+/** Container for ForSt paths. */


it would be worth describing each path here. The text talks of Forst paths , but there are only 2 of the 6 that mention Forst - localForStPath and remoteForStPath

davidradl · 2025-10-03T11:43:49Z

.../flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStPathContainer.java

+    private static final Logger LOG = LoggerFactory.getLogger(ForStResourceContainer.class);
+    public static final String DB_DIR_STRING = "db";
+
+    @Nullable final Path localJobPath;


why is this not private?

davidradl · 2025-10-03T11:44:36Z

.../flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStPathContainer.java

+    }
+
+    @Override
+    public String toString() {


maybe add equals and hashcode standard methods for beans

davidradl · 2025-10-03T11:47:09Z

...nk-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java

+        if (pathContainer.getRemoteForStPath() != null
+                && pathContainer.getLocalForStPath() != null) {
+            if (cacheBasePath == null && pathContainer.getLocalBasePath() != null) {
+                cacheBasePath = new Path(pathContainer.getLocalBasePath().getPath(), "cache");


I assume thr log is incorrect, not should be now

davidradl · 2025-10-03T11:48:16Z

...nk-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java

    public void clearDirectories() throws Exception {
-        if (remoteBasePath != null) {
-            forStFileSystem.delete(remoteBasePath, true);
+        if (pathContainer.getRemoteBasePath() != null) {


nit: use a variable for pathContainer.getRemoteBasePath() and pathContainer.getLocalBasePath()

davidradl · 2025-10-03T11:50:18Z

...nk-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStResourceContainer.java

    public void forceClearRemoteDirectories() throws Exception {
-        if (remoteBasePath != null && remotePathNewlyCreated) {
-            clearDirectories(remoteBasePath);
+        if (pathContainer.getRemoteBasePath() != null && remotePathNewlyCreated) {


The forceClear does not seem to force anything, in fact it will only clear the directories in a specific case when remotePathNewlyCreated is set.

davidradl · 2025-10-03T11:57:58Z

...ackend-forst/src/main/java/org/apache/flink/state/forst/sync/ForStSyncKeyedStateBackend.java

-                        optionsContainer.getLocalBasePath(),
-                        optionsContainer.getRemoteBasePath(),
-                        ex);
+                LOG.warn("Could not delete ForSt: {}.", optionsContainer.getPathContainer(), ex);


nit: it seems we lose information here. I read the log entry text before Could not delete ForSt local working directory {} to means it is only the local directory that could not be deleted and we put out the remote directory for information. It would be good to point to the actual folder that could not be deleted in the new log message - rather than list all the folders and not know which could not be deleted.

davidradl · 2025-10-03T11:59:04Z

...ends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStExtension.java

        Path instanceForStPath =
-                resourceContainer.getRemoteForStPath() == null
-                        ? resourceContainer.getLocalForStPath()
+                resourceContainer.getPathContainer().getRemoteForStPath() == null


nit: variable for resourceContainer.getPathContainer()

davidradl · 2025-10-03T12:00:30Z

...-forst/src/test/java/org/apache/flink/state/forst/datatransfer/DataTransferStrategyTest.java

+                RecoveryClaimMode.CLAIM,
+                CopyDataTransferStrategy.class);
+
+        testRestoreStrategyAsExpected(


parameterized test for all these permutations?

davidradl · 2025-10-03T12:20:00Z

...nk-statebackend-forst/src/main/java/org/apache/flink/state/forst/ForStKeyedStateBackend.java

    @VisibleForTesting
    Path getLocalBasePath() {
-        return optionsContainer.getLocalBasePath();
+        return optionsContainer.getPathContainer().getLocalBasePath();


is optionsContainer.getPathContainer() ever null?

AlexYinHan force-pushed the yh/dev_38336 branch from 77b41af to 9c4c31e Compare September 26, 2025 03:56

[FLINK-38336][state/forst] Avoid data copy during failover for ForSt …

a8460bc

…statebackend

AlexYinHan force-pushed the yh/dev_38336 branch from 9c4c31e to a8460bc Compare September 29, 2025 03:25

Zakelly reviewed Sep 29, 2025

View reviewed changes

davidradl reviewed Oct 3, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[FLINK-38336][state/forst] Avoid data copy during failover for ForSt statebackend #27042

[FLINK-38336][state/forst] Avoid data copy during failover for ForSt statebackend #27042

AlexYinHan commented Sep 25, 2025

Uh oh!

flinkbot commented Sep 25, 2025 •

edited

Loading

Uh oh!

Zakelly Sep 29, 2025

Uh oh!

Zakelly Sep 29, 2025

Uh oh!

davidradl Oct 3, 2025

Uh oh!

davidradl Oct 3, 2025

Uh oh!

davidradl Oct 3, 2025

Uh oh!

davidradl Oct 3, 2025

Uh oh!

davidradl Oct 3, 2025

Uh oh!

davidradl Oct 3, 2025

Uh oh!

davidradl Oct 3, 2025

Uh oh!

davidradl Oct 3, 2025

Uh oh!

davidradl Oct 3, 2025

Uh oh!

davidradl Oct 3, 2025

Uh oh!

davidradl Oct 3, 2025

Uh oh!

Uh oh!


		import javax.annotation.Nullable;

		/** Container for ForSt paths. */

[FLINK-38336][state/forst] Avoid data copy during failover for ForSt statebackend #27042

Are you sure you want to change the base?

[FLINK-38336][state/forst] Avoid data copy during failover for ForSt statebackend #27042

Conversation

AlexYinHan commented Sep 25, 2025

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

flinkbot commented Sep 25, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

flinkbot commented Sep 25, 2025 •

edited

Loading