You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by le...@apache.org on 2026/07/06 04:23:34 UTC
(flink-cdc) 05/24: [test][connector/postgres] Stabilize NewlyAddedTableITCase failover waits
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 <xb...@gmail.com>
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;
}