[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

@@ -51,13 +51,15 @@ public TieringSourceFetcherManager(
     }
 
     public void markTableAsTieringTimeOut(long tableId) {
-        SplitFetcher<TableBucketWriteResult<WriteResult>, TieringSplit> splitFetcher =
-                fetchers.get(0);
-        if (splitFetcher != null) {
+        if (!fetchers.isEmpty()) {
             // The fetcher thread is still running. This should be the majority of the cases.
-            enqueueMarkTableTieringTimeOutTask(splitFetcher, tableId);
+            fetchers.values()
+                    .forEach(
+                            splitFetcher ->
+                                    enqueueMarkTableTieringTimeOutTask(splitFetcher, tableId));
         } else {
-            splitFetcher = createSplitFetcher();
+            SplitFetcher<TableBucketWriteResult<WriteResult>, TieringSplit> splitFetcher =
+                    createSplitFetcher();
             enqueueMarkTableTieringTimeOutTask(splitFetcher, tableId);
             startFetcher(splitFetcher);
         }

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

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

@@ -64,7 +64,7 @@ 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;
@@ -73,11 +73,15 @@ public class TieringSplitReader<WriteResult>
 
     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> timeoutTables;
+
     private final Map<TableBucket, LakeWriter<WriteResult>> lakeWriters;
     private final Connection connection;
 
@@ -99,11 +103,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 +123,8 @@ public TieringSplitReader(
         this.currentTableSplitsByBucket = new HashMap<>();
         this.lakeWriters = new HashMap<>();
         this.currentPendingSnapshotSplits = new ArrayDeque<>();
-        this.currentTableTieringTimedOut = false;
+        this.timeoutTables = new HashSet<>();
+        this.pollTimeout = pollTimeout;
     }
 
     @Override
@@ -147,10 +156,10 @@ public RecordsWithSplitIds<TableBucketWriteResult<WriteResult>> fetch() throws I
             }
         } else {
             if (currentLogScanner != null) {
-                if (currentTableTieringTimedOut) {
+                if (timeoutTables.contains(currentTableId)) {
                     return forceCompleteTieringLogRecords();
                 }
-                ScanRecords scanRecords = currentLogScanner.poll(POLL_TIMEOUT);
+                ScanRecords scanRecords = currentLogScanner.poll(pollTimeout);
                 // force to complete records
                 return forLogRecords(scanRecords);
             } else {
@@ -273,25 +282,37 @@ 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()) {
+
+        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()) {
                 LogOffsetAndTimestamp logOffsetAndTimestamp =
                         currentTableTieredOffsetAndTimestamp.get(bucket);
+                long logEndOffset =
+                        logOffsetAndTimestamp == null
+                                ? UNKNOW_BUCKET_OFFSET
+                                : 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();
             }
         }
+        timeoutTables.remove(this.currentTableId);
+        mayFinishCurrentTable();
         return new TableBucketWriteResultWithSplitIds(writeResults, finishedSplitIds);
     }
 
@@ -343,7 +364,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 +402,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,
@@ -412,7 +440,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 +454,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(
@@ -496,11 +528,9 @@ private void finishCurrentTable() throws IOException {
      * 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;
+        if ((currentTableId != null && currentTableId.equals(tableId)
+                || pendingTieringSplits.containsKey(tableId))) {
+            timeoutTables.add(tableId);
         }
     }
 

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;

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,
+                numberOfSplits,
+                false);
+    }
+
+    public TieringLogSplit(
+            TablePath tablePath,
+            TableBucket tableBucket,
+            @Nullable String partitionName,
+            long startingOffset,
+            long stoppingOffset,
+            int numberOfSplits,
+            boolean forceIgnore) {
+        super(tablePath, tableBucket, partitionName, numberOfSplits, forceIgnore);
         this.startingOffset = startingOffset;
         this.stoppingOffset = stoppingOffset;
     }

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

@@ -43,10 +43,16 @@ public TieringSnapshotSplit(
             TableBucket tableBucket,
             @Nullable String partitionName,
             long snapshotId,
-            long logOffsetOfSnapshot) {
-        super(tablePath, tableBucket, partitionName, UNKNOWN_NUMBER_OF_SPLITS);
-        this.snapshotId = snapshotId;
-        this.logOffsetOfSnapshot = logOffsetOfSnapshot;
+            long logOffsetOfSnapshot,
+            int numberOfSplits) {
+        this(
+                tablePath,
+                tableBucket,
+                partitionName,
+                snapshotId,
+                logOffsetOfSnapshot,
+                numberOfSplits,
+                false);
     }
 
     public TieringSnapshotSplit(
@@ -55,8 +61,9 @@ public TieringSnapshotSplit(
             @Nullable String partitionName,
             long snapshotId,
             long logOffsetOfSnapshot,
-            int numberOfSplits) {
-        super(tablePath, tableBucket, partitionName, numberOfSplits);
+            int numberOfSplits,
+            boolean forceIgnore) {
+        super(tablePath, tableBucket, partitionName, numberOfSplits, forceIgnore);
         this.snapshotId = snapshotId;
         this.logOffsetOfSnapshot = logOffsetOfSnapshot;
     }
@@ -103,7 +110,8 @@ public TieringSnapshotSplit copy(int numberOfSplits) {
                 partitionName,
                 snapshotId,
                 logOffsetOfSnapshot,
-                numberOfSplits);
+                numberOfSplits,
+                forceIgnore);
     }
 
     @Override

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

@@ -41,13 +41,14 @@ public abstract class TieringSplit implements SourceSplit {
     // the total number of splits in one round of tiering
     protected final int numberOfSplits;
 
-    protected boolean forceIgnore = false;
+    protected boolean forceIgnore;
 
     public TieringSplit(
             TablePath tablePath,
             TableBucket tableBucket,
             @Nullable String partitionName,
-            int numberOfSplits) {
+            int numberOfSplits,
+            boolean forceIgnore) {
         this.tablePath = tablePath;
         this.tableBucket = tableBucket;
         this.partitionName = partitionName;
@@ -57,6 +58,7 @@ public TieringSplit(
                     "Partition name and partition id must be both null or both not null.");
         }
         this.numberOfSplits = numberOfSplits;
+        this.forceIgnore = forceIgnore;
     }
 
     /** Checks whether this split is a primary key table split to tier. */