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 2dc6c970f8 [Fix][E2E] Stabilize multiple flaky tests (#11102)
2dc6c970f8 is described below
commit 2dc6c970f84290fae89e79d3b3c38224f0036747
Author: zhiwei.niu <[email protected]>
AuthorDate: Tue Jun 16 16:21:53 2026 +0800
[Fix][E2E] Stabilize multiple flaky tests (#11102)
---
.../connector/cdc/sqlserver/SqlServerCDCIT.java | 39 +++++++++-
.../e2e/connector/tdengine/TDengineIT.java | 24 +++++-
.../engine/e2e/classloader/ClassLoaderITBase.java | 2 +-
.../engine/client/ConnectorPackageClientTest.java | 91 ++++++++++++----------
.../client/SeaTunnelEngineClusterRoleTest.java | 22 +++++-
5 files changed, 127 insertions(+), 51 deletions(-)
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/java/org/apache/seatunnel/e2e/connector/cdc/sqlserver/SqlServerCDCIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/java/org/apache/seatunnel/e2e/connector/cdc/sqlserver/SqlServerCDCIT.java
index fc48f3f81e..f45cc8f7e3 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/java/org/apache/seatunnel/e2e/connector/cdc/sqlserver/SqlServerCDCIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/java/org/apache/seatunnel/e2e/connector/cdc/sqlserver/SqlServerCDCIT.java
@@ -620,18 +620,51 @@ public class SqlServerCDCIT extends TestSuiteBase
implements TestResource {
final String ddlFile = String.format("ddl/%s.sql", sqlFile);
final URL ddlTestFile =
TestSuiteBase.class.getClassLoader().getResource(ddlFile);
Assertions.assertNotNull(ddlTestFile, "Cannot locate " + ddlFile);
- try (Connection connection = getJdbcConnection();
- Statement statement = connection.createStatement()) {
+ try {
List<String> statements =
parseStatements(Files.readAllLines(Paths.get(ddlTestFile.toURI())));
+ String currentDatabase = null;
for (String stmt : statements) {
- statement.execute(stmt);
+ String trimmed = stmt.trim();
+ if (trimmed.toUpperCase().startsWith("USE ")) {
+ currentDatabase =
trimmed.substring(4).replaceAll(";\\s*$", "").trim();
+ continue;
+ }
+ executeWithDeadlockRetry(stmt, currentDatabase);
}
} catch (Exception e) {
throw new RuntimeException(e);
}
}
+ private void executeWithDeadlockRetry(String sql, String database) {
+ Awaitility.await(
+ "Executing: "
+ + sql.substring(0, Math.min(80, sql.length()))
+ .replaceAll("\\s+", " "))
+ .atMost(60, TimeUnit.SECONDS)
+ .pollInterval(2, TimeUnit.SECONDS)
+ .until(
+ () -> {
+ try (Connection connection = getJdbcConnection();
+ Statement statement =
connection.createStatement()) {
+ if (database != null) {
+ statement.execute("USE " + database);
+ }
+ statement.execute(sql);
+ return true;
+ } catch (SQLException e) {
+ if (e.getMessage() != null
+ && (e.getMessage().contains("deadlock")
+ ||
e.getMessage().contains("Deadlock"))) {
+ log.warn("Deadlock detected, will retry:
{}", sql);
+ return false;
+ }
+ throw new RuntimeException(e);
+ }
+ });
+ }
+
private void initializeSqlServerTable(String sqlFile) {
final String ddlFile = String.format("ddl/%s.sql", sqlFile);
final URL ddlTestFile =
TestSuiteBase.class.getClassLoader().getResource(ddlFile);
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-tdengine-e2e/src/test/java/org/apache/seatunnel/e2e/connector/tdengine/TDengineIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-tdengine-e2e/src/test/java/org/apache/seatunnel/e2e/connector/tdengine/TDengineIT.java
index 12935b378d..6c60ac6bf1 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-tdengine-e2e/src/test/java/org/apache/seatunnel/e2e/connector/tdengine/TDengineIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-tdengine-e2e/src/test/java/org/apache/seatunnel/e2e/connector/tdengine/TDengineIT.java
@@ -108,16 +108,16 @@ public class TDengineIT extends TestSuiteBase implements
TestResource {
@SneakyThrows
private int generateTestDataSet() {
int rowCount;
+ waitForDatabaseReady(connection1, "CREATE DATABASE power KEEP 3650");
try (Statement stmt = connection1.createStatement()) {
- stmt.execute("CREATE DATABASE power KEEP 3650");
stmt.execute(
"CREATE STABLE power.meters (ts TIMESTAMP, current FLOAT,
voltage INT, phase FLOAT, off BOOL, nc NCHAR(10)) "
+ "TAGS (location BINARY(64), groupId INT)");
String sql = getSQL();
rowCount = stmt.executeUpdate(sql);
}
+ waitForDatabaseReady(connection2, "CREATE DATABASE power2 KEEP 3650");
try (Statement stmt = connection2.createStatement()) {
- stmt.execute("CREATE DATABASE power2 KEEP 3650");
stmt.execute(
"CREATE STABLE power2.meters2 (ts TIMESTAMP, current
FLOAT, voltage INT, phase FLOAT, off BOOL, nc NCHAR(10)) "
+ "TAGS (location BINARY(64), groupId INT)");
@@ -134,8 +134,8 @@ public class TDengineIT extends TestSuiteBase implements
TestResource {
"CREATE STABLE power2.meters4 (ts TIMESTAMP, current
FLOAT, voltage INT, phase FLOAT, off BOOL, nc NCHAR(10)) "
+ "TAGS (location BINARY(64), groupId INT)");
}
+ waitForDatabaseReady(connection2, "CREATE DATABASE power3 KEEP 3650");
try (Statement stmt = connection2.createStatement()) {
- stmt.execute("CREATE DATABASE power3 KEEP 3650");
stmt.execute(
"CREATE STABLE power3.meters5 (ts TIMESTAMP, current
FLOAT, voltage INT, phase FLOAT, off BOOL, nc NCHAR(10)) "
+ "TAGS (location BINARY(64), groupId INT)");
@@ -143,6 +143,24 @@ public class TDengineIT extends TestSuiteBase implements
TestResource {
return rowCount;
}
+ /**
+ * Waits for TDengine dnode to be fully online before creating the
database. The REST adapter
+ * (port 6041) may accept connections before the dnode registers with the
management node,
+ * causing "Out of dnodes" errors on CREATE DATABASE.
+ */
+ private void waitForDatabaseReady(Connection connection, String
createDbSql) {
+ given().ignoreExceptions()
+ .await()
+ .pollInterval(2, TimeUnit.SECONDS)
+ .atMost(60, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ try (Statement stmt =
connection.createStatement()) {
+ stmt.execute(createDbSql);
+ }
+ });
+ }
+
@TestTemplate
public void testTDengine(TestContainer container) throws Exception {
Container.ExecResult execResult =
diff --git
a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/classloader/ClassLoaderITBase.java
b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/classloader/ClassLoaderITBase.java
index 26cea8651c..f0ff5d78bd 100644
---
a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/classloader/ClassLoaderITBase.java
+++
b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/classloader/ClassLoaderITBase.java
@@ -182,7 +182,7 @@ public abstract class ClassLoaderITBase extends
SeaTunnelEngineContainer {
private int getClassLoaderUpperBound(int initialClassLoaderCount, int
iteration) {
if (cacheMode()) {
- return initialClassLoaderCount;
+ return initialClassLoaderCount +
DISABLE_CACHE_MODE_CLASSLOADERS_PER_JOB;
}
return initialClassLoaderCount +
DISABLE_CACHE_MODE_CLASSLOADERS_PER_JOB * (iteration + 1);
}
diff --git
a/seatunnel-engine/seatunnel-engine-client/src/test/java/org/apache/seatunnel/engine/client/ConnectorPackageClientTest.java
b/seatunnel-engine/seatunnel-engine-client/src/test/java/org/apache/seatunnel/engine/client/ConnectorPackageClientTest.java
index a2273782a3..89b5899592 100644
---
a/seatunnel-engine/seatunnel-engine-client/src/test/java/org/apache/seatunnel/engine/client/ConnectorPackageClientTest.java
+++
b/seatunnel-engine/seatunnel-engine-client/src/test/java/org/apache/seatunnel/engine/client/ConnectorPackageClientTest.java
@@ -191,50 +191,55 @@ public class ConnectorPackageClientTest {
SeaTunnelHazelcastClient seaTunnelHazelcastClient =
new SeaTunnelHazelcastClient(clientConfig);
- Common.setDeployMode(DeployMode.CLIENT);
- String filePath =
ContentFormatUtilTest.getResource("/client_test.conf");
- Config seaTunnelJobConfig = ConfigBuilder.of(Paths.get(filePath));
- ReadonlyConfig envOptions =
ReadonlyConfig.fromConfig(seaTunnelJobConfig.getConfig("env"));
- JobConfig jobConfig = new JobConfig();
- jobConfig.setName("testUploadConnectorPluginJars");
- jobConfig.setJobContext(new JobContext(JOB_ID));
- fillJobConfig(jobConfig, envOptions);
-
- ConnectorPackageClient connectorPackageClient =
- new ConnectorPackageClient(seaTunnelHazelcastClient);
- Path connectorDir = Common.connectorDir();
- File[] files =
- connectorDir
- .toFile()
- .listFiles(
- new FileFilter() {
- @Override
- public boolean accept(File pathname) {
- return
pathname.getName().endsWith(".jar")
- &&
(StringUtils.startsWithIgnoreCase(
-
pathname.getName(),
-
"connector-fake")
- ||
StringUtils.startsWithIgnoreCase(
-
pathname.getName(),
-
"connector-file"));
- }
- });
- if (files != null) {
- for (File file : files) {
- ConnectorJarIdentifier connectorJarIdentifier =
- connectorPackageClient.uploadConnectorPluginJar(
- JOB_ID, file.toURI().toURL());
- await().atMost(60000, TimeUnit.MILLISECONDS)
- .untilAsserted(
- () -> {
- Assertions.assertTrue(
- StringUtils.isNotBlank(
-
connectorJarIdentifier.getStoragePath()));
- Assertions.assertEquals(
-
ConnectorJarType.CONNECTOR_PLUGIN_JAR,
- connectorJarIdentifier.getType());
- });
+ try {
+ Common.setDeployMode(DeployMode.CLIENT);
+ String filePath =
ContentFormatUtilTest.getResource("/client_test.conf");
+ Config seaTunnelJobConfig = ConfigBuilder.of(Paths.get(filePath));
+ ReadonlyConfig envOptions =
+
ReadonlyConfig.fromConfig(seaTunnelJobConfig.getConfig("env"));
+ JobConfig jobConfig = new JobConfig();
+ jobConfig.setName("testUploadConnectorPluginJars");
+ jobConfig.setJobContext(new JobContext(JOB_ID));
+ fillJobConfig(jobConfig, envOptions);
+
+ ConnectorPackageClient connectorPackageClient =
+ new ConnectorPackageClient(seaTunnelHazelcastClient);
+ Path connectorDir = Common.connectorDir();
+ File[] files =
+ connectorDir
+ .toFile()
+ .listFiles(
+ new FileFilter() {
+ @Override
+ public boolean accept(File pathname) {
+ return
pathname.getName().endsWith(".jar")
+ &&
(StringUtils.startsWithIgnoreCase(
+
pathname.getName(),
+
"connector-fake")
+ ||
StringUtils.startsWithIgnoreCase(
+
pathname.getName(),
+
"connector-file"));
+ }
+ });
+ if (files != null) {
+ for (File file : files) {
+ ConnectorJarIdentifier connectorJarIdentifier =
+ connectorPackageClient.uploadConnectorPluginJar(
+ JOB_ID, file.toURI().toURL());
+ await().atMost(60000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () -> {
+ Assertions.assertTrue(
+ StringUtils.isNotBlank(
+
connectorJarIdentifier.getStoragePath()));
+ Assertions.assertEquals(
+
ConnectorJarType.CONNECTOR_PLUGIN_JAR,
+
connectorJarIdentifier.getType());
+ });
+ }
}
+ } finally {
+ seaTunnelHazelcastClient.shutdown();
}
}
diff --git
a/seatunnel-engine/seatunnel-engine-client/src/test/java/org/apache/seatunnel/engine/client/SeaTunnelEngineClusterRoleTest.java
b/seatunnel-engine/seatunnel-engine-client/src/test/java/org/apache/seatunnel/engine/client/SeaTunnelEngineClusterRoleTest.java
index 7f37656c64..82f63ae9d8 100644
---
a/seatunnel-engine/seatunnel-engine-client/src/test/java/org/apache/seatunnel/engine/client/SeaTunnelEngineClusterRoleTest.java
+++
b/seatunnel-engine/seatunnel-engine-client/src/test/java/org/apache/seatunnel/engine/client/SeaTunnelEngineClusterRoleTest.java
@@ -332,6 +332,10 @@ public class SeaTunnelEngineClusterRoleTest {
@Test
public void testStartMasterNodeWithTcpIp() {
SeaTunnelConfig seaTunnelConfig =
ConfigProvider.locateAndGetSeaTunnelConfig();
+ seaTunnelConfig
+ .getHazelcastConfig()
+ .setClusterName(
+
ContentFormatUtilTest.getClusterName("Test_testStartMasterNodeWithTcpIp"));
HazelcastInstanceImpl instance =
SeaTunnelServerStarter.createMasterHazelcastInstance(seaTunnelConfig);
Assertions.assertNotNull(instance);
@@ -343,6 +347,11 @@ public class SeaTunnelEngineClusterRoleTest {
public void testStartMasterNodeWithMulticastJoin() {
SeaTunnelConfig seaTunnelConfig =
ConfigProvider.locateAndGetSeaTunnelConfig();
seaTunnelConfig.setHazelcastConfig(Config.loadFromString(getMulticastConfig()));
+ seaTunnelConfig
+ .getHazelcastConfig()
+ .setClusterName(
+ ContentFormatUtilTest.getClusterName(
+ "Test_testStartMasterNodeWithMulticastJoin"));
HazelcastInstanceImpl instance =
SeaTunnelServerStarter.createMasterHazelcastInstance(seaTunnelConfig);
Assertions.assertNotNull(instance);
@@ -353,6 +362,11 @@ public class SeaTunnelEngineClusterRoleTest {
@Test
public void testCannotOnlyStartWorkerNodeWithTcpIp() {
SeaTunnelConfig seaTunnelConfig =
ConfigProvider.locateAndGetSeaTunnelConfig();
+ seaTunnelConfig
+ .getHazelcastConfig()
+ .setClusterName(
+ ContentFormatUtilTest.getClusterName(
+
"Test_testCannotOnlyStartWorkerNodeWithTcpIp"));
Assertions.assertThrows(
IllegalStateException.class,
() -> {
@@ -364,6 +378,11 @@ public class SeaTunnelEngineClusterRoleTest {
public void testCannotOnlyStartWorkerNodeWithMulticastJoin() {
SeaTunnelConfig seaTunnelConfig =
ConfigProvider.locateAndGetSeaTunnelConfig();
seaTunnelConfig.setHazelcastConfig(Config.loadFromString(getMulticastConfig()));
+ seaTunnelConfig
+ .getHazelcastConfig()
+ .setClusterName(
+ ContentFormatUtilTest.getClusterName(
+
"Test_testCannotOnlyStartWorkerNodeWithMulticastJoin"));
Assertions.assertThrows(
IllegalStateException.class,
() -> {
@@ -462,8 +481,9 @@ public class SeaTunnelEngineClusterRoleTest {
.listJobStatus(true)
.contains("RUNNING")));
jobClient.cancelJob(jobId);
- // Master handoff can delay terminal status propagation on the
slower JDK 8 CI lane.
await().atMost(120000, TimeUnit.MILLISECONDS)
+ .pollDelay(5, TimeUnit.SECONDS)
+ .pollInterval(2, TimeUnit.SECONDS)
.untilAsserted(
() -> {
String status = jobClient.getJobStatus(jobId);
