[hotfix] Tiering Source Enumerator should clear pending split when failover (apache#1820)

luoyuxia · web-flow · commit ec14e1b4c7b9 · 2025-10-17T10:46:52.000+08:00
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperator.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperator.java
@@ -25,7 +25,7 @@
 import org.apache.fluss.exception.LakeTableSnapshotNotExistException;
 import org.apache.fluss.flink.tiering.event.FailedTieringEvent;
 import org.apache.fluss.flink.tiering.event.FinishedTieringEvent;
-import org.apache.fluss.flink.tiering.event.TieringRestoreEvent;
+import org.apache.fluss.flink.tiering.event.TieringFailOverEvent;
 import org.apache.fluss.flink.tiering.source.TableBucketWriteResult;
 import org.apache.fluss.flink.tiering.source.TieringSource;
 import org.apache.fluss.lake.committer.BucketOffset;
@@ -129,10 +129,10 @@ public void setup(
         super.setup(containingTask, config, output);
         int attemptNumber = getRuntimeContext().getAttemptNumber();
         if (attemptNumber > 0) {
-            LOG.info("Send TieringRestoreEvent");
+            LOG.info("Send TieringFailoverEvent, current attempt number: {}", attemptNumber);
             // attempt number is greater than zero, the job must failover
             operatorEventGateway.sendEventToCoordinator(
-                    new SourceEventWrapper(new TieringRestoreEvent()));
+                    new SourceEventWrapper(new TieringFailOverEvent()));
         }
     }
 
@@ -175,6 +175,9 @@ public void processElement(StreamRecord<TableBucketWriteResult<WriteResult>> str
                         new SourceEventWrapper(
                                 new FailedTieringEvent(
                                         tableId, ExceptionUtils.stringifyException(e))));
+                LOG.warn(
+                        "Fail to commit tiering write result, will try to tier again in next round.",
+                        e);
             } finally {
                 collectedTableBucketWriteResults.remove(tableId);
             }
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/event/TieringFailOverEvent.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/event/TieringFailOverEvent.java
@@ -20,7 +20,7 @@
 
 import org.apache.flink.api.connector.source.SourceEvent;
 
-/** SourceEvent used to represent tiering is restoring. */
-public class TieringRestoreEvent implements SourceEvent {
+/** SourceEvent used to represent tiering is failover. */
+public class TieringFailOverEvent implements SourceEvent {
     private static final long serialVersionUID = 1L;
 }
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java
@@ -26,7 +26,7 @@
 import org.apache.fluss.flink.metrics.FlinkMetricRegistry;
 import org.apache.fluss.flink.tiering.event.FailedTieringEvent;
 import org.apache.fluss.flink.tiering.event.FinishedTieringEvent;
-import org.apache.fluss.flink.tiering.event.TieringRestoreEvent;
+import org.apache.fluss.flink.tiering.event.TieringFailOverEvent;
 import org.apache.fluss.flink.tiering.source.split.TieringSplit;
 import org.apache.fluss.flink.tiering.source.split.TieringSplitGenerator;
 import org.apache.fluss.flink.tiering.source.state.TieringSourceEnumeratorState;
@@ -218,13 +218,15 @@ public void handleSourceEvent(int subtaskId, SourceEvent sourceEvent) {
             }
         }
 
