[lake] support lake tiering service force to commit due to timeout

Author: luoyuxia

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceFetcherManager.java

@@ -50,28 +50,31 @@ public TieringSourceFetcherManager(
         super(elementsQueue, splitReaderSupplier, configuration, splitFinishedHook);
     }
 
-    public void markTableAsTieringTimeOut(long tableId) {
-        SplitFetcher<TableBucketWriteResult<WriteResult>, TieringSplit> splitFetcher =
-                fetchers.get(0);
-        if (splitFetcher != null) {
+    public void markTableReachTieringDeadline(long tableId) {
+        if (!fetchers.isEmpty()) {
             // The fetcher thread is still running. This should be the majority of the cases.
-            enqueueMarkTableTieringTimeOutTask(splitFetcher, tableId);
+            fetchers.values()
+                    .forEach(
+                            splitFetcher ->
+                                    enqueueMarkTableReachTieringDeadlineTask(
+                                            splitFetcher, tableId));
         } else {
-            splitFetcher = createSplitFetcher();
-            enqueueMarkTableTieringTimeOutTask(splitFetcher, tableId);
+            SplitFetcher<TableBucketWriteResult<WriteResult>, TieringSplit> splitFetcher =
+                    createSplitFetcher();
+            enqueueMarkTableReachTieringDeadlineTask(splitFetcher, tableId);
             startFetcher(splitFetcher);
         }
     }
 
-    private void enqueueMarkTableTieringTimeOutTask(
+    private void enqueueMarkTableReachTieringDeadlineTask(
             SplitFetcher<TableBucketWriteResult<WriteResult>, TieringSplit> splitFetcher,
-            long tieringTimeOutTable) {
+            long reachTieringDeadlineTable) {
         splitFetcher.enqueueTask(
                 new SplitFetcherTask() {
                     @Override
                     public boolean run() {
                         ((TieringSplitReader<WriteResult>) splitFetcher.getSplitReader())
-                                .handleTableTimeout(tieringTimeOutTable);
+                                .handleTableReachTieringDeadline(reachTieringDeadlineTable);
                         return true;
                     }
 

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceReader.java

@@ -18,6 +18,7 @@
 package org.apache.fluss.flink.tiering.source;
 
 import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.annotation.VisibleForTesting;
 import org.apache.fluss.client.Connection;
 import org.apache.fluss.flink.adapter.SingleThreadMultiplexSourceReaderBaseAdapter;
 import org.apache.fluss.flink.tiering.event.TieringReachMaxDurationEvent;
@@ -31,10 +32,13 @@
 import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
 import org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
 
+import java.time.Duration;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 
+import static org.apache.fluss.flink.tiering.source.TieringSplitReader.DEFAULT_POLL_TIMEOUT;
+
 /** A {@link SourceReader} that read records from Fluss and write to lake. */
 @Internal
 public final class TieringSourceReader<WriteResult>
@@ -52,11 +56,22 @@ public TieringSourceReader(
             SourceReaderContext context,
             Connection connection,
             LakeTieringFactory<WriteResult, ?> lakeTieringFactory) {
+        this(elementsQueue, context, connection, lakeTieringFactory, DEFAULT_POLL_TIMEOUT);
+    }
+
+    @VisibleForTesting
+    TieringSourceReader(
+            FutureCompletingBlockingQueue<RecordsWithSplitIds<TableBucketWriteResult<WriteResult>>>
+                    elementsQueue,
+            SourceReaderContext context,
+            Connection connection,
+            LakeTieringFactory<WriteResult, ?> lakeTieringFactory,
+            Duration pollTimeout) {
         super(
                 elementsQueue,
                 new TieringSourceFetcherManager<>(
                         elementsQueue,
-                        () -> new TieringSplitReader<>(connection, lakeTieringFactory),
+                        () -> new TieringSplitReader<>(connection, lakeTieringFactory, pollTimeout),
                         context.getConfiguration(),
                         (ignore) -> {}),
                 new TableBucketWriteResultEmitter<>(),
@@ -107,7 +122,7 @@ public void handleSourceEvents(SourceEvent sourceEvent) {
                     (TieringReachMaxDurationEvent) sourceEvent;
             long tableId = reachMaxDurationEvent.getTableId();
             ((TieringSourceFetcherManager<WriteResult>) splitFetcherManager)
-                    .markTableAsTieringTimeOut(tableId);
+                    .markTableReachTieringDeadline(tableId);
         }
     }
 

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java

@@ -64,20 +64,25 @@ public class TieringSplitReader<WriteResult>
 
     private static final Logger LOG = LoggerFactory.getLogger(TieringSplitReader.class);
 
-    private static final Duration POLL_TIMEOUT = Duration.ofMillis(10000L);
+    public static final Duration DEFAULT_POLL_TIMEOUT = Duration.ofMillis(10_000L);
 
     // unknown bucket timestamp for empty split or snapshot split
     private static final long UNKNOWN_BUCKET_TIMESTAMP = -1;
 
-    private static final long UNKNOW_BUCKET_OFFSET = -1;
+    // unknown bucket offset for empty split or snapshot split
+    private static final long UNKNOWN_BUCKET_OFFSET = -1;
 
     private final LakeTieringFactory<WriteResult, ?> lakeTieringFactory;
 
+    private final Duration pollTimeout;
+
     // the id for the pending tables to be tiered
     private final Queue<Long> pendingTieringTables;
     // the table_id to the pending splits
     private final Map<Long, Set<TieringSplit>> pendingTieringSplits;
 
+    private final Set<Long> reachTieringDeadlineTables;
+
     private final Map<TableBucket, LakeWriter<WriteResult>> lakeWriters;
     private final Connection connection;
 
@@ -99,11 +104,15 @@ public class TieringSplitReader<WriteResult>
 
     private final Set<TieringSplit> currentEmptySplits;
 
-    // Flag to indicate if the current table has timed out and should be force completed
-    private boolean currentTableTieringTimedOut;
-
     public TieringSplitReader(
             Connection connection, LakeTieringFactory<WriteResult, ?> lakeTieringFactory) {
+        this(connection, lakeTieringFactory, DEFAULT_POLL_TIMEOUT);
+    }
+
+    protected TieringSplitReader(
+            Connection connection,
+            LakeTieringFactory<WriteResult, ?> lakeTieringFactory,
+            Duration pollTimeout) {
         this.lakeTieringFactory = lakeTieringFactory;
         // owned by TieringSourceReader
         this.connection = connection;
@@ -115,7 +124,8 @@ public TieringSplitReader(
         this.currentTableSplitsByBucket = new HashMap<>();
         this.lakeWriters = new HashMap<>();
         this.currentPendingSnapshotSplits = new ArrayDeque<>();
-        this.currentTableTieringTimedOut = false;
+        this.reachTieringDeadlineTables = new HashSet<>();
+        this.pollTimeout = pollTimeout;
     }
 
     @Override
@@ -147,10 +157,10 @@ public RecordsWithSplitIds<TableBucketWriteResult<WriteResult>> fetch() throws I
             }
         } else {
             if (currentLogScanner != null) {
-                if (currentTableTieringTimedOut) {
+                if (reachTieringDeadlineTables.contains(currentTableId)) {
                     return forceCompleteTieringLogRecords();
                 }
-                ScanRecords scanRecords = currentLogScanner.poll(POLL_TIMEOUT);
+                ScanRecords scanRecords = currentLogScanner.poll(pollTimeout);
                 // force to complete records
                 return forLogRecords(scanRecords);
             } else {
@@ -170,6 +180,8 @@ public void handleSplitsChanges(SplitsChange<TieringSplit> splitsChange) {
         for (TieringSplit split : splitsChange.splits()) {
             LOG.info("add split {}", split.splitId());
             if (split.isForceIgnore()) {
+                // if the split is forced to ignore,
+                // mark it as empty
                 currentEmptySplits.add(split);
                 continue;
             }
@@ -273,25 +285,40 @@ private void mayCreateLogScanner() {
             forceCompleteTieringLogRecords() throws IOException {
         Map<TableBucket, TableBucketWriteResult<WriteResult>> writeResults = new HashMap<>();
         Map<TableBucket, String> finishedSplitIds = new HashMap<>();
-        for (Map.Entry<TableBucket, LakeWriter<WriteResult>> entry : lakeWriters.entrySet()) {
+
+        // force finish all splits
+        Iterator<Map.Entry<TableBucket, TieringSplit>> currentTieringSplitsIterator =
+                currentTableSplitsByBucket.entrySet().iterator();
+        while (currentTieringSplitsIterator.hasNext()) {
+            Map.Entry<TableBucket, TieringSplit> entry = currentTieringSplitsIterator.next();
             TableBucket bucket = entry.getKey();
-            TieringSplit split = currentTableSplitsByBucket.get(bucket);
+            TieringSplit split = entry.getValue();
             if (split != null && split.isTieringLogSplit()) {
+                // get the current offset, timestamp that tiered so far
                 LogOffsetAndTimestamp logOffsetAndTimestamp =
                         currentTableTieredOffsetAndTimestamp.get(bucket);
+                long logEndOffset =
+                        logOffsetAndTimestamp == null
+                                ? UNKNOWN_BUCKET_OFFSET
+                                // logEngOffset is equal to offset tiered + 1
+                                : logOffsetAndTimestamp.logOffset + 1;
+                long timestamp =
+                        logOffsetAndTimestamp == null
+                                ? UNKNOWN_BUCKET_TIMESTAMP
+                                : logOffsetAndTimestamp.timestamp;
                 TableBucketWriteResult<WriteResult> bucketWriteResult =
                         completeLakeWriter(
-                                bucket,
-                                split.getPartitionName(),
-                                logOffsetAndTimestamp.logOffset,
-                                logOffsetAndTimestamp.timestamp);
+                                bucket, split.getPartitionName(), logEndOffset, timestamp);
                 writeResults.put(bucket, bucketWriteResult);
                 finishedSplitIds.put(bucket, split.splitId());
                 LOG.info(
                         "Split {} is forced to be finished due to tiering timeout.",
                         split.splitId());
+                currentTieringSplitsIterator.remove();
             }
         }
+        reachTieringDeadlineTables.remove(this.currentTableId);
+        mayFinishCurrentTable();
         return new TableBucketWriteResultWithSplitIds(writeResults, finishedSplitIds);
     }
 
@@ -343,7 +370,11 @@ private RecordsWithSplitIds<TableBucketWriteResult<WriteResult>> forLogRecords(
                                 lastRecord.timestamp()));
                 // put split of the bucket
                 finishedSplitIds.put(bucket, currentSplitId);
-                LOG.info("Split {} has been finished.", currentSplitId);
+                LOG.info(
+                        "Finish tier bucket {} for table {}, split: {}.",
+                        bucket,
+                        currentTablePath,
+                        currentSplitId);
             }
         }
 
@@ -377,8 +408,11 @@ private TableBucketWriteResult<WriteResult> completeLakeWriter(
             long maxTimestamp)
             throws IOException {
         LakeWriter<WriteResult> lakeWriter = lakeWriters.remove(bucket);
-        WriteResult writeResult = lakeWriter.complete();
-        lakeWriter.close();
+        WriteResult writeResult = null;
+        if (lakeWriter != null) {
+            writeResult = lakeWriter.complete();
+            lakeWriter.close();
+        }
         return toTableBucketWriteResult(
                 currentTablePath,
                 bucket,
@@ -402,7 +436,7 @@ private TableBucketWriteResultWithSplitIds forEmptySplits(Set<TieringSplit> empt
                             tableBucket,
                             tieringSplit.getPartitionName(),
                             null,
-                            UNKNOW_BUCKET_OFFSET,
+                            UNKNOWN_BUCKET_OFFSET,
                             UNKNOWN_BUCKET_TIMESTAMP,
                             tieringSplit.getNumberOfSplits()));
         }
@@ -412,7 +446,6 @@ private TableBucketWriteResultWithSplitIds forEmptySplits(Set<TieringSplit> empt
     private void mayFinishCurrentTable() throws IOException {
         // no any pending splits for the table, just finish the table
         if (currentTableSplitsByBucket.isEmpty()) {
-            LOG.info("Finish tier table {} of table id {}.", currentTablePath, currentTableId);
             finishCurrentTable();
         }
     }
@@ -427,6 +460,11 @@ private TableBucketWriteResultWithSplitIds finishCurrentSnapshotSplit() throws I
                         currentSnapshotSplit.getPartitionName(),
                         logEndOffset,
                         UNKNOWN_BUCKET_TIMESTAMP);
+        LOG.info(
+                "Finish tier bucket {} for table {}, split: {}.",
+                tableBucket,
+                currentTablePath,
+                splitId);
         closeCurrentSnapshotSplit();
         mayFinishCurrentTable();
         return new TableBucketWriteResultWithSplitIds(
@@ -492,15 +530,13 @@ private void finishCurrentTable() throws IOException {
     }
 
     /**
-     * Handle timeout event for a table. This will mark the current table as timed out, and it will
-     * be force completed in the next fetch cycle.
+     * 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.
      */
-    public void handleTableTimeout(long tableId) {
-        if (currentTableId != null && currentTableId.equals(tableId)) {
-            LOG.debug(
-                    "Table {} tiering timeout event received, will try best to force complete after current processing.",
-                    tableId);
-            currentTableTieringTimedOut = true;
+    public void handleTableReachTieringDeadline(long tableId) {
+        if ((currentTableId != null && currentTableId.equals(tableId)
+                || pendingTieringSplits.containsKey(tableId))) {
+            reachTieringDeadlineTables.add(tableId);
         }
     }
 
@@ -624,6 +660,7 @@ public Set<String> finishedSplits() {
     }
 
     private static final class LogOffsetAndTimestamp {
+
         private final long logOffset;
         private final long timestamp;
 

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java

@@ -17,12 +17,6 @@
 
 package org.apache.fluss.flink.tiering.source.enumerator;
 
-import org.apache.flink.api.connector.source.SourceEvent;
-import org.apache.flink.api.connector.source.SplitEnumerator;
-import org.apache.flink.api.connector.source.SplitEnumeratorContext;
-import org.apache.flink.api.java.tuple.Tuple3;
-import org.apache.flink.metrics.groups.SplitEnumeratorMetricGroup;
-import org.apache.flink.util.FlinkRuntimeException;
 import org.apache.fluss.annotation.VisibleForTesting;
 import org.apache.fluss.client.Connection;
 import org.apache.fluss.client.ConnectionFactory;
@@ -49,10 +43,18 @@
 import org.apache.fluss.utils.MapUtils;
 import org.apache.fluss.utils.clock.Clock;
 import org.apache.fluss.utils.clock.SystemClock;
+
+import org.apache.flink.api.connector.source.SourceEvent;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.metrics.groups.SplitEnumeratorMetricGroup;
+import org.apache.flink.util.FlinkRuntimeException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import javax.annotation.Nullable;
+
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -347,11 +349,6 @@ private void assignSplits() {
                     if (!pendingSplits.isEmpty()) {
                         TieringSplit tieringSplit = pendingSplits.remove(0);
                         context.assignSplit(tieringSplit, nextAwaitingReader);
-                        long tableId = tieringSplit.getTableBucket().getTableId();
-                        if (!tieringTablesDeadline.containsKey(tableId)) {
-                            tieringTablesDeadline.put(
-                                    tableId, clock.milliseconds() + tieringTableDurationMaxMs);
-                        }
                         readersAwaitingSplit.remove(nextAwaitingReader);
                     }
                 }
@@ -432,6 +429,8 @@ private void generateTieringSplits(Tuple3<Long, Long, TablePath> tieringTable)
             } else {
                 tieringTableEpochs.put(tieringTable.f0, tieringTable.f1);
                 pendingSplits.addAll(tieringSplits);
+                tieringTablesDeadline.put(
+                        tieringTable.f0, clock.milliseconds() + tieringTableDurationMaxMs);
             }
         } catch (Exception e) {
             LOG.warn("Fail to generate Tiering splits for table {}.", tieringTable.f2, e);

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringLogSplit.java

@@ -54,7 +54,25 @@ public TieringLogSplit(
             long startingOffset,
             long stoppingOffset,
             int numberOfSplits) {
-        super(tablePath, tableBucket, partitionName, numberOfSplits);
+        this(
+                tablePath,
+                tableBucket,
+                partitionName,
+                startingOffset,
+                stoppingOffset,
+                false,
+                numberOfSplits);
+    }
+
+    public TieringLogSplit(
+            TablePath tablePath,
+            TableBucket tableBucket,
+            @Nullable String partitionName,
+            long startingOffset,
+            long stoppingOffset,
+            boolean forceIgnore,
+            int numberOfSplits) {
+        super(tablePath, tableBucket, partitionName, numberOfSplits, forceIgnore);
         this.startingOffset = startingOffset;
         this.stoppingOffset = stoppingOffset;
     }
@@ -101,6 +119,7 @@ public TieringLogSplit copy(int numberOfSplits) {
                 partitionName,
                 startingOffset,
                 stoppingOffset,
+                forceIgnore,
                 numberOfSplits);
     }