This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 48b23d0e2f [Fix][E2E] Retry Fluss readiness connection failures
(#13007)
48b23d0e2f is described below
commit 48b23d0e2f163bf2aa15577a419d49f8a53636db
Author: zhiweiniu <[email protected]>
AuthorDate: Tue Oct 27 10:45:21 2026 +0000
[Fix][E2E] Retry Fluss readiness connection failures (#13007)
---
.../e2e/connector/fluss/FlussTestBase.java | 43 ++++++++++++----------
1 file changed, 23 insertions(+), 20 deletions(-)
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-fluss-e2e/src/test/java/org/apache/seatunnel/e2e/connector/fluss/FlussTestBase.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-fluss-e2e/src/test/java/org/apache/seatunnel/e2e/connector/fluss/FlussTestBase.java
index b87453c9d8..8c809df8d9 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-fluss-e2e/src/test/java/org/apache/seatunnel/e2e/connector/fluss/FlussTestBase.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-fluss-e2e/src/test/java/org/apache/seatunnel/e2e/connector/fluss/FlussTestBase.java
@@ -247,26 +247,29 @@ public abstract class FlussTestBase extends TestSuiteBase
implements TestResourc
}
private void waitForBucketLeaders(TablePath tablePath) throws Exception {
- // Use a new connection so no metadata cached for an earlier table at
this path is used.
- try (Connection connection =
ConnectionFactory.createConnection(flussConfig())) {
- Admin admin = connection.getAdmin();
- TableInfo tableInfo = admin.getTableInfo(tablePath).get();
- List<Integer> buckets =
- IntStream.range(0, tableInfo.getNumBuckets())
- .boxed()
- .collect(Collectors.toList());
- // Listing offsets fails while a bucket has no leader.
- given().ignoreExceptions()
- .await()
- .atMost(30, TimeUnit.SECONDS)
- .pollInterval(1, TimeUnit.SECONDS)
- .untilAsserted(
- () ->
- admin.listOffsets(
- tablePath, buckets, new
OffsetSpec.LatestSpec())
- .all()
- .get(10, TimeUnit.SECONDS));
- }
+ given().ignoreExceptions()
+ .await()
+ .atMost(30, TimeUnit.SECONDS)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ // Bootstrap can disconnect too. Retry it with
metadata discovery so a
+ // failed client or cached metadata from a dropped
table is not reused.
+ try (Connection connection =
+
ConnectionFactory.createConnection(flussConfig())) {
+ Admin admin = connection.getAdmin();
+ TableInfo tableInfo =
+ admin.getTableInfo(tablePath).get(10,
TimeUnit.SECONDS);
+ List<Integer> buckets =
+ IntStream.range(0,
tableInfo.getNumBuckets())
+ .boxed()
+ .collect(Collectors.toList());
+ // Listing offsets fails while a bucket has no
leader.
+ admin.listOffsets(tablePath, buckets, new
OffsetSpec.LatestSpec())
+ .all()
+ .get(10, TimeUnit.SECONDS);
+ }
+ });
}
protected static boolean checkPort(String host, int port, int timeoutMs)
throws IOException {
