[flink] respect flink io tmp multi dirs to fix disk exhaustion

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkSource.java

@@ -288,9 +288,10 @@ public SourceReader<OUT, SourceSplitBase> createReader(SourceReaderContext conte
         FlinkSourceReaderMetrics flinkSourceReaderMetrics =
                 new FlinkSourceReaderMetrics(context.metricGroup());
 
-        flussConf.set(
+        Configuration readerConf = new Configuration(flussConf);
+        readerConf.set(
                 CLIENT_SCANNER_IO_TMP_DIR,
-                getClientScannerIoTmpDir(flussConf, context.getConfiguration()));
+                getClientScannerIoTmpDir(readerConf, context.getConfiguration()));
         deserializationSchema.open(
                 new DeserializerInitContextImpl(
                         context.metricGroup().addGroup("deserializer"),
@@ -300,7 +301,7 @@ public SourceReader<OUT, SourceSplitBase> createReader(SourceReaderContext conte
 
         return new FlinkSourceReader<>(
                 elementsQueue,
-                flussConf,
+                readerConf,
                 tablePath,
                 sourceOutputType,
                 context,

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSource.java

@@ -114,13 +114,12 @@ public SourceReader<TableBucketWriteResult<WriteResult>, TieringSplit> createRea
             SourceReaderContext sourceReaderContext) {
         FutureCompletingBlockingQueue<RecordsWithSplitIds<TableBucketWriteResult<WriteResult>>>
                 elementsQueue = new FutureCompletingBlockingQueue<>();
-        Configuration tieringReaderConf = new Configuration(flussConf);
-        tieringReaderConf.set(
+        Configuration readerConf = new Configuration(flussConf);
+        readerConf.set(CLIENT_SCANNER_LOG_READ_PREFERENCE, REMOTE_FIRST);
+        readerConf.set(
                 CLIENT_SCANNER_IO_TMP_DIR,
-                getClientScannerIoTmpDir(
-                        tieringReaderConf, sourceReaderContext.getConfiguration()));
-        tieringReaderConf.set(CLIENT_SCANNER_LOG_READ_PREFERENCE, REMOTE_FIRST);
-        Connection connection = ConnectionFactory.createConnection(tieringReaderConf);
+                getClientScannerIoTmpDir(readerConf, sourceReaderContext.getConfiguration()));
+        Connection connection = ConnectionFactory.createConnection(readerConf);
         return new TieringSourceReader<>(
                 elementsQueue, sourceReaderContext, connection, lakeTieringFactory);
     }

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConnectorOptionsUtils.java

@@ -23,7 +23,7 @@
 import org.apache.fluss.flink.sink.shuffle.DistributionMode;
 import org.apache.fluss.metadata.MergeEngineType;
 
-import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.configuration.ConfigurationUtils;
 import org.apache.flink.configuration.ReadableConfig;
 import org.apache.flink.table.api.ValidationException;
 import org.apache.flink.table.api.config.TableConfigOptions;
@@ -41,7 +41,6 @@
 import java.util.Optional;
 import java.util.stream.Collectors;
 
-import static org.apache.flink.configuration.CoreOptions.TMP_DIRS;
 import static org.apache.fluss.config.ConfigOptions.CLIENT_SCANNER_IO_TMP_DIR;
 import static org.apache.fluss.flink.FlinkConnectorOptions.SCAN_SPLIT_ASSIGNMENT_BATCH_SIZE;
 import static org.apache.fluss.flink.FlinkConnectorOptions.SCAN_STARTUP_MODE;
@@ -201,13 +200,15 @@ public static long parseTimestamp(String timestampStr, String optionKey, ZoneId
 
     public static String getClientScannerIoTmpDir(
             Configuration flussConf, org.apache.flink.configuration.Configuration flinkConfig) {
-        if (!flussConf.contains(CLIENT_SCANNER_IO_TMP_DIR)) {
-            if (flinkConfig.contains(TMP_DIRS)) {
-                // pass flink io tmp dir to fluss client.
-                return new File(flinkConfig.get(CoreOptions.TMP_DIRS), "/fluss").getAbsolutePath();
-            }
-        }
-        return flussConf.getString(CLIENT_SCANNER_IO_TMP_DIR);
+        return flussConf
+                .getOptional(CLIENT_SCANNER_IO_TMP_DIR)
+                .orElseGet(
+                        () ->
+                                new File(
+                                                ConfigurationUtils.getRandomTempDirectory(
+                                                        flinkConfig),
+                                                "fluss")
+                                        .getAbsolutePath());
     }
 
     /** Fluss startup options. * */

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConnectorOptionsUtilTest.java

@@ -22,7 +22,10 @@
 import org.apache.flink.table.api.ValidationException;
 import org.junit.jupiter.api.Test;
 
+import java.io.File;
 import java.time.ZoneId;
+import java.util.HashSet;
+import java.util.Set;
 import java.util.TimeZone;
 
 import static org.apache.flink.configuration.CoreOptions.TMP_DIRS;
@@ -96,7 +99,7 @@ void testGetClientScannerIoTmpDir() {
                         FlinkConnectorOptionsUtils.getClientScannerIoTmpDir(
                                 new Configuration(),
                                 new org.apache.flink.configuration.Configuration()))
-                .isEqualTo(property + "/fluss");
+                .isEqualTo(new File(property, "fluss").getAbsolutePath());
 
         // only replace when flussConfig not contains CLIENT_SCANNER_IO_TMP_DIR while flinkConfig
         // contains TMP_DIRS.
@@ -105,4 +108,35 @@ void testGetClientScannerIoTmpDir() {
                                 new Configuration(), flinkConfig))
                 .isEqualTo("/flink_tmp_dir/fluss");
     }
+
+    @Test
+    void testGetClientScannerIoTmpDirForMultipleReadersOnSameTaskManager() {
+        org.apache.flink.configuration.Configuration commaSeparatedFlinkConfig =
+                new org.apache.flink.configuration.Configuration()
+                        .set(TMP_DIRS, "/flink_tmp_dir_0,/flink_tmp_dir_1");
+
+        Set<String> selectedDirectories = new HashSet<>();
+        for (int i = 0; i < 100; i++) {
+            selectedDirectories.add(
+                    FlinkConnectorOptionsUtils.getClientScannerIoTmpDir(
+                            new Configuration(), commaSeparatedFlinkConfig));
+        }
+        assertThat(selectedDirectories)
+                .containsExactlyInAnyOrder(
+                        new File("/flink_tmp_dir_0", "fluss").getAbsolutePath(),
+                        new File("/flink_tmp_dir_1", "fluss").getAbsolutePath());
+
+        org.apache.flink.configuration.Configuration pathSeparatorFlinkConfig =
+                new org.apache.flink.configuration.Configuration()
+                        .set(
+                                TMP_DIRS,
+                                "/flink_tmp_dir_0" + File.pathSeparator + "/flink_tmp_dir_1");
+
+        assertThat(
+                        FlinkConnectorOptionsUtils.getClientScannerIoTmpDir(
+                                new Configuration(), pathSeparatorFlinkConfig))
+                .isIn(
+                        new File("/flink_tmp_dir_0", "fluss").getAbsolutePath(),
+                        new File("/flink_tmp_dir_1", "fluss").getAbsolutePath());
+    }
 }