This is an automated email from the ASF dual-hosted git repository. leonardBang pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
commit 3427027b4ea468e3ec6ff67e54709e012935fd79 Author: Leonard Xu <[email protected]> AuthorDate: Wed Jul 1 14:57:43 2026 +0800 [test][connector/postgres] Stabilize NewlyAddedTableITCase failover waits --- .../cdc/connectors/postgres/PostgresTestBase.java | 6 ++--- .../postgres/source/NewlyAddedTableITCase.java | 31 +++++++++++++++++++--- 2 files changed, 29 insertions(+), 8 deletions(-) diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/PostgresTestBase.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/PostgresTestBase.java index b0ba6c4e3..951b47314 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/PostgresTestBase.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/PostgresTestBase.java @@ -53,7 +53,7 @@ import java.util.Arrays; import java.util.Iterator; import java.util.List; import java.util.Map; -import java.util.Random; +import java.util.UUID; import java.util.regex.Matcher; import java.util.regex.Pattern; import java.util.stream.Collectors; @@ -131,9 +131,7 @@ public abstract class PostgresTestBase extends AbstractTestBase { } public static String getSlotName() { - final Random random = new Random(); - int id = random.nextInt(10000); - return "flink_" + id; + return "flink_" + UUID.randomUUID().toString().replace("-", ""); } /** diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/NewlyAddedTableITCase.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/NewlyAddedTableITCase.java index 327e662eb..241fc5920 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/NewlyAddedTableITCase.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/NewlyAddedTableITCase.java @@ -26,6 +26,7 @@ import org.apache.flink.configuration.StateRecoveryOptions; import org.apache.flink.core.execution.JobClient; import org.apache.flink.core.execution.SavepointFormatType; import org.apache.flink.runtime.checkpoint.CheckpointException; +import org.apache.flink.runtime.messages.FlinkJobNotFoundException; import org.apache.flink.runtime.minicluster.RpcServiceSharing; import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; @@ -519,6 +520,11 @@ class NewlyAddedTableITCase extends PostgresTestBase { // trigger failover after some snapshot data read finished if (failoverPhase == PostgresTestUtils.FailoverPhase.SNAPSHOT) { + // Wait until the job is RUNNING before triggering snapshot-phase failover. + // On JM failover, revoking leadership before the JobMaster leader election + // is established tears down the MiniCluster HA services with + // "high availability services are shut down". + PostgresTestUtils.waitUntilJobRunning(tableResult); PostgresTestUtils.triggerFailover( failoverType, jobClient.getJobID(), @@ -608,6 +614,7 @@ class NewlyAddedTableITCase extends PostgresTestBase { jobClient.getJobID(), miniClusterResource.get().getMiniCluster(), () -> sleepMs(100)); + PostgresTestUtils.waitUntilJobRunning(tableResult); } fetchedDataList.addAll(expectedWalLogDataThisRound); @@ -718,6 +725,11 @@ class NewlyAddedTableITCase extends PostgresTestBase { // trigger failover after some snapshot data read finished if (failoverPhase == PostgresTestUtils.FailoverPhase.SNAPSHOT) { + // Wait until the job is RUNNING before triggering snapshot-phase failover. + // On JM failover, revoking leadership before the JobMaster leader election + // is established tears down the MiniCluster HA services with + // "high availability services are shut down". + PostgresTestUtils.waitUntilJobRunning(tableResult); PostgresTestUtils.triggerFailover( failoverType, jobClient.getJobID(), @@ -739,6 +751,7 @@ class NewlyAddedTableITCase extends PostgresTestBase { jobClient.getJobID(), miniClusterResource.get().getMiniCluster(), () -> sleepMs(100)); + PostgresTestUtils.waitUntilJobRunning(tableResult); } makeSecondPartWalLogForAddressTable(getConnection(), newlyAddedTable); @@ -903,10 +916,16 @@ class NewlyAddedTableITCase extends PostgresTestBase { .triggerSavepoint(savepointDirectory, SavepointFormatType.DEFAULT) .get(); } catch (Exception e) { - Optional<CheckpointException> exception = + Optional<CheckpointException> checkpointException = ExceptionUtils.findThrowable(e, CheckpointException.class); - if (exception.isPresent() - && exception.get().getMessage().contains("Checkpoint triggering task")) { + Optional<FlinkJobNotFoundException> jobNotFoundException = + ExceptionUtils.findThrowable(e, FlinkJobNotFoundException.class); + if ((checkpointException.isPresent() + && checkpointException + .get() + .getMessage() + .contains("Checkpoint triggering task")) + || jobNotFoundException.isPresent()) { Thread.sleep(100); retryTimes++; } else { @@ -927,7 +946,11 @@ class NewlyAddedTableITCase extends PostgresTestBase { StreamExecutionEnvironment.getExecutionEnvironment(configuration); env.setParallelism(parallelism); env.enableCheckpointing(200L); - RestartStrategyUtils.configureFixedDelayRestartStrategy(env, 3, 100L); + // A single TM failover fails all subtasks of the pipelined region, and racing + // "disconnect from JobManager"/"TaskExecutor is shutting down" failures can be + // counted individually against the restart budget. Allow enough attempts so one + // failover never exhausts the strategy and pushes the job to FAILED. + RestartStrategyUtils.configureFixedDelayRestartStrategy(env, 10, 100L); return env; }
