refactor tiering options

luoyuxia · luoyuxia · commit 9ae01ad0a056 · 2025-12-25T17:54:52.000+08:00
diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
@@ -1811,6 +1811,23 @@ public class ConfigOptions {
                                     + ConfigOptions.TABLE_DATALAKE_AUTO_EXPIRE_SNAPSHOT
                                     + " is false.");
 
+    public static final ConfigOption<Duration> LAKE_TIERING_TABLE_DURATION_MAX =
+            key("lake.tiering.table.duration.max")
+                    .durationType()
+                    .defaultValue(Duration.ofMinutes(30))
+                    .withDescription(
+                            "The maximum duration for tiering a single table. If tiering a table exceeds this duration, "
+                                    + "it will be force completed: the tiering will be finalized and committed to the data lake "
+                                    + "(e.g., Paimon) immediately, even if they haven't reached their desired stopping offsets.");
+
+    public static final ConfigOption<Duration> LAKE_TIERING_TABLE_DURATION_DETECT_INTERVAL =
+            key("lake.tiering.table.duration.detect-interval")
+                    .durationType()
+                    .defaultValue(Duration.ofSeconds(30))
+                    .withDescription(
+                            "The interval to check if a table tiering operation has reached the maximum duration. "
+                                    + "The enumerator will periodically check tiering tables and force complete those that exceed the maximum duration.");
+
     // ------------------------------------------------------------------------
     //  ConfigOptions for fluss kafka
     // ------------------------------------------------------------------------
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/LakeTieringJobBuilder.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/LakeTieringJobBuilder.java
@@ -34,10 +34,10 @@
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
 
+import static org.apache.fluss.config.ConfigOptions.LAKE_TIERING_TABLE_DURATION_DETECT_INTERVAL;
+import static org.apache.fluss.config.ConfigOptions.LAKE_TIERING_TABLE_DURATION_MAX;
 import static org.apache.fluss.flink.tiering.source.TieringSource.TIERING_SOURCE_TRANSFORMATION_UID;
 import static org.apache.fluss.flink.tiering.source.TieringSourceOptions.POLL_TIERING_TABLE_INTERVAL;
-import static org.apache.fluss.flink.tiering.source.TieringSourceOptions.TIERING_TABLE_DURATION_DETECT_INTERVAL;
-import static org.apache.fluss.flink.tiering.source.TieringSourceOptions.TIERING_TABLE_DURATION_MAX;
 import static org.apache.fluss.utils.Preconditions.checkNotNull;
 
 /** The builder to build Flink lake tiering job. */
@@ -92,14 +92,14 @@ public JobClient build() throws Exception {
                     flussConfig.get(POLL_TIERING_TABLE_INTERVAL).toMillis());
         }
 
-        if (flussConfig.get(TIERING_TABLE_DURATION_MAX) != null) {
+        if (lakeTieringConfig.get(LAKE_TIERING_TABLE_DURATION_MAX) != null) {
             tieringSourceBuilder.withTieringTableDurationMax(
-                    flussConfig.get(TIERING_TABLE_DURATION_MAX).toMillis());
+                    lakeTieringConfig.get(LAKE_TIERING_TABLE_DURATION_MAX).toMillis());
         }
 
-        if (flussConfig.get(TIERING_TABLE_DURATION_DETECT_INTERVAL) != null) {
+        if (lakeTieringConfig.get(LAKE_TIERING_TABLE_DURATION_DETECT_INTERVAL) != null) {
             tieringSourceBuilder.withTieringTableDurationDetectInterval(
-                    flussConfig.get(TIERING_TABLE_DURATION_DETECT_INTERVAL).toMillis());
+                    lakeTieringConfig.get(LAKE_TIERING_TABLE_DURATION_DETECT_INTERVAL).toMillis());
         }
 
         TieringSource<?> tieringSource = tieringSourceBuilder.build();
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
@@ -44,9 +44,9 @@
 
 import java.nio.charset.StandardCharsets;
 
