This is an automated email from the ASF dual-hosted git repository.
davidzollo 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 d5fb65855b [Fix][E2E] Fix flaky jdbc-connectors-it-ddl tests with
timeout tuning and DM wait strategy (#10948)
d5fb65855b is described below
commit d5fb65855bfbf31ce60529150647736ae685dff2
Author: zhiwei.niu <[email protected]>
AuthorDate: Mon May 25 22:43:29 2026 +0800
[Fix][E2E] Fix flaky jdbc-connectors-it-ddl tests with timeout tuning and
DM wait strategy (#10948)
---
.../connectors/jdbc/AbstractSchemaChangeBaseIT.java | 18 +++++++++---------
.../seatunnel/connectors/jdbc/DmSchemaChangeIT.java | 6 ++++++
2 files changed, 15 insertions(+), 9 deletions(-)
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-ddl/src/test/java/org/apache/seatunnel/connectors/jdbc/AbstractSchemaChangeBaseIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-ddl/src/test/java/org/apache/seatunnel/connectors/jdbc/AbstractSchemaChangeBaseIT.java
index 56ec9c2c75..32a3d0076f 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-ddl/src/test/java/org/apache/seatunnel/connectors/jdbc/AbstractSchemaChangeBaseIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-ddl/src/test/java/org/apache/seatunnel/connectors/jdbc/AbstractSchemaChangeBaseIT.java
@@ -199,7 +199,7 @@ public abstract class AbstractSchemaChangeBaseIT extends
TestSuiteBase implement
given().pollDelay(Duration.ofSeconds(5))
.pollInterval(Duration.ofMillis(1000))
.await()
- .atMost(30, TimeUnit.SECONDS)
+ .atMost(60, TimeUnit.SECONDS)
.untilAsserted(
() -> {
Assertions.assertEquals("RUNNING",
container.getJobStatus(jobId));
@@ -231,7 +231,7 @@ public abstract class AbstractSchemaChangeBaseIT extends
TestSuiteBase implement
// savepoint 2
given().pollDelay(Duration.ofSeconds(5))
- .atMost(30000, TimeUnit.MILLISECONDS)
+ .atMost(60, TimeUnit.SECONDS)
.untilAsserted(
() ->
Assertions.assertEquals(
@@ -284,7 +284,7 @@ public abstract class AbstractSchemaChangeBaseIT extends
TestSuiteBase implement
given().pollDelay(Duration.ofSeconds(5))
.pollInterval(Duration.ofMillis(1000))
.await()
- .atMost(30, TimeUnit.SECONDS)
+ .atMost(60, TimeUnit.SECONDS)
.untilAsserted(
() -> {
Assertions.assertEquals("RUNNING",
container.getJobStatus(jobId));
@@ -294,7 +294,7 @@ public abstract class AbstractSchemaChangeBaseIT extends
TestSuiteBase implement
}
private void assertSchemaEvolution(String sourceTable, String sinkTable) {
- await().atMost(60000, TimeUnit.MILLISECONDS)
+ await().atMost(120, TimeUnit.SECONDS)
.untilAsserted(
() ->
Assertions.assertIterableEquals(
@@ -309,7 +309,7 @@ public abstract class AbstractSchemaChangeBaseIT extends
TestSuiteBase implement
// case1 add columns with cdc data at same time
sourceDatabase.setTemplateName("add_columns").createAndInitialize();
- await().atMost(60000, TimeUnit.MILLISECONDS)
+ await().atMost(120, TimeUnit.SECONDS)
.untilAsserted(
() ->
Assertions.assertIterableEquals(
@@ -323,7 +323,7 @@ public abstract class AbstractSchemaChangeBaseIT extends
TestSuiteBase implement
schemaChangeCase.getSinkQueryColumns(),
schemaChangeCase.getSchemaName(),
sinkTable))));
- await().atMost(60000, TimeUnit.MILLISECONDS)
+ await().atMost(120, TimeUnit.SECONDS)
.untilAsserted(
() -> {
Assertions.assertIterableEquals(
@@ -368,7 +368,7 @@ public abstract class AbstractSchemaChangeBaseIT extends
TestSuiteBase implement
}
private void assertSchemaEvolutionForAddColumns(String sourceTable, String
sinkTable) {
- await().atMost(60000, TimeUnit.MILLISECONDS)
+ await().atMost(120, TimeUnit.SECONDS)
.untilAsserted(
() ->
Assertions.assertIterableEquals(
@@ -385,7 +385,7 @@ public abstract class AbstractSchemaChangeBaseIT extends
TestSuiteBase implement
sourceDatabase.setTemplateName("add_columns").createAndInitialize();
given().pollDelay(Duration.ofSeconds(5))
.await()
- .atMost(120000, TimeUnit.MILLISECONDS)
+ .atMost(120, TimeUnit.SECONDS)
.untilAsserted(
() ->
Assertions.assertIterableEquals(
@@ -397,7 +397,7 @@ public abstract class AbstractSchemaChangeBaseIT extends
TestSuiteBase implement
schemaChangeCase.getSchemaName(),
sinkTable)
+ ORDER_BY)));
- await().atMost(60000, TimeUnit.MILLISECONDS)
+ await().atMost(120, TimeUnit.SECONDS)
.untilAsserted(
() -> {
Assertions.assertIterableEquals(
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-ddl/src/test/java/org/apache/seatunnel/connectors/jdbc/DmSchemaChangeIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-ddl/src/test/java/org/apache/seatunnel/connectors/jdbc/DmSchemaChangeIT.java
index 82d8a95fbf..7bb5ec1005 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-ddl/src/test/java/org/apache/seatunnel/connectors/jdbc/DmSchemaChangeIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-ddl/src/test/java/org/apache/seatunnel/connectors/jdbc/DmSchemaChangeIT.java
@@ -21,8 +21,11 @@ import
org.apache.seatunnel.shade.com.google.common.collect.Lists;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.Wait;
import org.testcontainers.utility.DockerLoggerFactory;
+import java.time.Duration;
+
public class DmSchemaChangeIT extends AbstractSchemaChangeBaseIT {
private static final String DATABASE_TYPE = "Dameng";
@@ -70,6 +73,9 @@ public class DmSchemaChangeIT extends
AbstractSchemaChangeBaseIT {
new GenericContainer<>(DM_IMAGE)
.withNetwork(NETWORK)
.withNetworkAliases(DM_CONTAINER_HOST)
+ .withExposedPorts(DM_PORT)
+ .waitingFor(Wait.forListeningPort())
+ .withStartupTimeout(Duration.ofMinutes(5))
.withLogConsumer(
new
Slf4jLogConsumer(DockerLoggerFactory.getLogger(DM_IMAGE)));
container.setPortBindings(Lists.newArrayList(String.format("%s:%s",
DM_PORT, DM_PORT)));
