You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by yu...@apache.org on 2026/08/04 09:40:52 UTC
(fluss) branch main updated: [flink] respect flink io tmp multi dirs to fix disk exhaustion (#3617)
This is an automated email from the ASF dual-hosted git repository.
luoyuxia pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new b39c61ed6 [flink] respect flink io tmp multi dirs to fix disk exhaustion (#3617)
b39c61ed6 is described below
commit b39c61ed62e123be3449ead95eb6eaae30507460
Author: Junfan Zhang <zu...@apache.org>
AuthorDate: Tue Aug 4 17:40:46 2026 +0800
[flink] respect flink io tmp multi dirs to fix disk exhaustion (#3617)
---
.../org/apache/fluss/flink/source/FlinkSource.java | 7 +++--
.../fluss/flink/tiering/source/TieringSource.java | 11 +++----
.../flink/utils/FlinkConnectorOptionsUtils.java | 19 ++++++------
.../flink/utils/FlinkConnectorOptionsUtilTest.java | 36 +++++++++++++++++++++-
4 files changed, 54 insertions(+), 19 deletions(-)
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkSource.java
index 227db3fab..b6cc0049c 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkSource.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkSource.java
@@ -288,9 +288,10 @@ public class FlinkSource<OUT>
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 class FlinkSource<OUT>
return new FlinkSourceReader<>(
elementsQueue,
- flussConf,
+ readerConf,
tablePath,
sourceOutputType,
context,
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSource.java
index 16fc0a6db..b852fd635 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSource.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSource.java
@@ -114,13 +114,12 @@ public class TieringSource<WriteResult>
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);
}
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConnectorOptionsUtils.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConnectorOptionsUtils.java
index 249e1be08..cb14c1abe 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConnectorOptionsUtils.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConnectorOptionsUtils.java
@@ -23,7 +23,7 @@ import org.apache.fluss.flink.FlinkConnectorOptions.ScanStartupMode;
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.List;
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 class FlinkConnectorOptionsUtils {
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. * */
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConnectorOptionsUtilTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConnectorOptionsUtilTest.java
index 0e06dc101..b2f12ba90 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConnectorOptionsUtilTest.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConnectorOptionsUtilTest.java
@@ -22,7 +22,10 @@ import org.apache.fluss.config.Configuration;
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 @@ class FlinkConnectorOptionsUtilTest {
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 @@ class FlinkConnectorOptionsUtilTest {
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());
+ }
}