[hotfix] Make ForSt state backend use checkpoint directory as its primary storage by default

Author: Zakelly

docs/layouts/shortcodes/generated/forst_configuration.html

@@ -12,7 +12,7 @@
             <td><h5>state.backend.forst.cache.dir</h5></td>
             <td style="word-wrap: break-word;">(none)</td>
             <td>String</td>
-            <td>The directory where ForSt caches its SST files, fallback to state.backend.forst.local-dir/cache if not configured.</td>
+            <td>The directory where ForSt caches its SST files, fallback to the subdirectory of '/cache' under the value of 'state.backend.forst.local-dir' if not configured.</td>
         </tr>
         <tr>
             <td><h5>state.backend.forst.cache.reserve-size</h5></td>
@@ -99,10 +99,10 @@
             <td>The options factory class for users to add customized options in DBOptions and ColumnFamilyOptions for ForSt. If set, the ForSt state backend will load the class and apply configs to DBOptions and ColumnFamilyOptions after loading ones from 'ForStConfigurableOptions' and pre-defined options.</td>
         </tr>
         <tr>
-            <td><h5>state.backend.forst.remote-dir</h5></td>
-            <td style="word-wrap: break-word;">(none)</td>
+            <td><h5>state.backend.forst.primary-dir</h5></td>
+            <td style="word-wrap: break-word;">"checkpoint-dir"</td>
             <td>String</td>
-            <td>The remote directory where ForSt puts its SST files, fallback to state.backend.forst.local-dir if not configured. Recognized shortcut name is 'checkpoint-dir', which means that forst shares the directory with checkpoint.</td>
+            <td>The primary directory where ForSt puts its SST files. By default, it will be the same as the checkpoint directory. Recognized shortcut name is 'checkpoint-dir', which means that ForSt shares the directory with checkpoint, and 'local-dir', which means that ForSt will use the local directory of TaskManager.</td>
         </tr>
         <tr>
             <td><h5>state.backend.forst.timer-service.cache-size</h5></td>

docs/layouts/shortcodes/generated/state_backend_forst_section.html

@@ -12,7 +12,7 @@
             <td><h5>state.backend.forst.cache.dir</h5></td>
             <td style="word-wrap: break-word;">(none)</td>
             <td>String</td>
-            <td>The directory where ForSt caches its SST files, fallback to state.backend.forst.local-dir/cache if not configured.</td>
+            <td>The directory where ForSt caches its SST files, fallback to the subdirectory of '/cache' under the value of 'state.backend.forst.local-dir' if not configured.</td>
         </tr>
         <tr>
             <td><h5>state.backend.forst.cache.reserve-size</h5></td>
@@ -45,10 +45,10 @@
             <td>With partitioning, the index/filter block of an SST file is partitioned into smaller blocks with an additional top-level index on them. When reading an index/filter, only top-level index is loaded into memory. The partitioned index/filter then uses the top-level index to load on demand into the block cache the partitions that are required to perform the index/filter query. This option only has an effect when 'state.backend.forst.memory.managed' or 'state.backend.forst.memory.fixed-per-slot' are configured.</td>
         </tr>
         <tr>
-            <td><h5>state.backend.forst.remote-dir</h5></td>
-            <td style="word-wrap: break-word;">(none)</td>
+            <td><h5>state.backend.forst.primary-dir</h5></td>
+            <td style="word-wrap: break-word;">"checkpoint-dir"</td>
             <td>String</td>
-            <td>The remote directory where ForSt puts its SST files, fallback to state.backend.forst.local-dir if not configured. Recognized shortcut name is 'checkpoint-dir', which means that forst shares the directory with checkpoint.</td>
+            <td>The primary directory where ForSt puts its SST files. By default, it will be the same as the checkpoint directory. Recognized shortcut name is 'checkpoint-dir', which means that ForSt shares the directory with checkpoint, and 'local-dir', which means that ForSt will use the local directory of TaskManager.</td>
         </tr>
         <tr>
             <td><h5>state.backend.forst.timer-service.cache-size</h5></td>

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