-        if (sourceEvent instanceof TieringRestoreEvent) {
+        if (sourceEvent instanceof TieringFailOverEvent) {
             LOG.info(
-                    "Receiving tiering restore event, mark current tiering table epoch {} as failed.",
+                    "Receiving tiering failover event, mark current tiering table epoch {} as failed.",
                     tieringTableEpochs);
             // we need to make all as failed
             failedTableEpochs.putAll(new HashMap<>(tieringTableEpochs));
             tieringTableEpochs.clear();
+            // also clean all pending splits since we mark all as failed
+            pendingSplits.clear();
         }
 
         if (!finishedTableEpochs.isEmpty() || !failedTableEpochs.isEmpty()) {
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java
@@ -20,6 +20,7 @@
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.flink.tiering.event.FailedTieringEvent;
 import org.apache.fluss.flink.tiering.event.FinishedTieringEvent;
+import org.apache.fluss.flink.tiering.event.TieringFailOverEvent;
 import org.apache.fluss.flink.tiering.source.TieringTestBase;
 import org.apache.fluss.flink.tiering.source.split.TieringLogSplit;
 import org.apache.fluss.flink.tiering.source.split.TieringSnapshotSplit;
@@ -644,6 +645,48 @@ void testHandleFailedTieringTableEvent() throws Throwable {
         }
     }
 
+    @Test
+    void testHandleFailOverEvent() throws Throwable {
+        TablePath tablePath1 = TablePath.of(DEFAULT_DB, "tiering-failover-test-log-table1");
+        createTable(tablePath1, DEFAULT_LOG_TABLE_DESCRIPTOR);
+
+        TablePath tablePath2 = TablePath.of(DEFAULT_DB, "tiering-failover-test-log-table2");
+        createTable(tablePath2, DEFAULT_LOG_TABLE_DESCRIPTOR);
+
+        int numSubtasks = 1;
+        try (MockSplitEnumeratorContext<TieringSplit> context =
+                new MockSplitEnumeratorContext<>(numSubtasks)) {
+            TieringSourceEnumerator enumerator =
+                    new TieringSourceEnumerator(flussConf, context, 500);
+
+            enumerator.start();
+            assertThat(context.getSplitsAssignmentSequence()).isEmpty();
+
+            // register one reader
+            int subtaskId = 0;
+            registerReader(context, enumerator, subtaskId, "localhost-" + subtaskId);
+
+            // handle split request
+            enumerator.handleSplitRequest(subtaskId, "localhost-" + subtaskId);
+
+            // should get one tiering split, and the split is for tablePath1
+            verifyTieringSplitAssignment(context, 1, tablePath1);
+
+            // clean assignment
+            context.getSplitsAssignmentSequence().clear();
+
+            // enumerator handle TieringFailOverEvent, which will mark current tiering tablePath1 as
+            // fail, and all pending splits should be clear
+            enumerator.handleSourceEvent(subtaskId, new TieringFailOverEvent());
+
+            // handle split request
+            enumerator.handleSplitRequest(subtaskId, "localhost-" + subtaskId);
+            // now, should get another one tiering split, the split is for tablePath2 since all
+            // pending split for tablePath1 is clear
+            verifyTieringSplitAssignment(context, 1, tablePath2);
+        }
+    }
+
     private static CommitLakeTableSnapshotRequest genCommitLakeTableSnapshotRequest(
             long tableId,
             @Nullable Long partitionId,
@@ -697,4 +740,23 @@ private static List<TieringSplit> sortSplits(List<TieringSplit> splits) {
                 .sorted(Comparator.comparing(Object::toString))
                 .collect(Collectors.toList());
     }
+
+    private void verifyTieringSplitAssignment(
+            MockSplitEnumeratorContext<TieringSplit> context,
+            int expectedSplitSize,
+            TablePath expectedTablePath)
+            throws Throwable {
+        waitUntilTieringTableSplitAssignmentReady(context, 1, 200);
+        List<SplitsAssignment<TieringSplit>> actualAssignment =
+                context.getSplitsAssignmentSequence();
+
+        List<TieringSplit> allTieringSplits =
+                actualAssignment.stream()
+                        .flatMap(assignments -> assignments.assignment().values().stream())
+                        .flatMap(List::stream)
+                        .collect(Collectors.toList());
+        assertThat(allTieringSplits).hasSize(expectedSplitSize);
+        assertThat(allTieringSplits)
+                .allMatch(tieringSplit -> tieringSplit.getTablePath().equals(expectedTablePath));
+    }
 }
diff --git a/fluss-test-coverage/pom.xml b/fluss-test-coverage/pom.xml
@@ -395,7 +395,6 @@
                                         <exclude>org.apache.fluss.flink.tiering.source.TieringSourceOptions</exclude>
                                         <exclude>org.apache.fluss.flink.tiering.source.TieringSource.Builder</exclude>
                                         <exclude>org.apache.fluss.flink.tiering.source.TieringSource</exclude>
-                                        <exclude>org.apache.fluss.flink.tiering.event.TieringRestoreEvent</exclude>
                                         <exclude>
                                             org.apache.fluss.flink.tiering.source.enumerator.TieringSourceEnumerator
                                         </exclude>

Original file line number	Diff line number	Diff line change
`@@ -20,7 +20,7 @@`
`20`	`20`
`21`	`21`	`import org.apache.flink.api.connector.source.SourceEvent;`
`22`	`22`
`23`		`-/** SourceEvent used to represent tiering is restoring. */`
`24`		`-public class TieringRestoreEvent implements SourceEvent {`
	`23`	`+/** SourceEvent used to represent tiering is failover. */`
	`24`	`+public class TieringFailOverEvent implements SourceEvent {`
`25`	`25`	`private static final long serialVersionUID = 1L;`
`26`	`26`	`}`