You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "zuston (via GitHub)" <gi...@apache.org> on 2026/07/03 10:04:07 UTC
[PR] [tiering] introduce the option of 'lake.tiering.io.tmpdir' to avoid disk exhaustion [fluss]
zuston opened a new pull request, #3569:
URL: https://github.com/apache/fluss/pull/3569
### Purpose
Linked issue: close #3568
### Brief change log
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [tiering] introduce the option of `lake.tiering.io.tmpdir` to avoid disk exhaustion [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #3569:
URL: https://github.com/apache/fluss/pull/3569#discussion_r3607721532
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSource.java:
##########
@@ -117,7 +118,11 @@ public SourceReader<TableBucketWriteResult<WriteResult>, TieringSplit> createRea
getClientScannerIoTmpDir(flussConf, sourceReaderContext.getConfiguration()));
Connection connection = ConnectionFactory.createConnection(flussConf);
return new TieringSourceReader<>(
- elementsQueue, sourceReaderContext, connection, lakeTieringFactory);
+ elementsQueue,
+ sourceReaderContext,
+ connection,
+ lakeTieringFactory,
+ getLakeTieringIoTmpDir(flussConf, sourceReaderContext.getConfiguration()));
Review Comment:
Could we pass or read this option from `lakeTieringConfig` instead? `FlussLakeTiering` puts CLI keys starting with `lake.tiering.` into `lakeTieringConfig`, while `flussConf` only contains `--fluss.*` keys after removing the `fluss.` prefix. `LakeTieringJobBuilder` currently passes only `flussConfig` to `TieringSource`, so a standard `--lake.tiering.io.tmpdir /data/tmp` argument is absent here and always falls back to the Flink/JVM temporary directory. Please wire `lakeTieringConfig` (or the resolved option) into the source and add a test that starts from the CLI arguments.
##########
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java:
##########
@@ -2285,6 +2285,15 @@ public class ConfigOptions {
+ ConfigOptions.TABLE_DATALAKE_AUTO_EXPIRE_SNAPSHOT
+ " is false.");
+ public static final ConfigOption<String> LAKE_TIERING_IO_TMP_DIR =
Review Comment:
Could we also add this user-facing option to the `Datalake Tiering Service Options` table in `website/docs/maintenance/tiered-storage/lakehouse-storage.md`, together with a CLI example? That table currently lists only `lake.tiering.auto-expire-snapshot`, so users have no documentation showing how to set the new `lake.tiering.io.tmpdir` option.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConnectorOptionsUtils.java:
##########
@@ -202,14 +203,26 @@ 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 getFlinkIoTmpDir(flinkConfig);
}
return flussConf.getString(CLIENT_SCANNER_IO_TMP_DIR);
}
+ public static String getFlinkIoTmpDir(
+ org.apache.flink.configuration.Configuration flinkConfig) {
+ if (flinkConfig.contains(TMP_DIRS)) {
+ return new File(flinkConfig.get(CoreOptions.TMP_DIRS), "/fluss").getAbsolutePath();
Review Comment:
`CoreOptions.TMP_DIRS` can contain multiple directories separated by comma, pipe, or the platform path separator. This code treats the entire raw value as one `File` parent, and Paimon later calls `IOManager.create(String)`, which also treats it as one directory. For example, `/disk1/tmp,/disk2/tmp` is handled as one combined path instead of two temporary directories. Please parse the value with Flink `ConfigurationUtils.parseTempDirectories()` and either select one valid directory or pass the resulting directories to Paimon as an array. A multiple-directory test would catch this case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [tiering] introduce the option of `lake.tiering.io.tmpdir` to avoid disk exhaustion [fluss]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #3569:
URL: https://github.com/apache/fluss/pull/3569#discussion_r3519127621
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/mergetree/MergeTreeWriter.java:
##########
@@ -31,17 +31,13 @@
import javax.annotation.Nullable;
import java.util.List;
-import java.util.Map;
import static org.apache.fluss.lake.paimon.tiering.PaimonLakeTieringFactory.FLUSS_LAKE_TIERING_COMMIT_USER;
import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toRowKind;
/** A {@link RecordWriter} to write to Paimon's primary-key table. */
public class MergeTreeWriter extends RecordWriter<KeyValue> {
- // the option key to configure the temporary directory used by fluss tiering
- private static final String FLUSS_TIERING_TMP_DIR_KEY = "fluss.tiering.io-tmpdir";
Review Comment:
We don't need to reserve this option that it haven't been valid.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [tiering] introduce the option of `lake.tiering.io.tmpdirs` to avoid disk exhaustion [fluss]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #3569:
URL: https://github.com/apache/fluss/pull/3569#issuecomment-5188712927
thanks for your review! @luoyuxia
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [tiering] introduce the option of `lake.tiering.io.tmpdir` to avoid disk exhaustion [fluss]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3569:
URL: https://github.com/apache/fluss/pull/3569#discussion_r3607514253
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConnectorOptionsUtils.java:
##########
@@ -202,14 +203,26 @@ 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 getFlinkIoTmpDir(flinkConfig);
}
return flussConf.getString(CLIENT_SCANNER_IO_TMP_DIR);
}
+ public static String getFlinkIoTmpDir(
+ org.apache.flink.configuration.Configuration flinkConfig) {
+ if (flinkConfig.contains(TMP_DIRS)) {
+ return new File(flinkConfig.get(CoreOptions.TMP_DIRS), "/fluss").getAbsolutePath();
+ }
+ return System.getProperty("java.io.tmpdir") + "/fluss";
+ }
Review Comment:
`getFlinkIoTmpDir` builds the child path using `new File(parent, "/fluss")`. Because the child starts with a leading '/', Java treats it as an absolute path and ignores the parent directory on Unix-like systems, which can yield `/fluss` instead of `<tmp>/fluss` and defeats the intent of using Flink's tmp dirs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [tiering] introduce the option of `lake.tiering.io.tmpdirs` to avoid disk exhaustion [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia merged PR #3569:
URL: https://github.com/apache/fluss/pull/3569
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [tiering] introduce the option of `lake.tiering.io.tmpdir` to avoid disk exhaustion [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #3569:
URL: https://github.com/apache/fluss/pull/3569#discussion_r3607721532
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSource.java:
##########
@@ -117,7 +118,11 @@ public SourceReader<TableBucketWriteResult<WriteResult>, TieringSplit> createRea
getClientScannerIoTmpDir(flussConf, sourceReaderContext.getConfiguration()));
Connection connection = ConnectionFactory.createConnection(flussConf);
return new TieringSourceReader<>(
- elementsQueue, sourceReaderContext, connection, lakeTieringFactory);
+ elementsQueue,
+ sourceReaderContext,
+ connection,
+ lakeTieringFactory,
+ getLakeTieringIoTmpDir(flussConf, sourceReaderContext.getConfiguration()));
Review Comment:
Could we pass or read this option from `lakeTieringConfig` instead? `FlussLakeTiering` puts CLI keys starting with `lake.tiering.` into `lakeTieringConfig`, while `flussConf` only contains `--fluss.*` keys after removing the `fluss.` prefix. `LakeTieringJobBuilder` currently passes only `flussConfig` to `TieringSource`, so a standard `--lake.tiering.io.tmpdir /data/tmp` argument is absent here and always falls back to the Flink/JVM temporary directory. Please wire `lakeTieringConfig` (or the resolved option) into the source.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [tiering] introduce the option of `lake.tiering.io.tmpdir` to avoid disk exhaustion [fluss]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #3569:
URL: https://github.com/apache/fluss/pull/3569#issuecomment-4875070640
link the PR of #3193 . Could you help review this? @luoyuxia
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [tiering] introduce the option of `lake.tiering.io.tmpdir` to avoid disk exhaustion [fluss]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #3569:
URL: https://github.com/apache/fluss/pull/3569#issuecomment-4889502861
cc @luoyuxia
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [tiering] introduce the option of `lake.tiering.io.tmpdir` to avoid disk exhaustion [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #3569:
URL: https://github.com/apache/fluss/pull/3569#discussion_r3607744786
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceReader.java:
##########
@@ -61,7 +63,23 @@ public TieringSourceReader(
SourceReaderContext context,
Connection connection,
LakeTieringFactory<WriteResult, ?> lakeTieringFactory) {
- this(elementsQueue, context, connection, lakeTieringFactory, DEFAULT_POLL_TIMEOUT);
+ this(elementsQueue, context, connection, lakeTieringFactory, (String) null);
+ }
+
+ public TieringSourceReader(
+ FutureCompletingBlockingQueue<RecordsWithSplitIds<TableBucketWriteResult<WriteResult>>>
+ elementsQueue,
+ SourceReaderContext context,
+ Connection connection,
+ LakeTieringFactory<WriteResult, ?> lakeTieringFactory,
+ @Nullable String ioTmpDir) {
Review Comment:
When it'll be null? Can we remove Nullable notation?
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceReader.java:
##########
Review Comment:
public TieringSourceReader(
FutureCompletingBlockingQueue<RecordsWithSplitIds<TableBucketWriteResult<WriteResult>>>
elementsQueue,
SourceReaderContext context,
Connection connection,
LakeTieringFactory<WriteResult, ?> lakeTieringFactory) {
this(elementsQueue, context, connection, lakeTieringFactory, (String) null);
}
is not called anymore, seems can be removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [tiering] introduce the option of `lake.tiering.io.tmpdir` to avoid disk exhaustion [fluss]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #3569:
URL: https://github.com/apache/fluss/pull/3569#discussion_r3611743224
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConnectorOptionsUtils.java:
##########
@@ -202,14 +203,26 @@ 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 getFlinkIoTmpDir(flinkConfig);
}
return flussConf.getString(CLIENT_SCANNER_IO_TMP_DIR);
}
+ public static String getFlinkIoTmpDir(
+ org.apache.flink.configuration.Configuration flinkConfig) {
+ if (flinkConfig.contains(TMP_DIRS)) {
+ return new File(flinkConfig.get(CoreOptions.TMP_DIRS), "/fluss").getAbsolutePath();
Review Comment:
got it. But I think maybe we should merge this PR #3617 firstly, and then I will change this on that baseline.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org