@@ -27,8 +27,9 @@
 import org.apache.flink.configuration.description.Description;
 import org.apache.flink.configuration.description.TextElement;
 
+import static org.apache.flink.state.forst.ForStStateBackend.CHECKPOINT_DIR_AS_PRIMARY_SHORTCUT;
+import static org.apache.flink.state.forst.ForStStateBackend.LOCAL_DIR_AS_PRIMARY_SHORTCUT;
 import static org.apache.flink.state.forst.ForStStateBackend.PriorityQueueStateType.ForStDB;
-import static org.apache.flink.state.forst.ForStStateBackend.REMOTE_SHORTCUT_CHECKPOINT;
 
 /** Configuration options for the ForStStateBackend. */
 @Experimental
@@ -52,15 +53,17 @@ public class ForStOptions {
 
     /** The remote directory where ForSt puts its SST files. */
     @Documentation.Section(Documentation.Sections.STATE_BACKEND_FORST)
-    public static final ConfigOption<String> REMOTE_DIRECTORY =
-            ConfigOptions.key("state.backend.forst.remote-dir")
+    public static final ConfigOption<String> PRIMARY_DIRECTORY =
+            ConfigOptions.key("state.backend.forst.primary-dir")
                     .stringType()
-                    .noDefaultValue()
+                    .defaultValue(CHECKPOINT_DIR_AS_PRIMARY_SHORTCUT)
                     .withDescription(
                             String.format(
-                                    "The remote directory where ForSt puts its SST files, fallback to %s if not configured."
-                                            + " Recognized shortcut name is '%s', which means that forst shares the directory with checkpoint.",
-                                    LOCAL_DIRECTORIES.key(), REMOTE_SHORTCUT_CHECKPOINT));
+                                    "The primary directory where ForSt puts its SST files. By default, it will be the same as the checkpoint directory. "
+                                            + "Recognized shortcut name is '%s', which means that ForSt shares the directory with checkpoint, "
+                                            + "and '%s', which means that ForSt will use the local directory of TaskManager.",
+                                    CHECKPOINT_DIR_AS_PRIMARY_SHORTCUT,
+                                    LOCAL_DIR_AS_PRIMARY_SHORTCUT));
 
     @Documentation.Section(Documentation.Sections.STATE_BACKEND_FORST)
     public static final ConfigOption<String> CACHE_DIRECTORY =
@@ -69,7 +72,8 @@ public class ForStOptions {
                     .noDefaultValue()
                     .withDescription(
                             String.format(
-                                    "The directory where ForSt caches its SST files, fallback to %s/cache if not configured.",
+                                    "The directory where ForSt caches its SST files, fallback to the "
+                                            + "subdirectory of '/cache' under the value of '%s' if not configured.",
                                     LOCAL_DIRECTORIES.key()));
 
     @Documentation.Section(Documentation.Sections.STATE_BACKEND_FORST)

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

@@ -386,7 +386,11 @@ public void prepareDirectories() throws Exception {
                             remoteForStPath.toUri(),
                             localForStPath,
                             ForStFlinkFileSystem.getFileBasedCache(
-                                    cacheBasePath, cacheCapacity, cacheReservedSize, metricGroup));
+                                    cacheBasePath,
+                                    remoteForStPath,
+                                    cacheCapacity,
+                                    cacheReservedSize,
+                                    metricGroup));
         } else {
             forStFileSystem = null;
         }

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