+import static org.apache.fluss.config.ConfigOptions.LAKE_TIERING_TABLE_DURATION_DETECT_INTERVAL;
+import static org.apache.fluss.config.ConfigOptions.LAKE_TIERING_TABLE_DURATION_MAX;
 import static org.apache.fluss.flink.tiering.source.TieringSourceOptions.POLL_TIERING_TABLE_INTERVAL;
-import static org.apache.fluss.flink.tiering.source.TieringSourceOptions.TIERING_TABLE_DURATION_DETECT_INTERVAL;
-import static org.apache.fluss.flink.tiering.source.TieringSourceOptions.TIERING_TABLE_DURATION_MAX;
 
 /**
  * The flink source implementation for tiering data from Fluss to downstream lake.
@@ -147,9 +147,9 @@ public static class Builder<WriteResult> {
         private long pollTieringTableIntervalMs =
                 POLL_TIERING_TABLE_INTERVAL.defaultValue().toMillis();
         private long tieringTableDurationMaxMs =
-                TIERING_TABLE_DURATION_MAX.defaultValue().toMillis();
+                LAKE_TIERING_TABLE_DURATION_MAX.defaultValue().toMillis();
         private long tieringTableDurationDetectIntervalMs =
-                TIERING_TABLE_DURATION_DETECT_INTERVAL.defaultValue().toMillis();
+                LAKE_TIERING_TABLE_DURATION_DETECT_INTERVAL.defaultValue().toMillis();
 
         public Builder(
                 Configuration flussConf, LakeTieringFactory<WriteResult, ?> lakeTieringFactory) {
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceFetcherManager.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceFetcherManager.java
@@ -33,8 +33,8 @@
 import java.util.function.Supplier;
 
 /**
- * The SplitFetcherManager for Fluss source. This class is needed to help notify a table reaches to
- * deadline of tiering to {@link TieringSplitReader}.
+ * The SplitFetcherManager for tiering source. This class is needed to help notify a table reaches
+ * to deadline of tiering to {@link TieringSplitReader}.
  */
 public class TieringSourceFetcherManager<WriteResult>
         extends SingleThreadFetcherManagerAdapter<
@@ -50,31 +50,31 @@ public TieringSourceFetcherManager(
         super(elementsQueue, splitReaderSupplier, configuration, splitFinishedHook);
     }
 
