This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-13007-bddd3b0e782920fdc784224dd7cf028383969e93 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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 {
