refactor tiering options

Author: luoyuxia

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
     // ------------------------------------------------------------------------

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();

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) {

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;
                     }
 

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.");
 }

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);
         }
     }
 

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;
             }
@@ -300,7 +304,7 @@ private void mayCreateLogScanner() {
                 long logEndOffset =
                         logOffsetAndTimestamp == null
                                 ? UNKNOWN_BUCKET_OFFSET
-                                // logEngOffset is equal to offset tiered + 1
+                                // logEndOffset is equal to offset tiered + 1
                                 : logOffsetAndTimestamp.logOffset + 1;
                 long timestamp =
                         logOffsetAndTimestamp == null
@@ -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 unknown, 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) {
-        if ((currentTableId != null && currentTableId.equals(tableId)
-                || pendingTieringSplits.containsKey(tableId))) {
-            reachTieringDeadlineTables.add(tableId);
+    public void handleTableReachTieringMaxDuration(long tableId) {
+        if ((currentTableId != null && currentTableId.equals(tableId))
+                || pendingTieringSplits.containsKey(tableId)) {
+            reachTieringMaxDurationTables.add(tableId);
         }
     }
 

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;
     }

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;
     }

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. */
@@ -140,11 +139,12 @@ public boolean equals(Object object) {
         return Objects.equals(tablePath, that.tablePath)
                 && Objects.equals(tableBucket, that.tableBucket)
                 && Objects.equals(partitionName, that.partitionName)
+                && forceIgnore == that.forceIgnore
                 && numberOfSplits == that.numberOfSplits;
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(tablePath, tableBucket, partitionName, numberOfSplits);
+        return Objects.hash(tablePath, tableBucket, partitionName, forceIgnore, numberOfSplits);
     }
 }