From b2c8d4246deca86fe9aa449025bdd72eb7c3ca4a Mon Sep 17 00:00:00 2001 From: Yang Guo Date: Mon, 19 Oct 2026 17:58:47 +0800 Subject: [PATCH] [test] Stabilize cluster health test setup Wait for leader initialization of all tables created by the merge-engine test before teardown deletes the schemas needed by KV recovery. Retry the initial cluster health assertions against one response until the cluster is GREEN before stopping a tablet server. This allows prior table deletion and leader activation to settle without racing a second health query. Fixes #4722 Co-Authored-By: Codex AI-Model: gpt-6 AI-Contributed/Feature: 0/0 AI-Contributed/UT: 31/31 --- .../fluss/client/admin/FlussAdminITCase.java | 31 ++++++++++++++++--- 1 file changed, 27 insertions(+), 4 deletions(-) diff --git a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java index b39739c6b01..884f4c9c592 100644 --- a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java +++ b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java @@ -125,6 +125,7 @@ import static org.apache.fluss.row.encode.KvValueLayout.PLAIN; import static org.apache.fluss.testutils.DataTestUtils.row; import static org.apache.fluss.testutils.InternalRowAssert.assertThatRow; +import static org.apache.fluss.testutils.common.CommonTestUtils.retry; import static org.apache.fluss.testutils.common.CommonTestUtils.waitUntil; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -2644,6 +2645,22 @@ void testCreateTableWithAggregateEngineAndWalChangelog() throws Exception { .build(); admin.createTable(tablePath6, tableDescriptor6, false).get(); assertThat(admin.tableExists(tablePath6).get()).isTrue(); + + // Finish leader initialization before teardown deletes the schemas needed by KV recovery. + for (TablePath tablePath : + Arrays.asList( + DEFAULT_TABLE_PATH, + tablePath2, + tablePath3, + tablePath4, + tablePath5, + tablePath6)) { + TableInfo tableInfo = admin.getTableInfo(tablePath).get(); + for (int bucket = 0; bucket < tableInfo.getNumBuckets(); bucket++) { + FLUSS_CLUSTER_EXTENSION.waitAndGetLeaderReplica( + new TableBucket(tableInfo.getTableId(), bucket)); + } + } } // ==================== Table Statistics Tests ==================== @@ -2905,10 +2922,16 @@ void testClusterHealthDuringRollingUpgrade() throws Exception { waitAllReplicasReady(tableId, 3); // Phase 1: Cluster is healthy — status should be GREEN. - ClusterHealth health = admin.getClusterHealth().get(); - assertThat(health.getStatus()).isEqualTo(ClusterHealthStatus.GREEN); - assertThat(health.getNumReplicas()).isEqualTo(health.getInSyncReplicas()); - assertThat(health.getNumLeaderReplicas()).isEqualTo(health.getActiveLeaderReplicas()); + // Wait for previous table deletions and coordinator-side leader activation to settle. + retry( + Duration.ofMinutes(1), + () -> { + ClusterHealth health = admin.getClusterHealth().get(); + assertThat(health.getStatus()).isEqualTo(ClusterHealthStatus.GREEN); + assertThat(health.getNumReplicas()).isEqualTo(health.getInSyncReplicas()); + assertThat(health.getNumLeaderReplicas()) + .isEqualTo(health.getActiveLeaderReplicas()); + }); // Phase 2: Stop one tablet server (simulate server crash during rolling upgrade). int stoppedServerId = 0;