(fluss) branch main updated: [lake] Handle coordinator disconnect in TieringSourceEnumerator (#3107)

[yuxia]

This is an automated email from the ASF dual-hosted git repository.

luoyuxia pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git


The following commit(s) were added to refs/heads/main by this push:
     new 3d6ababcd [lake] Handle coordinator disconnect in 
TieringSourceEnumerator (#3107)
3d6ababcd is described below

commit 3d6ababcde10a69aa74170721ece02f69710aafc
Author: Uğur Tafralı <tafraliu...@gmail.com>
AuthorDate: Wed May 27 13:02:20 2026 +0300

    [lake] Handle coordinator disconnect in TieringSourceEnumerator (#3107)
    
    ---------
    
    Co-authored-by: luoyuxia <luoyu...@alumni.sjtu.edu.cn>
---
 .../source/enumerator/TieringSourceEnumerator.java      |  6 ++++--
 .../source/enumerator/TieringSourceEnumeratorTest.java  | 17 +++++++++++++++++
 2 files changed, 21 insertions(+), 2 deletions(-)

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
index 89a10ee89..7c79596e8 100644
--- 
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
@@ -42,6 +42,7 @@ import org.apache.fluss.rpc.messages.PbHeartbeatReqForTable;
 import org.apache.fluss.rpc.messages.PbLakeTieringStats;
 import org.apache.fluss.rpc.messages.PbLakeTieringTableInfo;
 import org.apache.fluss.rpc.metrics.ClientMetricGroup;
+import org.apache.fluss.utils.ExceptionUtils;
 
 import org.apache.flink.api.connector.source.ReaderInfo;
 import org.apache.flink.api.connector.source.SourceEvent;
@@ -352,10 +353,11 @@ public class TieringSourceEnumerator
         }
     }
 
-    private void generateAndAssignSplits(
+    @VisibleForTesting
+    void generateAndAssignSplits(
             @Nullable Tuple3<Long, Long, TablePath> tieringTable, Throwable 
throwable) {
         if (throwable != null) {
-            LOG.warn("Failed to request tiering table, will retry later.", 
throwable);
+            ExceptionUtils.rethrow(throwable);
         }
         if (tieringTable != null) {
             generateTieringSplits(tieringTable);
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
index b725f99e7..446534688 100644
--- 
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
@@ -19,6 +19,7 @@ package org.apache.fluss.flink.tiering.source.enumerator;
 
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.NetworkException;
 import org.apache.fluss.flink.tiering.event.FailedTieringEvent;
 import org.apache.fluss.flink.tiering.event.FinishedTieringEvent;
 import org.apache.fluss.flink.tiering.event.TieringReachMaxDurationEvent;
@@ -37,6 +38,7 @@ import org.apache.fluss.rpc.messages.PbLakeTableSnapshotInfo;
 import org.apache.flink.api.connector.source.SourceEvent;
 import org.apache.flink.api.connector.source.SplitsAssignment;
 import org.apache.flink.api.connector.source.mocks.MockSplitEnumeratorContext;
+import org.apache.flink.util.FlinkRuntimeException;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -55,6 +57,7 @@ import static 
org.apache.fluss.client.table.scanner.log.LogScanner.EARLIEST_OFFS
 import static 
org.apache.fluss.config.ConfigOptions.TABLE_AUTO_PARTITION_NUM_PRECREATE;
 import static org.apache.fluss.testutils.common.CommonTestUtils.retry;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** Unit tests for {@link TieringSourceEnumerator} and {@link 
TieringSplitGenerator}. */
 class TieringSourceEnumeratorTest extends TieringTestBase {
@@ -730,6 +733,20 @@ class TieringSourceEnumeratorTest extends TieringTestBase {
         return new TieringSourceEnumerator(flussConf, context, 500);
     }
 
+    @Test
+    void testNetworkErrorInHeartbeatTriggersFailover() throws Exception {
+        try (FlussMockSplitEnumeratorContext<TieringSplit> context =
+                new FlussMockSplitEnumeratorContext<>(1)) {
+            TieringSourceEnumerator enumerator = 
createTieringSourceEnumerator(flussConf, context);
+            FlinkRuntimeException networkError =
+                    new FlinkRuntimeException(
+                            "Failed to wait heartbeat response due to ",
+                            new NetworkException("coordinator disconnected"));
+            assertThatThrownBy(() -> enumerator.generateAndAssignSplits(null, 
networkError))
+                    .isSameAs(networkError);
+        }
+    }
+
     @Test
     void testTableReachMaxTieringDuration() throws Throwable {
         TablePath tablePath = TablePath.of(DEFAULT_DB, 
"tiering-max-duration-test-log-table");