-    public void markTableReachTieringDeadline(long tableId) {
+    public void markTableReachTieringMaxDuration(long tableId) {
         if (!fetchers.isEmpty()) {
             // The fetcher thread is still running. This should be the majority of the cases.
             fetchers.values()
                     .forEach(
                             splitFetcher ->
-                                    enqueueMarkTableReachTieringDeadlineTask(
+                                    enqueueMarkTableReachTieringMaxDurationTask(
                                             splitFetcher, tableId));
         } else {
             SplitFetcher<TableBucketWriteResult<WriteResult>, TieringSplit> splitFetcher =
                     createSplitFetcher();
-            enqueueMarkTableReachTieringDeadlineTask(splitFetcher, tableId);
+            enqueueMarkTableReachTieringMaxDurationTask(splitFetcher, tableId);
             startFetcher(splitFetcher);
         }
     }
 
-    private void enqueueMarkTableReachTieringDeadlineTask(
+    private void enqueueMarkTableReachTieringMaxDurationTask(
             SplitFetcher<TableBucketWriteResult<WriteResult>, TieringSplit> splitFetcher,
             long reachTieringDeadlineTable) {
         splitFetcher.enqueueTask(
                 new SplitFetcherTask() {
                     @Override
                     public boolean run() {
                         ((TieringSplitReader<WriteResult>) splitFetcher.getSplitReader())
-                                .handleTableReachTieringDeadline(reachTieringDeadlineTable);
+                                .handleTableReachTieringMaxDuration(reachTieringDeadlineTable);
                         return true;
                     }
 
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceOptions.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceOptions.java
@@ -34,21 +34,4 @@ public class TieringSourceOptions {
                     .defaultValue(Duration.ofSeconds(30))
                     .withDescription(
                             "The fixed interval to request tiering table from Fluss cluster, by default 30 seconds.");
-
-    public static final ConfigOption<Duration> TIERING_TABLE_DURATION_MAX =
-            key("tiering.table.duration.max")
-                    .durationType()
-                    .defaultValue(Duration.ofMinutes(30))
-                    .withDescription(
-                            "The maximum duration for tiering a single table. If tiering a table exceeds this duration, "
-                                    + "it will be force completed: the tiering will be finalized and committed to the data lake "
-                                    + "(e.g., Paimon) immediately, even if they haven't reached their desired stopping offsets.");
-
-    public static final ConfigOption<Duration> TIERING_TABLE_DURATION_DETECT_INTERVAL =
-            key("tiering.table.duration.detect-interval")
-                    .durationType()
-                    .defaultValue(Duration.ofSeconds(30))
-                    .withDescription(
-                            "The interval to check if a table tiering operation has reached the maximum duration. "
-                                    + "The enumerator will periodically check tiering tables and force complete those that exceed the maximum duration.");
 }
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceReader.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceReader.java
@@ -122,7 +122,7 @@ public void handleSourceEvents(SourceEvent sourceEvent) {
                     (TieringReachMaxDurationEvent) sourceEvent;
             long tableId = reachMaxDurationEvent.getTableId();
             ((TieringSourceFetcherManager<WriteResult>) splitFetcherManager)
-                    .markTableReachTieringDeadline(tableId);
+                    .markTableReachTieringMaxDuration(tableId);
         }
     }
 
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java
@@ -57,6 +57,7 @@
 
 import static org.apache.fluss.utils.Preconditions.checkArgument;
 import static org.apache.fluss.utils.Preconditions.checkNotNull;
+import static org.apache.fluss.utils.Preconditions.checkState;
 
 /** The {@link SplitReader} implementation which will read Fluss and write to lake. */
 public class TieringSplitReader<WriteResult>
@@ -81,7 +82,7 @@ public class TieringSplitReader<WriteResult>
     // the table_id to the pending splits
     private final Map<Long, Set<TieringSplit>> pendingTieringSplits;
 
-    private final Set<Long> reachTieringDeadlineTables;
+    private final Set<Long> reachTieringMaxDurationTables;
 
     private final Map<TableBucket, LakeWriter<WriteResult>> lakeWriters;
     private final Connection connection;
@@ -124,7 +125,7 @@ protected TieringSplitReader(
         this.currentTableSplitsByBucket = new HashMap<>();
         this.lakeWriters = new HashMap<>();
         this.currentPendingSnapshotSplits = new ArrayDeque<>();
-        this.reachTieringDeadlineTables = new HashSet<>();
+        this.reachTieringMaxDurationTables = new HashSet<>();
         this.pollTimeout = pollTimeout;
     }
 
@@ -157,11 +158,11 @@ public RecordsWithSplitIds<TableBucketWriteResult<WriteResult>> fetch() throws I
             }
         } else {
             if (currentLogScanner != null) {
-                if (reachTieringDeadlineTables.contains(currentTableId)) {
+                // force to complete records
+                if (reachTieringMaxDurationTables.contains(currentTableId)) {
                     return forceCompleteTieringLogRecords();
                 }
                 ScanRecords scanRecords = currentLogScanner.poll(pollTimeout);
-                // force to complete records
                 return forLogRecords(scanRecords);
             } else {
                 return emptyTableBucketWriteResultWithSplitIds();
@@ -182,6 +183,9 @@ public void handleSplitsChanges(SplitsChange<TieringSplit> splitsChange) {
             if (split.isForceIgnore()) {
                 // if the split is forced to ignore,
                 // mark it as empty
+                LOG.info(
+                        "ignore split {} since the split is set to force to ignore",
+                        split.splitId());
                 currentEmptySplits.add(split);
                 continue;
             }
@@ -309,15 +313,26 @@ private void mayCreateLogScanner() {
                 TableBucketWriteResult<WriteResult> bucketWriteResult =
                         completeLakeWriter(
                                 bucket, split.getPartitionName(), logEndOffset, timestamp);
+
+                if (logEndOffset == UNKNOWN_BUCKET_OFFSET) {
+                    // when the log end offset is unknow, the write result must be
+                    // null, otherwise, we should throw exception directly to avoid data
+                    // inconsistent
+                    checkState(
+                            bucketWriteResult.writeResult() == null,
+                            "bucketWriteResult must be null when log end offset is unknown when tiering "
+                                    + split);
+                }
+
                 writeResults.put(bucket, bucketWriteResult);
                 finishedSplitIds.put(bucket, split.splitId());
                 LOG.info(
-                        "Split {} is forced to be finished due to tiering timeout.",
+                        "Split {} is forced to be finished due to tiering reach max duration.",
                         split.splitId());
                 currentTieringSplitsIterator.remove();
             }
         }
-        reachTieringDeadlineTables.remove(this.currentTableId);
+        reachTieringMaxDurationTables.remove(this.currentTableId);
         mayFinishCurrentTable();
         return new TableBucketWriteResultWithSplitIds(writeResults, finishedSplitIds);
     }
@@ -530,13 +545,13 @@ private void finishCurrentTable() throws IOException {
     }
 
     /**
-     * Handle a table reach tiered deadline. This will mark the current table as timed out, and it
-     * will be force completed in the next fetch cycle.
+     * Handle a table reach max tiering duration. This will mark the current table as reaching max
+     * duration, and it will be force completed in the next fetch cycle.
      */
-    public void handleTableReachTieringDeadline(long tableId) {
+    public void handleTableReachTieringMaxDuration(long tableId) {
         if ((currentTableId != null && currentTableId.equals(tableId)
                 || pendingTieringSplits.containsKey(tableId))) {
-            reachTieringDeadlineTables.add(tableId);
+            reachTieringMaxDurationTables.add(tableId);
         }
     }
 
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringLogSplit.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringLogSplit.java
@@ -72,7 +72,7 @@ public TieringLogSplit(
             long stoppingOffset,
             boolean forceIgnore,
             int numberOfSplits) {
-        super(tablePath, tableBucket, partitionName, numberOfSplits, forceIgnore);
+        super(tablePath, tableBucket, partitionName, forceIgnore, numberOfSplits);
         this.startingOffset = startingOffset;
         this.stoppingOffset = stoppingOffset;
     }
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSnapshotSplit.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSnapshotSplit.java
@@ -63,7 +63,7 @@ public TieringSnapshotSplit(
             long logOffsetOfSnapshot,
             int numberOfSplits,
             boolean forceIgnore) {
-        super(tablePath, tableBucket, partitionName, numberOfSplits, forceIgnore);
+        super(tablePath, tableBucket, partitionName, forceIgnore, numberOfSplits);
         this.snapshotId = snapshotId;
         this.logOffsetOfSnapshot = logOffsetOfSnapshot;
     }
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplit.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplit.java
@@ -38,17 +38,16 @@ public abstract class TieringSplit implements SourceSplit {
     protected final TableBucket tableBucket;
     @Nullable protected final String partitionName;
 
+    protected boolean forceIgnore;
     // the total number of splits in one round of tiering
     protected final int numberOfSplits;
 
-    protected boolean forceIgnore;
-
     public TieringSplit(
             TablePath tablePath,
             TableBucket tableBucket,
             @Nullable String partitionName,
-            int numberOfSplits,
-            boolean forceIgnore) {
+            boolean forceIgnore,
+            int numberOfSplits) {
         this.tablePath = tablePath;
         this.tableBucket = tableBucket;
         this.partitionName = partitionName;
@@ -57,8 +56,8 @@ public TieringSplit(
             throw new IllegalArgumentException(
                     "Partition name and partition id must be both null or both not null.");
         }
-        this.numberOfSplits = numberOfSplits;
         this.forceIgnore = forceIgnore;
+        this.numberOfSplits = numberOfSplits;
     }
 
     /** Checks whether this split is a primary key table split to tier. */
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/TieringITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/TieringITCase.java
@@ -58,8 +58,8 @@
 import java.util.Map;
 import java.util.Optional;
 
-import static org.apache.fluss.flink.tiering.source.TieringSourceOptions.TIERING_TABLE_DURATION_DETECT_INTERVAL;
-import static org.apache.fluss.flink.tiering.source.TieringSourceOptions.TIERING_TABLE_DURATION_MAX;
+import static org.apache.fluss.config.ConfigOptions.LAKE_TIERING_TABLE_DURATION_DETECT_INTERVAL;
+import static org.apache.fluss.config.ConfigOptions.LAKE_TIERING_TABLE_DURATION_MAX;
 import static org.apache.fluss.testutils.common.CommonTestUtils.waitValue;
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -233,15 +233,13 @@ private void writeRows(TablePath tablePath, List<InternalRow> rows, boolean appe
 
     private JobClient buildTieringJob(StreamExecutionEnvironment execEnv) throws Exception {
         Configuration lakeTieringConfig = new Configuration();
-        lakeTieringConfig.set(TIERING_TABLE_DURATION_MAX, Duration.ofSeconds(1));
-        lakeTieringConfig.set(TIERING_TABLE_DURATION_DETECT_INTERVAL, Duration.ofMillis(100));
+        lakeTieringConfig.set(LAKE_TIERING_TABLE_DURATION_MAX, Duration.ofSeconds(1));
+        lakeTieringConfig.set(LAKE_TIERING_TABLE_DURATION_DETECT_INTERVAL, Duration.ofMillis(100));
 
         Configuration flussConfig = new Configuration();
         flussConfig.setString(
                 ConfigOptions.BOOTSTRAP_SERVERS.key(),
                 FLUSS_CLUSTER_EXTENSION.getBootstrapServers());
-        flussConfig.set(TIERING_TABLE_DURATION_MAX, Duration.ofSeconds(1));
-        flussConfig.set(TIERING_TABLE_DURATION_DETECT_INTERVAL, Duration.ofMillis(100));
         return LakeTieringJobBuilder.newBuilder(
                         execEnv,
                         flussConfig,
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/TieringSourceReaderTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/TieringSourceReaderTest.java
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/lakehouse/TestingPaimonStoragePlugin.java b/fluss-server/src/test/java/org/apache/fluss/server/lakehouse/TestingPaimonStoragePlugin.java

Original file line number	Diff line number	Diff line change
`@@ -122,7 +122,7 @@ public void handleSourceEvents(SourceEvent sourceEvent) {`
`122`	`122`	`(TieringReachMaxDurationEvent) sourceEvent;`
`123`	`123`	`long tableId = reachMaxDurationEvent.getTableId();`
`124`	`124`	`((TieringSourceFetcherManager<WriteResult>) splitFetcherManager)`
`125`		`- .markTableReachTieringDeadline(tableId);`
	`125`	`+ .markTableReachTieringMaxDuration(tableId);`
`126`	`126`	`}`
`127`	`127`	`}`
`128`	`128`
Original file line number	Diff line number	Diff line change
`@@ -72,7 +72,7 @@ public TieringLogSplit(`
`72`	`72`	`long stoppingOffset,`
`73`	`73`	`boolean forceIgnore,`
`74`	`74`	`int numberOfSplits) {`
`75`		`- super(tablePath, tableBucket, partitionName, numberOfSplits, forceIgnore);`
	`75`	`+ super(tablePath, tableBucket, partitionName, forceIgnore, numberOfSplits);`
`76`	`76`	`this.startingOffset = startingOffset;`
`77`	`77`	`this.stoppingOffset = stoppingOffset;`
`78`	`78`	`}`
Original file line number	Diff line number	Diff line change
`@@ -63,7 +63,7 @@ public TieringSnapshotSplit(`
`63`	`63`	`long logOffsetOfSnapshot,`
`64`	`64`	`int numberOfSplits,`
`65`	`65`	`boolean forceIgnore) {`
`66`		`- super(tablePath, tableBucket, partitionName, numberOfSplits, forceIgnore);`
	`66`	`+ super(tablePath, tableBucket, partitionName, forceIgnore, numberOfSplits);`
`67`	`67`	`this.snapshotId = snapshotId;`
`68`	`68`	`this.logOffsetOfSnapshot = logOffsetOfSnapshot;`
`69`	`69`	`}`