@@ -95,7 +95,8 @@
 public class ForStStateBackend extends AbstractManagedMemoryStateBackend
         implements ConfigurableStateBackend {
 
-    public static final String REMOTE_SHORTCUT_CHECKPOINT = "checkpoint-dir";
+    public static final String CHECKPOINT_DIR_AS_PRIMARY_SHORTCUT = "checkpoint-dir";
+    public static final String LOCAL_DIR_AS_PRIMARY_SHORTCUT = "local-dir";
 
     private static final long serialVersionUID = 1L;
 
@@ -216,12 +217,15 @@ private ForStStateBackend(
         if (original.remoteForStDirectory != null) {
             this.remoteForStDirectory = original.remoteForStDirectory;
         } else {
-            String remoteDirStr = config.get(ForStOptions.REMOTE_DIRECTORY);
-            if (REMOTE_SHORTCUT_CHECKPOINT.equals(remoteDirStr)) {
+            String remoteDirStr = config.get(ForStOptions.PRIMARY_DIRECTORY);
+            if (CHECKPOINT_DIR_AS_PRIMARY_SHORTCUT.equals(remoteDirStr)) {
                 this.remoteForStDirectory = null;
                 this.remoteShareWithCheckpoint = true;
             } else {
-                this.remoteForStDirectory = remoteDirStr == null ? null : new Path(remoteDirStr);
+                this.remoteForStDirectory =
+                        remoteDirStr == null || LOCAL_DIR_AS_PRIMARY_SHORTCUT.equals(remoteDirStr)
+                                ? null
+                                : new Path(remoteDirStr);
             }
         }
 

flink-state-backends/flink-statebackend-forst/src/main/java/org/apache/flink/state/forst/fs/ForStFlinkFileSystem.java

@@ -108,11 +108,21 @@ public static ForStFlinkFileSystem get(URI uri, Path localBase, FileBasedCache f
     }
 
     public static FileBasedCache getFileBasedCache(
-            Path cacheBase, long cacheCapacity, long cacheReservedSize, MetricGroup metricGroup)
+            Path cacheBase,
+            Path remoteForStPath,
+            long cacheCapacity,
+            long cacheReservedSize,
+            MetricGroup metricGroup)
             throws IOException {
         if (cacheBase == null || cacheCapacity <= 0 && cacheReservedSize <= 0) {
             return null;
         }
+        if (cacheBase.getFileSystem().equals(remoteForStPath.getFileSystem())) {
+            LOG.info(
+                    "Skip creating ForSt cache "
+                            + "since the cache and primary path are on the same file system.");
+            return null;
+        }
         CacheLimitPolicy cacheLimitPolicy = null;
         if (cacheCapacity > 0 && cacheReservedSize > 0) {
             cacheLimitPolicy =

flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendConfigTest.java

@@ -69,6 +69,7 @@
 import java.util.Collection;
 import java.util.Collections;
 
+import static org.apache.flink.state.forst.ForStStateBackend.LOCAL_DIR_AS_PRIMARY_SHORTCUT;
 import static org.apache.flink.state.forst.ForStTestUtils.createKeyedStateBackend;
 import static org.hamcrest.CoreMatchers.anyOf;
 import static org.hamcrest.MatcherAssert.assertThat;
@@ -283,6 +284,7 @@ public void testCleanRelocatedDbLogs() throws Exception {
         conf.set(ForStConfigurableOptions.LOG_LEVEL, InfoLogLevel.DEBUG_LEVEL);
         conf.set(ForStConfigurableOptions.LOG_FILE_NUM, 4);
         conf.set(ForStConfigurableOptions.LOG_MAX_FILE_SIZE, MemorySize.parse("1kb"));
+        conf.set(ForStOptions.PRIMARY_DIRECTORY, LOCAL_DIR_AS_PRIMARY_SHORTCUT);
         final ForStStateBackend forStBackend =
                 new ForStStateBackend().configure(conf, getClass().getClassLoader());
         final String dbStoragePath = new Path(folder.toURI().toString()).toString();
@@ -742,10 +744,11 @@ public void testConfigureIllegalMemoryControlParameters() {
     }
 
     @Test
-    public void testRemoteDirectory() throws Exception {
+    public void testPrimaryDirectory() throws Exception {
         FileSystem.initialize(new Configuration(), null);
         Configuration configuration = new Configuration();
-        configuration.set(ForStOptions.REMOTE_DIRECTORY, tempFolder.newFolder().toURI().toString());
+        configuration.set(
+                ForStOptions.PRIMARY_DIRECTORY, tempFolder.newFolder().toURI().toString());
         ForStStateBackend forStStateBackend =
                 new ForStStateBackend().configure(configuration, null);
         ForStKeyedStateBackend<Integer> keyedBackend = null;
@@ -759,7 +762,7 @@ public void testRemoteDirectory() throws Exception {
                     keyedBackend
                             .getRemoteBasePath()
                             .toString()
-                            .startsWith(configuration.get(ForStOptions.REMOTE_DIRECTORY)));
+                            .startsWith(configuration.get(ForStOptions.PRIMARY_DIRECTORY)));
         } finally {
             if (keyedBackend != null) {
                 keyedBackend.dispose();

flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateBackendV2Test.java

@@ -54,8 +54,9 @@
 import static org.apache.flink.state.forst.ForStConfigurableOptions.USE_DELETE_FILES_IN_RANGE_DURING_RESCALING;
 import static org.apache.flink.state.forst.ForStConfigurableOptions.USE_INGEST_DB_RESTORE_MODE;
 import static org.apache.flink.state.forst.ForStOptions.LOCAL_DIRECTORIES;
-import static org.apache.flink.state.forst.ForStOptions.REMOTE_DIRECTORY;
-import static org.apache.flink.state.forst.ForStStateBackend.REMOTE_SHORTCUT_CHECKPOINT;
+import static org.apache.flink.state.forst.ForStOptions.PRIMARY_DIRECTORY;
+import static org.apache.flink.state.forst.ForStStateBackend.CHECKPOINT_DIR_AS_PRIMARY_SHORTCUT;
+import static org.apache.flink.state.forst.ForStStateBackend.LOCAL_DIR_AS_PRIMARY_SHORTCUT;
 import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
 
 /** Tests for the async keyed state backend part of {@link ForStStateBackend}. */
@@ -169,7 +170,9 @@ protected ConfigurableStateBackend getStateBackend() throws Exception {
             config.set(LOCAL_DIRECTORIES, tempFolderForForStLocal.toString());
         }
         if (hasRemoteDir) {
-            config.set(REMOTE_DIRECTORY, tempFolderForForstRemote.toString());
+            config.set(PRIMARY_DIRECTORY, tempFolderForForstRemote.toString());
+        } else {
+            config.set(PRIMARY_DIRECTORY, LOCAL_DIR_AS_PRIMARY_SHORTCUT);
         }
         config.set(USE_INGEST_DB_RESTORE_MODE, useIngestDbRestoreMode);
         config.set(USE_DELETE_FILES_IN_RANGE_DURING_RESCALING, useDeleteFileInRange);
@@ -207,7 +210,7 @@ void testRemoteDirShareCheckpointDir(boolean createJob) throws Exception {
 
         Configuration config = new Configuration();
         config.set(LOCAL_DIRECTORIES, tempFolderForForStLocal.toString());
-        config.set(REMOTE_DIRECTORY, REMOTE_SHORTCUT_CHECKPOINT);
+        config.set(PRIMARY_DIRECTORY, CHECKPOINT_DIR_AS_PRIMARY_SHORTCUT);
         config.set(CheckpointingOptions.CREATE_CHECKPOINT_SUB_DIR, createJob);
 
         checkpointStorage =

flink-state-backends/flink-statebackend-forst/src/test/java/org/apache/flink/state/forst/ForStStateTestBase.java

@@ -61,7 +61,7 @@ public class ForStStateTestBase {
     public void setup(@TempDir File temporaryFolder) throws IOException {
         FileSystem.initialize(new Configuration(), null);
         Configuration configuration = new Configuration();
-        configuration.set(ForStOptions.REMOTE_DIRECTORY, temporaryFolder.toURI().toString());
+        configuration.set(ForStOptions.PRIMARY_DIRECTORY, temporaryFolder.toURI().toString());
         ForStStateBackend forStStateBackend =
                 new ForStStateBackend().configure(configuration, null);