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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
