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 12ab1d0770 [Fix][E2E] Stabilize flaky connector E2E tests (#11456)
12ab1d0770 is described below
commit 12ab1d07701f6344506f15af38e8bb3526f5e73c
Author: zhiwei.niu <[email protected]>
AuthorDate: Sun Jul 19 17:35:19 2026 +0800
[Fix][E2E] Stabilize flaky connector E2E tests (#11456)
---
.../src/test/java/mongodb/MongodbCDCIT.java | 13 +--
.../test/java/mongodb/MongodbCDCMultiSourceIT.java | 14 +--
.../src/test/java/mongodb/MysqlDriverResolver.java | 66 ++++++++++++++
.../cdc/mysql/AbstractMysqlCDCITBase.java | 14 +--
.../cdc/mysql/MysqlCDCDriverResolver.java | 66 ++++++++++++++
.../mysql/MysqlCDCSpecificStartingOffsetIT.java | 15 +---
.../cdc/mysql/MysqlCDCWithBinlogDeleteIT.java | 15 +---
.../cdc/mysql/MysqlCDCWithFlinkSchemaChangeIT.java | 52 +++++------
.../MysqlCDCWithMissingBinlogDiagnosticsIT.java | 14 +--
.../cdc/mysql/MysqlCDCWithSchemaChangeIT.java | 15 +---
.../seatunnel/cdc/postgres/OpengaussCDCIT.java | 35 ++++++--
.../seatunnel/cdc/oracle/AbstractOracleCDCIT.java | 67 +++++++++++++-
.../seatunnel/cdc/oracle/OracleCDCIT.java | 11 +--
.../cdc/oracle/OracleCDCWithSchemaChangeIT.java | 16 +---
.../seatunnel/cdc/postgres/PostgresCDCIT.java | 34 +++++--
.../connector-cdc-tidb-e2e/pom.xml | 7 ++
.../seatunnel/e2e/connector/tidb/TiDBCDCIT.java | 75 +++++++++++-----
.../e2e/connector/databend/DatabendCDCSinkIT.java | 32 ++++++-
.../e2e/connector/doris/AbstractDorisIT.java | 41 ++++++++-
.../e2e/connector/doris/DorisCDCSinkIT.java | 13 +--
.../e2e/connector/doris/DorisErrorIT.java | 19 ++--
.../seatunnel/e2e/connector/doris/DorisIT.java | 13 +--
.../e2e/connector/doris/DorisMultiReadIT.java | 12 +--
.../e2e/connector/doris/DorisSchemaChangeIT.java | 15 +---
.../connector/edgesocket/AbstractEdgeSocketIT.java | 35 ++++++++
.../edgesocket/EdgeSocketCheckpointRestoreIT.java | 15 +---
.../connector/edgesocket/EdgeSocketMysql8_4IT.java | 15 +---
.../e2e/connector/file/fstp/SftpFileIT.java | 15 ++--
.../seatunnel/e2e/connector/http/HttpIT.java | 32 +++++--
.../e2e/connector/kafka/KafkaFormatIT.java | 31 +++++--
.../seatunnel/e2e/connector/kafka/KafkaIT.java | 100 ++++++++++++++++-----
.../e2e/connector/pulsar/CanalToPulsarIT.java | 34 +++++--
.../e2e/connector/rocketmq/RocketMqIT.java | 62 ++++++++++---
.../connector/starrocks/StarRocksCDCSinkIT.java | 60 ++++++++-----
.../e2e/connector/starrocks/StarRocksIT.java | 63 ++++++++-----
.../starrocks/StarRocksSchemaChangeIT.java | 59 +++++++-----
36 files changed, 793 insertions(+), 402 deletions(-)
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mongodb-e2e/src/test/java/mongodb/MongodbCDCIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mongodb-e2e/src/test/java/mongodb/MongodbCDCIT.java
index 334f895dcb..480a950ebc 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mongodb-e2e/src/test/java/mongodb/MongodbCDCIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mongodb-e2e/src/test/java/mongodb/MongodbCDCIT.java
@@ -124,9 +124,6 @@ public class MongodbCDCIT extends TestSuiteBase implements
TestResource {
private static final String SINK_SQL_ORDERS =
"select order_number,order_date,quantity,product_id from orders
order by order_number asc";
- private static final String MYSQL_DRIVER_JAR =
-
"https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.16/mysql-connector-java-8.0.16.jar";;
-
private final UniqueDatabase inventoryDatabase =
new UniqueDatabase(MYSQL_CONTAINER, MYSQL_DATABASE);
@@ -146,15 +143,7 @@ public class MongodbCDCIT extends TestSuiteBase implements
TestResource {
@TestContainerExtension
private final ContainerExtendedFactory extendedFactory =
- container -> {
- Container.ExecResult extraCommands =
- container.execInContainer(
- "bash",
- "-c",
- "mkdir -p /tmp/seatunnel/plugins/Jdbc/lib &&
cd /tmp/seatunnel/plugins/Jdbc/lib && wget "
- + MYSQL_DRIVER_JAR);
- Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
- };
+ MysqlDriverResolver::copyMySQLDriverToJdbcContainer;
@BeforeAll
@Override
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mongodb-e2e/src/test/java/mongodb/MongodbCDCMultiSourceIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mongodb-e2e/src/test/java/mongodb/MongodbCDCMultiSourceIT.java
index 840cab7f5d..98da8d02f5 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mongodb-e2e/src/test/java/mongodb/MongodbCDCMultiSourceIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mongodb-e2e/src/test/java/mongodb/MongodbCDCMultiSourceIT.java
@@ -32,7 +32,6 @@ import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.TestTemplate;
-import org.testcontainers.containers.Container;
import org.testcontainers.containers.output.Slf4jLogConsumer;
import org.testcontainers.lifecycle.Startables;
import org.testcontainers.utility.DockerLoggerFactory;
@@ -77,9 +76,6 @@ public class MongodbCDCMultiSourceIT extends TestSuiteBase
implements TestResour
private static final String MYSQL_USER_NAME = "st_user";
private static final String MYSQL_USER_PASSWORD = "seatunnel";
private static final String MYSQL_DATABASE = "mongodb_cdc";
- private static final String MYSQL_DRIVER_JAR =
-
"https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.16/mysql-connector-java-8.0.16.jar";;
-
private static final MySqlContainer MYSQL_CONTAINER =
createMySqlContainer();
private final UniqueDatabase database = new
UniqueDatabase(MYSQL_CONTAINER, MYSQL_DATABASE);
@@ -98,15 +94,7 @@ public class MongodbCDCMultiSourceIT extends TestSuiteBase
implements TestResour
@TestContainerExtension
private final ContainerExtendedFactory extendedFactory =
- container -> {
- Container.ExecResult extraCommands =
- container.execInContainer(
- "bash",
- "-c",
- "mkdir -p /tmp/seatunnel/plugins/Jdbc/lib &&
cd /tmp/seatunnel/plugins/Jdbc/lib && wget "
- + MYSQL_DRIVER_JAR);
- Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
- };
+ MysqlDriverResolver::copyMySQLDriverToJdbcContainer;
@BeforeAll
@Override
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mongodb-e2e/src/test/java/mongodb/MysqlDriverResolver.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mongodb-e2e/src/test/java/mongodb/MysqlDriverResolver.java
new file mode 100644
index 0000000000..34c3adc603
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mongodb-e2e/src/test/java/mongodb/MysqlDriverResolver.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package mongodb;
+
+import org.junit.jupiter.api.Assertions;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.utility.MountableFile;
+
+import com.mysql.cj.jdbc.Driver;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+
+final class MysqlDriverResolver {
+
+ private static final String JDBC_PLUGIN_LIB =
"/tmp/seatunnel/plugins/Jdbc/lib";
+
+ private MysqlDriverResolver() {}
+
+ static void copyMySQLDriverToJdbcContainer(GenericContainer<?> container)
+ throws IOException, InterruptedException {
+ Container.ExecResult mkdirResult =
+ container.execInContainer("bash", "-c", "mkdir -p " +
JDBC_PLUGIN_LIB);
+ Assertions.assertEquals(0, mkdirResult.getExitCode(),
mkdirResult.getStderr());
+
+ Path driverJarPath = mysqlDriverJarPath();
+ container.copyFileToContainer(
+ MountableFile.forHostPath(driverJarPath),
+ JDBC_PLUGIN_LIB + "/" + driverJarPath.getFileName());
+ }
+
+ private static Path mysqlDriverJarPath() {
+ try {
+ Path driverJarPath =
+ Paths.get(
+ Driver.class
+ .getProtectionDomain()
+ .getCodeSource()
+ .getLocation()
+ .toURI());
+ Assertions.assertTrue(Files.isRegularFile(driverJarPath));
+ return driverJarPath;
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to resolve MySQL JDBC driver jar from the test
classpath", e);
+ }
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/AbstractMysqlCDCITBase.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/AbstractMysqlCDCITBase.java
index aab29eca19..430478c64e 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/AbstractMysqlCDCITBase.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/AbstractMysqlCDCITBase.java
@@ -122,21 +122,9 @@ public abstract class AbstractMysqlCDCITBase extends
TestSuiteBase implements Te
new
Slf4jLogConsumer(DockerLoggerFactory.getLogger("mysql-docker-image")));
}
- private String driverUrl() {
- return
"https://repo1.maven.org/maven2/com/mysql/mysql-connector-j/8.0.32/mysql-connector-j-8.0.32.jar";;
- }
-
@TestContainerExtension
protected final ContainerExtendedFactory extendedFactory =
- container -> {
- Container.ExecResult extraCommands =
- container.execInContainer(
- "bash",
- "-c",
- "mkdir -p /tmp/seatunnel/plugins/MySQL-CDC/lib
&& cd /tmp/seatunnel/plugins/MySQL-CDC/lib && wget "
- + driverUrl());
- Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
- };
+ MysqlCDCDriverResolver::copyMySQLDriverToContainer;
@BeforeAll
@Override
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCDriverResolver.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCDriverResolver.java
new file mode 100644
index 0000000000..7861ddd925
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCDriverResolver.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.cdc.mysql;
+
+import org.junit.jupiter.api.Assertions;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.utility.MountableFile;
+
+import com.mysql.cj.jdbc.Driver;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+
+final class MysqlCDCDriverResolver {
+
+ private static final String MYSQL_CDC_PLUGIN_LIB =
"/tmp/seatunnel/plugins/MySQL-CDC/lib";
+
+ private MysqlCDCDriverResolver() {}
+
+ static void copyMySQLDriverToContainer(GenericContainer<?> container)
+ throws IOException, InterruptedException {
+ Container.ExecResult mkdirResult =
+ container.execInContainer("bash", "-c", "mkdir -p " +
MYSQL_CDC_PLUGIN_LIB);
+ Assertions.assertEquals(0, mkdirResult.getExitCode(),
mkdirResult.getStderr());
+
+ Path driverJarPath = mysqlDriverJarPath();
+ container.copyFileToContainer(
+ MountableFile.forHostPath(driverJarPath),
+ MYSQL_CDC_PLUGIN_LIB + "/" + driverJarPath.getFileName());
+ }
+
+ private static Path mysqlDriverJarPath() {
+ try {
+ Path driverJarPath =
+ Paths.get(
+ Driver.class
+ .getProtectionDomain()
+ .getCodeSource()
+ .getLocation()
+ .toURI());
+ Assertions.assertTrue(Files.isRegularFile(driverJarPath));
+ return driverJarPath;
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to resolve MySQL JDBC driver jar from the test
classpath", e);
+ }
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCSpecificStartingOffsetIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCSpecificStartingOffsetIT.java
index dfdf7a0413..66762b64d1 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCSpecificStartingOffsetIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCSpecificStartingOffsetIT.java
@@ -39,7 +39,6 @@ import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.TestTemplate;
-import org.testcontainers.containers.Container;
import org.testcontainers.containers.output.Slf4jLogConsumer;
import org.testcontainers.lifecycle.Startables;
import org.testcontainers.utility.DockerLoggerFactory;
@@ -122,21 +121,9 @@ public class MysqlCDCSpecificStartingOffsetIT extends
TestSuiteBase implements T
new
Slf4jLogConsumer(DockerLoggerFactory.getLogger("mysql-docker-image")));
}
- private String driverUrl() {
- return
"https://repo1.maven.org/maven2/com/mysql/mysql-connector-j/8.0.32/mysql-connector-j-8.0.32.jar";;
- }
-
@TestContainerExtension
protected final ContainerExtendedFactory extendedFactory =
- container -> {
- Container.ExecResult extraCommands =
- container.execInContainer(
- "bash",
- "-c",
- "mkdir -p /tmp/seatunnel/plugins/MySQL-CDC/lib
&& cd /tmp/seatunnel/plugins/MySQL-CDC/lib && wget "
- + driverUrl());
- Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
- };
+ MysqlCDCDriverResolver::copyMySQLDriverToContainer;
@BeforeAll
@Override
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithBinlogDeleteIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithBinlogDeleteIT.java
index 122b9cf483..4f0d35cf24 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithBinlogDeleteIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithBinlogDeleteIT.java
@@ -33,7 +33,6 @@ import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.TestTemplate;
-import org.testcontainers.containers.Container;
import org.testcontainers.containers.output.Slf4jLogConsumer;
import org.testcontainers.lifecycle.Startables;
import org.testcontainers.utility.DockerLoggerFactory;
@@ -104,21 +103,9 @@ public class MysqlCDCWithBinlogDeleteIT extends
TestSuiteBase implements TestRes
new
Slf4jLogConsumer(DockerLoggerFactory.getLogger("mysql-docker-image")));
}
- private String driverUrl() {
- return
"https://repo1.maven.org/maven2/com/mysql/mysql-connector-j/8.0.32/mysql-connector-j-8.0.32.jar";;
- }
-
@TestContainerExtension
protected final ContainerExtendedFactory extendedFactory =
- container -> {
- Container.ExecResult extraCommands =
- container.execInContainer(
- "bash",
- "-c",
- "mkdir -p /tmp/seatunnel/plugins/MySQL-CDC/lib
&& cd /tmp/seatunnel/plugins/MySQL-CDC/lib && wget "
- + driverUrl());
- Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
- };
+ MysqlCDCDriverResolver::copyMySQLDriverToContainer;
@BeforeAll
@Override
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithFlinkSchemaChangeIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithFlinkSchemaChangeIT.java
index 3ce40de783..a060f17ce5 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithFlinkSchemaChangeIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithFlinkSchemaChangeIT.java
@@ -35,7 +35,6 @@ import org.junit.jupiter.api.MethodOrderer;
import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.TestMethodOrder;
import org.junit.jupiter.api.TestTemplate;
-import org.testcontainers.containers.Container;
import org.testcontainers.containers.output.Slf4jLogConsumer;
import org.testcontainers.lifecycle.Startables;
import org.testcontainers.utility.DockerLoggerFactory;
@@ -65,6 +64,15 @@ import static org.awaitility.Awaitility.await;
disabledReason =
"Currently SPARK do not support cdc, only test the change
process related to Flink.")
public class MysqlCDCWithFlinkSchemaChangeIT extends TestSuiteBase implements
TestResource {
+ /**
+ * Flink schema evolution can restart after XA recover/rollback on loaded
CI runners, so these
+ * assertions need enough time for the job to recover and replay the
schema-change event.
+ */
+ private static final long SCHEMA_EVOLUTION_ASSERT_TIMEOUT_MILLIS =
300_000L;
+
+ private static final long STRUCTURE_AND_DATA_ASSERT_TIMEOUT_MILLIS =
300_000L;
+ private static final int MAX_TIMESTAMP_DRIFT_SECONDS = 60;
+
private static final String MYSQL_DATABASE = "shop";
private static final String SOURCE_TABLE = "products";
private static final String SINK_TABLE =
"mysql_cdc_e2e_sink_table_with_schema_change";
@@ -97,21 +105,9 @@ public class MysqlCDCWithFlinkSchemaChangeIT extends
TestSuiteBase implements Te
new
Slf4jLogConsumer(DockerLoggerFactory.getLogger("mysql-docker-image")));
}
- private String driverUrl() {
- return
"https://repo1.maven.org/maven2/com/mysql/mysql-connector-j/8.0.32/mysql-connector-j-8.0.32.jar";;
- }
-
@TestContainerExtension
protected final ContainerExtendedFactory extendedFactory =
- container -> {
- Container.ExecResult extraCommands =
- container.execInContainer(
- "bash",
- "-c",
- "mkdir -p /tmp/seatunnel/plugins/MySQL-CDC/lib
&& cd /tmp/seatunnel/plugins/MySQL-CDC/lib && wget "
- + driverUrl());
- Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
- };
+ MysqlCDCDriverResolver::copyMySQLDriverToContainer;
@Order(1)
@TestTemplate
@@ -169,7 +165,7 @@ public class MysqlCDCWithFlinkSchemaChangeIT extends
TestSuiteBase implements Te
}
private void assertSchemaEvolution(String database, String sourceTable,
String sinkTable) {
- await().atMost(180000, TimeUnit.MILLISECONDS)
+ await().atMost(SCHEMA_EVOLUTION_ASSERT_TIMEOUT_MILLIS,
TimeUnit.MILLISECONDS)
.untilAsserted(
() ->
assertTableDataEqualsBySourceColumnOrder(
@@ -177,12 +173,12 @@ public class MysqlCDCWithFlinkSchemaChangeIT extends
TestSuiteBase implements Te
// case1 add columns with cdc data at same time
shopDatabase.setTemplateName("add_columns").createAndInitialize();
- await().atMost(180000, TimeUnit.MILLISECONDS)
+ await().atMost(SCHEMA_EVOLUTION_ASSERT_TIMEOUT_MILLIS,
TimeUnit.MILLISECONDS)
.untilAsserted(
() ->
assertSchemaDescriptionEqualsIgnoringColumnOrder(
database, sourceTable, sinkTable));
- await().atMost(180000, TimeUnit.MILLISECONDS)
+ await().atMost(SCHEMA_EVOLUTION_ASSERT_TIMEOUT_MILLIS,
TimeUnit.MILLISECONDS)
.untilAsserted(
() -> {
assertTableDataEqualsBySourceColumnOrder(
@@ -209,8 +205,10 @@ public class MysqlCDCWithFlinkSchemaChangeIT extends
TestSuiteBase implements Te
while (resultSet.next()) {
int timeDiff =
resultSet.getInt("time_diff");
Assertions.assertTrue(
- timeDiff <= 60,
- "Time difference exceeds 60
seconds: "
+ timeDiff <=
MAX_TIMESTAMP_DRIFT_SECONDS,
+ "Time difference exceeds "
+ +
MAX_TIMESTAMP_DRIFT_SECONDS
+ + " seconds: "
+ timeDiff
+ " seconds");
}
@@ -246,7 +244,7 @@ public class MysqlCDCWithFlinkSchemaChangeIT extends
TestSuiteBase implements Te
private void assertSchemaEvolutionForAddColumns(
String database, String sourceTable, String sinkTable) {
- await().atMost(180000, TimeUnit.MILLISECONDS)
+ await().atMost(SCHEMA_EVOLUTION_ASSERT_TIMEOUT_MILLIS,
TimeUnit.MILLISECONDS)
.untilAsserted(
() ->
assertTableDataEqualsBySourceColumnOrder(
@@ -254,12 +252,12 @@ public class MysqlCDCWithFlinkSchemaChangeIT extends
TestSuiteBase implements Te
// case1 add columns with cdc data at same time
shopDatabase.setTemplateName("add_columns").createAndInitialize();
- await().atMost(180000, TimeUnit.MILLISECONDS)
+ await().atMost(SCHEMA_EVOLUTION_ASSERT_TIMEOUT_MILLIS,
TimeUnit.MILLISECONDS)
.untilAsserted(
() ->
assertSchemaDescriptionEqualsIgnoringColumnOrder(
database, sourceTable, sinkTable));
- await().atMost(180000, TimeUnit.MILLISECONDS)
+ await().atMost(SCHEMA_EVOLUTION_ASSERT_TIMEOUT_MILLIS,
TimeUnit.MILLISECONDS)
.untilAsserted(
() -> {
assertTableDataEqualsBySourceColumnOrder(
@@ -286,8 +284,10 @@ public class MysqlCDCWithFlinkSchemaChangeIT extends
TestSuiteBase implements Te
while (resultSet.next()) {
int timeDiff =
resultSet.getInt("time_diff");
Assertions.assertTrue(
- timeDiff <= 60,
- "Time difference exceeds 60
seconds: "
+ timeDiff <=
MAX_TIMESTAMP_DRIFT_SECONDS,
+ "Time difference exceeds "
+ +
MAX_TIMESTAMP_DRIFT_SECONDS
+ + " seconds: "
+ timeDiff
+ " seconds");
}
@@ -297,12 +297,12 @@ public class MysqlCDCWithFlinkSchemaChangeIT extends
TestSuiteBase implements Te
private void assertTableStructureAndData(
String database, String sourceTable, String sinkTable) {
- await().atMost(300000, TimeUnit.MILLISECONDS)
+ await().atMost(STRUCTURE_AND_DATA_ASSERT_TIMEOUT_MILLIS,
TimeUnit.MILLISECONDS)
.untilAsserted(
() ->
assertSchemaDescriptionEqualsIgnoringColumnOrder(
database, sourceTable, sinkTable));
- await().atMost(300000, TimeUnit.MILLISECONDS)
+ await().atMost(STRUCTURE_AND_DATA_ASSERT_TIMEOUT_MILLIS,
TimeUnit.MILLISECONDS)
.untilAsserted(
() ->
assertTableDataEqualsBySourceColumnOrder(
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithMissingBinlogDiagnosticsIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithMissingBinlogDiagnosticsIT.java
index 7f0929b133..06b9a3ac6d 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithMissingBinlogDiagnosticsIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithMissingBinlogDiagnosticsIT.java
@@ -86,21 +86,9 @@ public class MysqlCDCWithMissingBinlogDiagnosticsIT extends
TestSuiteBase implem
new
Slf4jLogConsumer(DockerLoggerFactory.getLogger("mysql-docker-image")));
}
- private String driverUrl() {
- return
"https://repo1.maven.org/maven2/com/mysql/mysql-connector-j/8.0.32/mysql-connector-j-8.0.32.jar";;
- }
-
@TestContainerExtension
protected final ContainerExtendedFactory extendedFactory =
- container -> {
- Container.ExecResult extraCommands =
- container.execInContainer(
- "bash",
- "-c",
- "mkdir -p /tmp/seatunnel/plugins/MySQL-CDC/lib
&& cd /tmp/seatunnel/plugins/MySQL-CDC/lib && wget "
- + driverUrl());
- Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
- };
+ MysqlCDCDriverResolver::copyMySQLDriverToContainer;
@BeforeAll
@Override
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithSchemaChangeIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithSchemaChangeIT.java
index 71fd28215a..8999da08f8 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithSchemaChangeIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithSchemaChangeIT.java
@@ -36,7 +36,6 @@ import org.junit.jupiter.api.MethodOrderer;
import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.TestMethodOrder;
import org.junit.jupiter.api.TestTemplate;
-import org.testcontainers.containers.Container;
import org.testcontainers.containers.output.Slf4jLogConsumer;
import org.testcontainers.lifecycle.Startables;
import org.testcontainers.utility.DockerLoggerFactory;
@@ -113,21 +112,9 @@ public class MysqlCDCWithSchemaChangeIT extends
TestSuiteBase implements TestRes
new
Slf4jLogConsumer(DockerLoggerFactory.getLogger("mysql-docker-image")));
}
- private String driverUrl() {
- return
"https://repo1.maven.org/maven2/com/mysql/mysql-connector-j/8.0.32/mysql-connector-j-8.0.32.jar";;
- }
-
@TestContainerExtension
protected final ContainerExtendedFactory extendedFactory =
- container -> {
- Container.ExecResult extraCommands =
- container.execInContainer(
- "bash",
- "-c",
- "mkdir -p /tmp/seatunnel/plugins/MySQL-CDC/lib
&& cd /tmp/seatunnel/plugins/MySQL-CDC/lib && wget "
- + driverUrl());
- Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
- };
+ MysqlCDCDriverResolver::copyMySQLDriverToContainer;
@Order(1)
@TestTemplate
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-opengauss-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/OpengaussCDCIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-opengauss-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/OpengaussCDCIT.java
index a22b2b5795..65a2063171 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-opengauss-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/OpengaussCDCIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-opengauss-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/OpengaussCDCIT.java
@@ -38,6 +38,7 @@ import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.output.Slf4jLogConsumer;
import org.testcontainers.lifecycle.Startables;
import org.testcontainers.utility.DockerImageName;
+import org.testcontainers.utility.MountableFile;
import lombok.extern.slf4j.Slf4j;
@@ -45,6 +46,7 @@ import java.io.IOException;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
+import java.nio.file.Path;
import java.nio.file.Paths;
import java.sql.Connection;
import java.sql.DriverManager;
@@ -93,6 +95,7 @@ public class OpengaussCDCIT extends TestSuiteBase implements
TestResource {
protected static final DockerImageName OPENGAUSS_IMAGE =
DockerImageName.parse("opengauss/opengauss:5.0.0")
.asCompatibleSubstituteFor("postgres");
+ private static final String JDBC_PLUGIN_LIB =
"/tmp/seatunnel/plugins/JDBC/lib";
private static final String SOURCE_SQL_TEMPLATE = "select * from %s.%s
order by id";
@@ -103,22 +106,36 @@ public class OpengaussCDCIT extends TestSuiteBase
implements TestResource {
.withEnv("GS_PASSWORD", PASSWORD)
.withLogConsumer(new Slf4jLogConsumer(log));
- private String driverUrl() {
- return
"https://repo1.maven.org/maven2/org/postgresql/postgresql/42.5.1/postgresql-42.5.1.jar";;
- }
-
@TestContainerExtension
protected final ContainerExtendedFactory extendedFactory =
container -> {
Container.ExecResult extraCommands =
- container.execInContainer(
- "bash",
- "-c",
- "mkdir -p /tmp/seatunnel/plugins/JDBC/lib &&
cd /tmp/seatunnel/plugins/JDBC/lib && wget "
- + driverUrl());
+ container.execInContainer("bash", "-c", "mkdir -p " +
JDBC_PLUGIN_LIB);
Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
+
+ Path driverJarPath = postgresDriverJarPath();
+ container.copyFileToContainer(
+ MountableFile.forHostPath(driverJarPath),
+ JDBC_PLUGIN_LIB + "/" + driverJarPath.getFileName());
};
+ private Path postgresDriverJarPath() {
+ try {
+ Path driverJarPath =
+ Paths.get(
+ org.postgresql.Driver.class
+ .getProtectionDomain()
+ .getCodeSource()
+ .getLocation()
+ .toURI());
+ Assertions.assertTrue(Files.isRegularFile(driverJarPath));
+ return driverJarPath;
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to resolve PostgreSQL JDBC driver jar from the
test classpath", e);
+ }
+ }
+
@BeforeAll
@Override
public void startUp() throws Exception {
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/AbstractOracleCDCIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/AbstractOracleCDCIT.java
index 6c97f1e50e..5cbaf5cf50 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/AbstractOracleCDCIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/AbstractOracleCDCIT.java
@@ -19,16 +19,22 @@ package
org.apache.seatunnel.connectors.seatunnel.cdc.oracle;
import org.apache.seatunnel.e2e.common.TestSuiteBase;
+import org.junit.jupiter.api.Assertions;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.output.Slf4jLogConsumer;
import org.testcontainers.utility.DockerLoggerFactory;
+import org.testcontainers.utility.MountableFile;
import lombok.extern.slf4j.Slf4j;
import oracle.sql.TIMESTAMP;
import oracle.sql.TIMESTAMPLTZ;
+import java.io.IOException;
import java.math.BigDecimal;
import java.net.URL;
import java.nio.file.Files;
+import java.nio.file.Path;
import java.nio.file.Paths;
import java.sql.Connection;
import java.sql.DriverManager;
@@ -80,6 +86,8 @@ public class AbstractOracleCDCIT extends TestSuiteBase {
protected static final String SOURCE_TABLE1 = "FULL_TYPES";
protected static final String SOURCE_TABLE2 = "FULL_TYPES2";
+ private static final String ORACLE_CDC_PLUGIN_LIB =
"/tmp/seatunnel/plugins/Oracle-CDC/lib";
+ private static final String JDBC_PLUGIN_LIB =
"/tmp/seatunnel/plugins/Jdbc/lib";
protected static final OracleContainer ORACLE_CONTAINER =
new OracleContainer(getImage())
@@ -102,14 +110,65 @@ public class AbstractOracleCDCIT extends TestSuiteBase {
}
}
- protected String oracleDriverUrl() {
- return
"https://repo1.maven.org/maven2/com/oracle/database/jdbc/ojdbc8/12.2.0.1/ojdbc8-12.2.0.1.jar";;
- }
-
static {
System.setProperty("oracle.jdbc.timezoneAsRegion", "false");
}
+ protected void copyOracleDriverToContainer(GenericContainer<?> container)
+ throws IOException, InterruptedException {
+ copyDriverToContainer(container, ORACLE_CDC_PLUGIN_LIB,
oracleDriverJarPath());
+ }
+
+ protected void copyMySQLDriverToJdbcContainer(GenericContainer<?>
container)
+ throws IOException, InterruptedException {
+ copyDriverToContainer(container, JDBC_PLUGIN_LIB,
mysqlDriverJarPath());
+ }
+
+ private void copyDriverToContainer(
+ GenericContainer<?> container, String pluginLibDir, Path
driverJarPath)
+ throws IOException, InterruptedException {
+ Container.ExecResult mkdirResult =
+ container.execInContainer("bash", "-c", "mkdir -p " +
pluginLibDir);
+ Assertions.assertEquals(0, mkdirResult.getExitCode(),
mkdirResult.getStderr());
+ container.copyFileToContainer(
+ MountableFile.forHostPath(driverJarPath),
+ pluginLibDir + "/" + driverJarPath.getFileName());
+ }
+
+ private Path oracleDriverJarPath() {
+ try {
+ Path driverJarPath =
+ Paths.get(
+ oracle.jdbc.driver.OracleDriver.class
+ .getProtectionDomain()
+ .getCodeSource()
+ .getLocation()
+ .toURI());
+ Assertions.assertTrue(Files.isRegularFile(driverJarPath));
+ return driverJarPath;
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to resolve Oracle JDBC driver jar from the test
classpath", e);
+ }
+ }
+
+ private Path mysqlDriverJarPath() {
+ try {
+ Path driverJarPath =
+ Paths.get(
+ com.mysql.cj.jdbc.Driver.class
+ .getProtectionDomain()
+ .getCodeSource()
+ .getLocation()
+ .toURI());
+ Assertions.assertTrue(Files.isRegularFile(driverJarPath));
+ return driverJarPath;
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to resolve MySQL JDBC driver jar from the test
classpath", e);
+ }
+ }
+
protected static void createAndInitialize(String sqlFile, String username,
String password)
throws Exception {
final String ddlFile = String.format("ddl/%s.sql", sqlFile);
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/OracleCDCIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/OracleCDCIT.java
index adcacd89fa..e8b1208bdb 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/OracleCDCIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/OracleCDCIT.java
@@ -63,16 +63,7 @@ public class OracleCDCIT extends AbstractOracleCDCIT
implements TestResource {
private static final String SOURCE_SQL_TEMPLATE = "select * from %s.%s
ORDER BY ID";
@TestContainerExtension
- protected final ContainerExtendedFactory extendedFactory =
- container -> {
- Container.ExecResult extraCommands =
- container.execInContainer(
- "bash",
- "-c",
- "mkdir -p
/tmp/seatunnel/plugins/Oracle-CDC/lib && cd
/tmp/seatunnel/plugins/Oracle-CDC/lib && wget "
- + oracleDriverUrl());
- Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
- };
+ protected final ContainerExtendedFactory extendedFactory =
this::copyOracleDriverToContainer;
@BeforeAll
@Override
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/OracleCDCWithSchemaChangeIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/OracleCDCWithSchemaChangeIT.java
index 2466aa023f..cf4e69476f 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/OracleCDCWithSchemaChangeIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-oracle-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/OracleCDCWithSchemaChangeIT.java
@@ -47,7 +47,6 @@ import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.TestMethodOrder;
import org.junit.jupiter.api.TestTemplate;
import org.junit.platform.commons.util.StringUtils;
-import org.testcontainers.containers.Container;
import org.testcontainers.containers.output.Slf4jLogConsumer;
import org.testcontainers.lifecycle.Startables;
import org.testcontainers.utility.DockerLoggerFactory;
@@ -121,22 +120,11 @@ public class OracleCDCWithSchemaChangeIT extends
AbstractOracleCDCIT implements
new
Slf4jLogConsumer(DockerLoggerFactory.getLogger("mysql-docker-image")));
}
- private String mysqlDriverUrl() {
- return
"https://repo1.maven.org/maven2/com/mysql/mysql-connector-j/8.0.32/mysql-connector-j-8.0.32.jar";;
- }
-
@TestContainerExtension
protected final ContainerExtendedFactory extendedFactory =
container -> {
- Container.ExecResult extraCommands =
- container.execInContainer(
- "bash",
- "-c",
- "mkdir -p
/tmp/seatunnel/plugins/Oracle-CDC/lib && cd
/tmp/seatunnel/plugins/Oracle-CDC/lib && wget "
- + oracleDriverUrl()
- + " && mkdir -p
/tmp/seatunnel/plugins/Jdbc/lib && cd /tmp/seatunnel/plugins/Jdbc/lib && wget "
- + mysqlDriverUrl());
- Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
+ copyOracleDriverToContainer(container);
+ copyMySQLDriverToJdbcContainer(container);
};
@BeforeAll
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/PostgresCDCIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/PostgresCDCIT.java
index 652b3438ce..1582616e52 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/PostgresCDCIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/PostgresCDCIT.java
@@ -59,6 +59,7 @@ import org.testcontainers.containers.output.Slf4jLogConsumer;
import org.testcontainers.lifecycle.Startables;
import org.testcontainers.utility.DockerImageName;
import org.testcontainers.utility.DockerLoggerFactory;
+import org.testcontainers.utility.MountableFile;
import io.debezium.jdbc.JdbcConnection;
import io.debezium.relational.TableId;
@@ -68,6 +69,7 @@ import java.io.IOException;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
+import java.nio.file.Path;
import java.nio.file.Paths;
import java.sql.Connection;
import java.sql.DriverManager;
@@ -148,6 +150,7 @@ public class PostgresCDCIT extends TestSuiteBase implements
TestResource {
// when testing postgres 13, only 13-alpine supports both amd64 and arm64
protected static final DockerImageName PG_IMAGE =
DockerImageName.parse("debezium/postgres:11").asCompatibleSubstituteFor("postgres");
+ private static final String POSTGRES_CDC_PLUGIN_LIB =
"/tmp/seatunnel/plugins/Postgres-CDC/lib";
public static final PostgreSQLContainer<?> POSTGRES_CONTAINER =
new PostgreSQLContainer<>(PG_IMAGE)
@@ -175,22 +178,37 @@ public class PostgresCDCIT extends TestSuiteBase
implements TestResource {
DockerLoggerFactory.getLogger(KAFKA_IMAGE_NAME)));
}
- private String driverUrl() {
- return
"https://repo1.maven.org/maven2/org/postgresql/postgresql/42.5.1/postgresql-42.5.1.jar";;
- }
-
@TestContainerExtension
protected final ContainerExtendedFactory extendedFactory =
container -> {
Container.ExecResult extraCommands =
container.execInContainer(
- "bash",
- "-c",
- "mkdir -p
/tmp/seatunnel/plugins/Postgres-CDC/lib && cd
/tmp/seatunnel/plugins/Postgres-CDC/lib && wget "
- + driverUrl());
+ "bash", "-c", "mkdir -p " +
POSTGRES_CDC_PLUGIN_LIB);
Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
+
+ Path driverJarPath = postgresDriverJarPath();
+ container.copyFileToContainer(
+ MountableFile.forHostPath(driverJarPath),
+ POSTGRES_CDC_PLUGIN_LIB + "/" +
driverJarPath.getFileName());
};
+ private Path postgresDriverJarPath() {
+ try {
+ Path driverJarPath =
+ Paths.get(
+ org.postgresql.Driver.class
+ .getProtectionDomain()
+ .getCodeSource()
+ .getLocation()
+ .toURI());
+ Assertions.assertTrue(Files.isRegularFile(driverJarPath));
+ return driverJarPath;
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to resolve PostgreSQL JDBC driver jar from the
test classpath", e);
+ }
+ }
+
@BeforeAll
@Override
public void startUp() {
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-tidb-e2e/pom.xml
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-tidb-e2e/pom.xml
index c4a77b58fb..4d9119ccee 100644
--- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-tidb-e2e/pom.xml
+++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-tidb-e2e/pom.xml
@@ -29,6 +29,7 @@
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
</properties>
+
<dependencies>
<!-- SeaTunnel connectors -->
<dependency>
@@ -55,6 +56,12 @@
<version>${project.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.tikv</groupId>
+ <artifactId>tikv-client-java</artifactId>
+ <version>3.2.0</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-tidb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/tidb/TiDBCDCIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-tidb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/tidb/TiDBCDCIT.java
index 78c7077940..7b102203c2 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-tidb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/tidb/TiDBCDCIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-tidb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/tidb/TiDBCDCIT.java
@@ -31,10 +31,17 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.TestTemplate;
import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.utility.MountableFile;
+import org.tikv.common.TiSession;
+import com.mysql.cj.jdbc.Driver;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
@@ -76,37 +83,59 @@ public class TiDBCDCIT extends TiDBTestBase implements
TestResource {
+ " f_text, f_tinytext, f_varchar, f_date, f_datetime,
f_timestamp, f_bit1, f_bit64, f_char,"
+ " f_enum, f_mediumblob, f_long_varchar, f_real, f_time,
f_tinyint, f_tinyint_unsigned,"
+ " f_json, cast(f_year as year) from %s.%s";
-
- private String driverUrl() {
- return
"https://repo1.maven.org/maven2/com/mysql/mysql-connector-j/8.0.32/mysql-connector-j-8.0.32.jar";;
- }
-
- private String tiKVUrl() {
- return
"https://repo1.maven.org/maven2/org/tikv/tikv-client-java/3.2.0/tikv-client-java-3.2.0.jar";;
- }
+ private static final String TIDB_CDC_PLUGIN_LIB =
"/tmp/seatunnel/plugins/TiDB-CDC/lib";
@TestContainerExtension
protected final ContainerExtendedFactory extendedFactory =
container -> {
Container.ExecResult extraCommands =
- container.execInContainer(
- "bash",
- "-c",
- "mkdir -p /tmp/seatunnel/plugins/TiDB-CDC/lib
&& cd "
- + "/tmp/seatunnel/plugins/TiDB-CDC/lib
&& wget "
- + driverUrl());
+ container.execInContainer("bash", "-c", "mkdir -p " +
TIDB_CDC_PLUGIN_LIB);
Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
- Container.ExecResult extraCommands2 =
- container.execInContainer(
- "bash",
- "-c",
- "mkdir -p /tmp/seatunnel/plugins/TiDB-CDC/lib
&& cd "
- + "/tmp/seatunnel/plugins/TiDB-CDC/lib
&& wget "
- + tiKVUrl());
- Assertions.assertEquals(
- 0, extraCommands2.getExitCode(),
extraCommands2.getStderr());
+
+ copyDriverToContainer(container, mysqlDriverJarPath());
+ copyDriverToContainer(container, tiKVClientJarPath());
};
+ private void copyDriverToContainer(GenericContainer<?> container, Path
driverJarPath) {
+ container.copyFileToContainer(
+ MountableFile.forHostPath(driverJarPath),
+ TIDB_CDC_PLUGIN_LIB + "/" + driverJarPath.getFileName());
+ }
+
+ private Path mysqlDriverJarPath() {
+ try {
+ Path driverJarPath =
+ Paths.get(
+ Driver.class
+ .getProtectionDomain()
+ .getCodeSource()
+ .getLocation()
+ .toURI());
+ Assertions.assertTrue(Files.isRegularFile(driverJarPath));
+ return driverJarPath;
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to resolve MySQL JDBC driver jar from the test
classpath", e);
+ }
+ }
+
+ private Path tiKVClientJarPath() {
+ try {
+ Path driverJarPath =
+ Paths.get(
+ TiSession.class
+ .getProtectionDomain()
+ .getCodeSource()
+ .getLocation()
+ .toURI());
+ Assertions.assertTrue(Files.isRegularFile(driverJarPath));
+ return driverJarPath;
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to resolve TiKV client jar from the test
classpath", e);
+ }
+ }
+
@BeforeAll
@Override
public void startUp() throws Exception {
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-databend-e2e/src/test/java/org/apache/seatunnel/e2e/connector/databend/DatabendCDCSinkIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-databend-e2e/src/test/java/org/apache/seatunnel/e2e/connector/databend/DatabendCDCSinkIT.java
index 0811255e64..ad13d9bc8f 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-databend-e2e/src/test/java/org/apache/seatunnel/e2e/connector/databend/DatabendCDCSinkIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-databend-e2e/src/test/java/org/apache/seatunnel/e2e/connector/databend/DatabendCDCSinkIT.java
@@ -63,6 +63,9 @@ public class DatabendCDCSinkIT extends TestSuiteBase
implements TestResource {
private static final int PORT = 8000;
private static final int LOCAL_PORT = 8000;
private static final String DATABASE = "default";
+ private static final String DATABEND_CDC_JOB_CONFIG =
"/databend/fake_to_databend_cdc.conf";
+ private static final int MAX_JOB_SUBMIT_ATTEMPTS = 2;
+ private static final int JOB_SUBMIT_RETRY_INTERVAL_SECONDS = 10;
private DatabendContainer container;
private GenericContainer<?> minioContainer;
private Connection connection;
@@ -70,8 +73,7 @@ public class DatabendCDCSinkIT extends TestSuiteBase
implements TestResource {
@TestTemplate
public void testDatabendSinkCDC(TestContainer container) throws Exception {
// Run the CDC test job
- Container.ExecResult execResult =
- container.executeJob("/databend/fake_to_databend_cdc.conf");
+ Container.ExecResult execResult = executeDatabendCdcJob(container);
Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
Awaitility.await()
@@ -164,6 +166,32 @@ public class DatabendCDCSinkIT extends TestSuiteBase
implements TestResource {
clearSinkTable();
}
+ private Container.ExecResult executeDatabendCdcJob(TestContainer
container) throws Exception {
+ Container.ExecResult execResult = null;
+ for (int attempt = 1; attempt <= MAX_JOB_SUBMIT_ATTEMPTS; attempt++) {
+ execResult = container.executeJob(DATABEND_CDC_JOB_CONFIG);
+ if (execResult.getExitCode() == 0 ||
!isFlinkResourceNotReady(execResult)) {
+ return execResult;
+ }
+ if (attempt < MAX_JOB_SUBMIT_ATTEMPTS) {
+ LOG.warn(
+ "Databend CDC job failed because Flink resources were
not ready,"
+ + " retrying job submission ({}/{})",
+ attempt,
+ MAX_JOB_SUBMIT_ATTEMPTS);
+ TimeUnit.SECONDS.sleep(JOB_SUBMIT_RETRY_INTERVAL_SECONDS);
+ }
+ }
+ return execResult;
+ }
+
+ private boolean isFlinkResourceNotReady(Container.ExecResult execResult) {
+ String stderr = execResult.getStderr();
+ return stderr != null
+ && (stderr.contains("NoResourceAvailableException")
+ || stderr.contains("Could not acquire the minimum
required resources"));
+ }
+
private void clearSinkTable() throws SQLException {
try (Statement statement = connection.createStatement()) {
statement.execute("TRUNCATE TABLE sink_table");
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/AbstractDorisIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/AbstractDorisIT.java
index dcd0f6f7bd..4626eef4a1 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/AbstractDorisIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/AbstractDorisIT.java
@@ -25,14 +25,20 @@ import org.apache.seatunnel.e2e.common.TestSuiteBase;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
+import org.testcontainers.containers.Container;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.lifecycle.Startables;
+import org.testcontainers.utility.MountableFile;
import lombok.extern.slf4j.Slf4j;
+import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URL;
import java.net.URLClassLoader;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
import java.sql.Connection;
import java.sql.Driver;
import java.sql.ResultSet;
@@ -70,8 +76,6 @@ public abstract class AbstractDorisIT extends TestSuiteBase
implements TestResou
private static final String DROP_BE = "ALTER SYSTEM DROPP BACKEND
\"127.0.0.1:9050\"";
private static final String ADD_BE = "ALTER SYSTEM ADD BACKEND
\"%s:9050\"";
protected static final String DRIVER_CLASS = "com.mysql.cj.jdbc.Driver";
- protected static final String DRIVER_JAR =
-
"https://repo1.maven.org/maven2/com/mysql/mysql-connector-j/8.0.32/mysql-connector-j-8.0.32.jar";;
@BeforeAll
@Override
@@ -100,7 +104,9 @@ public abstract class AbstractDorisIT extends TestSuiteBase
implements TestResou
InstantiationException, IllegalAccessException {
log.info("doris initializing ...");
URLClassLoader urlClassLoader =
- new URLClassLoader(new URL[] {new URL(DRIVER_JAR)},
DorisIT.class.getClassLoader());
+ new URLClassLoader(
+ new URL[] {mysqlDriverJarPath().toUri().toURL()},
+ DorisIT.class.getClassLoader());
Thread.currentThread().setContextClassLoader(urlClassLoader);
Driver driver = (Driver)
urlClassLoader.loadClass(DRIVER_CLASS).newInstance();
Properties props = new Properties();
@@ -121,6 +127,35 @@ public abstract class AbstractDorisIT extends
TestSuiteBase implements TestResou
}
}
+ protected Path mysqlDriverJarPath() {
+ try {
+ return Paths.get(
+ com.mysql.cj.jdbc.Driver.class
+ .getProtectionDomain()
+ .getCodeSource()
+ .getLocation()
+ .toURI());
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to resolve MySQL JDBC driver jar from the test
classpath", e);
+ }
+ }
+
+ protected void copyMySQLDriverToContainer(GenericContainer<?> container,
String pluginLibDir)
+ throws IOException, InterruptedException {
+ Path driverJarPath = mysqlDriverJarPath();
+ Assertions.assertTrue(
+ Files.isRegularFile(driverJarPath),
+ "MySQL JDBC driver should be resolved from the test classpath
before E2E runs: "
+ + driverJarPath);
+ Container.ExecResult extraCommands =
+ container.execInContainer("bash", "-c", "mkdir -p " +
pluginLibDir);
+ Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
+ container.copyFileToContainer(
+ MountableFile.forHostPath(driverJarPath),
+ pluginLibDir + "/" + driverJarPath.getFileName());
+ }
+
// The Host of the official image [apache/doris:doris-all-in-one-2.1.0] BE
is 127.0.0.1, causing
// cross-container access failure. Delete the BE and add it again
private void initializeBE() {
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisCDCSinkIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisCDCSinkIT.java
index d13ff4a501..564a5820c5 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisCDCSinkIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisCDCSinkIT.java
@@ -30,7 +30,6 @@ import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.TestTemplate;
-import org.testcontainers.containers.Container;
import org.testcontainers.containers.output.Slf4jLogConsumer;
import org.testcontainers.lifecycle.Startables;
import org.testcontainers.utility.DockerLoggerFactory;
@@ -90,13 +89,7 @@ public class DorisCDCSinkIT extends AbstractDorisIT {
@TestContainerExtension
protected final ContainerExtendedFactory extendedFactory =
container -> {
- Container.ExecResult extraCommands =
- container.execInContainer(
- "bash",
- "-c",
- "mkdir -p /tmp/seatunnel/plugins/Doris-CDC/lib
&& cd /tmp/seatunnel/plugins/Doris-CDC/lib && wget "
- + driverUrl());
- Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
+ copyMySQLDriverToContainer(container,
"/tmp/seatunnel/plugins/Doris-CDC/lib");
};
private final UniqueDatabase inventoryDatabase =
@@ -116,10 +109,6 @@ public class DorisCDCSinkIT extends AbstractDorisIT {
new
Slf4jLogConsumer(DockerLoggerFactory.getLogger("mysql-docker-image")));
}
- private String driverUrl() {
- return
"https://repo1.maven.org/maven2/com/mysql/mysql-connector-j/8.0.32/mysql-connector-j-8.0.32.jar";;
- }
-
@BeforeAll
public void init() {
log.info("The second stage: Starting Mysql containers...");
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisErrorIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisErrorIT.java
index 517b600533..5661360418 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisErrorIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisErrorIT.java
@@ -47,15 +47,7 @@ public class DorisErrorIT extends AbstractDorisIT {
@TestContainerExtension
protected final ContainerExtendedFactory extendedFactory =
- container -> {
- Container.ExecResult extraCommands =
- container.execInContainer(
- "bash",
- "-c",
- "mkdir -p /tmp/seatunnel/plugins/jdbc/lib &&
cd /tmp/seatunnel/plugins/jdbc/lib && wget "
- + DRIVER_JAR);
- Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
- };
+ container -> copyMySQLDriverToContainer(container,
"/tmp/seatunnel/plugins/jdbc/lib");
@TestTemplate
@DisabledOnContainer(
@@ -74,8 +66,13 @@ public class DorisErrorIT extends AbstractDorisIT {
throw new RuntimeException(e);
}
});
- // wait for the job to start
- Thread.sleep(10 * 1000);
+ // Keep the Doris sink job running before stopping Doris so this test
exercises the
+ // stream-load failure path instead of an early submission failure.
+ given().pollInterval(1, TimeUnit.SECONDS)
+ .await()
+ .during(10, TimeUnit.SECONDS)
+ .atMost(15, TimeUnit.SECONDS)
+ .until(() -> !future.isDone());
super.container.stop();
Assertions.assertNotEquals(0, future.get().getExitCode());
Assertions.assertTrue(
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisIT.java
index f08df89760..3058292296 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisIT.java
@@ -160,15 +160,7 @@ public class DorisIT extends AbstractDorisIT {
@TestContainerExtension
protected final ContainerExtendedFactory extendedFactory =
- container -> {
- Container.ExecResult extraCommands =
- container.execInContainer(
- "bash",
- "-c",
- "mkdir -p /tmp/seatunnel/plugins/jdbc/lib &&
cd /tmp/seatunnel/plugins/jdbc/lib && wget "
- + DRIVER_JAR);
- Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
- };
+ container -> copyMySQLDriverToContainer(container,
"/tmp/seatunnel/plugins/jdbc/lib");
@TestTemplate
public void testCustomSql(TestContainer container) throws IOException,
InterruptedException {
@@ -419,7 +411,8 @@ public class DorisIT extends AbstractDorisIT {
try {
URLClassLoader urlClassLoader =
new URLClassLoader(
- new URL[] {new URL(DRIVER_JAR)},
DorisIT.class.getClassLoader());
+ new URL[] {mysqlDriverJarPath().toUri().toURL()},
+ DorisIT.class.getClassLoader());
Thread.currentThread().setContextClassLoader(urlClassLoader);
Driver driver = (Driver)
urlClassLoader.loadClass(DRIVER_CLASS).newInstance();
Properties props = new Properties();
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisMultiReadIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisMultiReadIT.java
index 8265e2d0e0..dc1aab6d52 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisMultiReadIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisMultiReadIT.java
@@ -116,15 +116,7 @@ public class DorisMultiReadIT extends AbstractDorisIT {
@TestContainerExtension
protected final ContainerExtendedFactory extendedFactory =
- container -> {
- Container.ExecResult extraCommands =
- container.execInContainer(
- "bash",
- "-c",
- "mkdir -p /tmp/seatunnel/plugins/jdbc/lib &&
cd /tmp/seatunnel/plugins/jdbc/lib && wget "
- + DRIVER_JAR);
- Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
- };
+ container -> copyMySQLDriverToContainer(container,
"/tmp/seatunnel/plugins/jdbc/lib");
@TestTemplate
public void testDorisMultiRead(TestContainer container)
@@ -343,7 +335,7 @@ public class DorisMultiReadIT extends AbstractDorisIT {
try {
URLClassLoader urlClassLoader =
new URLClassLoader(
- new URL[] {new URL(DRIVER_JAR)},
+ new URL[] {mysqlDriverJarPath().toUri().toURL()},
DorisMultiReadIT.class.getClassLoader());
Thread.currentThread().setContextClassLoader(urlClassLoader);
Driver driver = (Driver)
urlClassLoader.loadClass(DRIVER_CLASS).newInstance();
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisSchemaChangeIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisSchemaChangeIT.java
index 0eeb2cdbea..501c4d902a 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisSchemaChangeIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisSchemaChangeIT.java
@@ -31,7 +31,6 @@ import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.TestTemplate;
-import org.testcontainers.containers.Container;
import org.testcontainers.containers.output.Slf4jLogConsumer;
import org.testcontainers.lifecycle.Startables;
import org.testcontainers.utility.DockerLoggerFactory;
@@ -84,15 +83,7 @@ public class DorisSchemaChangeIT extends AbstractDorisIT {
@TestContainerExtension
private final ContainerExtendedFactory extendedFactory =
- container -> {
- Container.ExecResult extraCommands =
- container.execInContainer(
- "bash",
- "-c",
- "mkdir -p /tmp/seatunnel/plugins/Jdbc/lib &&
cd /tmp/seatunnel/plugins/Jdbc/lib && curl -O "
- + DRIVER_JAR);
- Assertions.assertEquals(0, extraCommands.getExitCode());
- };
+ container -> copyMySQLDriverToContainer(container,
"/tmp/seatunnel/plugins/Jdbc/lib");
private static MySqlContainer createMySqlContainer(MySqlVersion version) {
MySqlContainer mySqlContainer =
@@ -124,7 +115,6 @@ public class DorisSchemaChangeIT extends AbstractDorisIT {
throw new RuntimeException(e);
}
});
- TimeUnit.SECONDS.sleep(20);
// waiting for case1 completed
assertSchemaEvolutionForAddColumns(
DATABASE, SOURCE_TABLE, SINK_TABLE, mysqlConnection,
jdbcConnection);
@@ -183,7 +173,8 @@ public class DorisSchemaChangeIT extends AbstractDorisIT {
String sinkTable,
Connection sourceConnection,
Connection sinkConnection) {
- await().atMost(60000, TimeUnit.MILLISECONDS)
+ await().ignoreExceptions()
+ .atMost(60000, TimeUnit.MILLISECONDS)
.untilAsserted(
() ->
Assertions.assertIterableEquals(
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-edge-socket-e2e/src/test/java/org/apache/seatunnel/e2e/connector/edgesocket/AbstractEdgeSocketIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-edge-socket-e2e/src/test/java/org/apache/seatunnel/e2e/connector/edgesocket/AbstractEdgeSocketIT.java
index 60cbe7dec8..b1f6a90295 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-edge-socket-e2e/src/test/java/org/apache/seatunnel/e2e/connector/edgesocket/AbstractEdgeSocketIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-edge-socket-e2e/src/test/java/org/apache/seatunnel/e2e/connector/edgesocket/AbstractEdgeSocketIT.java
@@ -27,7 +27,9 @@ import org.junit.jupiter.api.BeforeAll;
import org.testcontainers.containers.Container;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.utility.DockerImageName;
+import org.testcontainers.utility.MountableFile;
+import com.mysql.cj.jdbc.Driver;
import lombok.extern.slf4j.Slf4j;
import javax.crypto.Cipher;
@@ -42,6 +44,9 @@ import java.io.OutputStreamWriter;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
import java.security.SecureRandom;
import java.util.ArrayList;
import java.util.Base64;
@@ -64,6 +69,7 @@ public abstract class AbstractEdgeSocketIT extends
TestSuiteBase implements Test
protected static final String TRANSFORM_SUFFIX = "_transformed";
protected static final String EDGE_BATCH_PREFIX = "__BATCH__:";
protected static final String EDGE_COMMIT_PREFIX = "__COMMIT__:";
+ private static final String JDBC_PLUGIN_LIB =
"/tmp/seatunnel/plugins/Jdbc/lib";
protected GenericContainer<?> edgeSocketForwarderContainer;
private String edgeSocketForwarderTargetHost;
@@ -74,6 +80,35 @@ public abstract class AbstractEdgeSocketIT extends
TestSuiteBase implements Test
protected abstract List<String> querySinkValues() throws Exception;
+ protected static void copyMySQLDriverToJdbcContainer(GenericContainer<?>
container)
+ throws IOException, InterruptedException {
+ Container.ExecResult mkdirResult =
+ container.execInContainer("bash", "-c", "mkdir -p " +
JDBC_PLUGIN_LIB);
+ Assertions.assertEquals(0, mkdirResult.getExitCode(),
mkdirResult.getStderr());
+
+ Path driverJarPath = mysqlDriverJarPath();
+ container.copyFileToContainer(
+ MountableFile.forHostPath(driverJarPath),
+ JDBC_PLUGIN_LIB + "/" + driverJarPath.getFileName());
+ }
+
+ private static Path mysqlDriverJarPath() {
+ try {
+ Path driverJarPath =
+ Paths.get(
+ Driver.class
+ .getProtectionDomain()
+ .getCodeSource()
+ .getLocation()
+ .toURI());
+ Assertions.assertTrue(Files.isRegularFile(driverJarPath));
+ return driverJarPath;
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to resolve MySQL JDBC driver jar from the test
classpath", e);
+ }
+ }
+
@BeforeAll
@Override
public void startUp() throws Exception {
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-edge-socket-e2e/src/test/java/org/apache/seatunnel/e2e/connector/edgesocket/EdgeSocketCheckpointRestoreIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-edge-socket-e2e/src/test/java/org/apache/seatunnel/e2e/connector/edgesocket/EdgeSocketCheckpointRestoreIT.java
index ae84947f02..8e83dc5b68 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-edge-socket-e2e/src/test/java/org/apache/seatunnel/e2e/connector/edgesocket/EdgeSocketCheckpointRestoreIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-edge-socket-e2e/src/test/java/org/apache/seatunnel/e2e/connector/edgesocket/EdgeSocketCheckpointRestoreIT.java
@@ -65,20 +65,7 @@ public class EdgeSocketCheckpointRestoreIT extends
AbstractEdgeSocketIT {
@TestContainerExtension
protected final ContainerExtendedFactory extendedFactory =
- container -> {
- Container.ExecResult extraCommands =
- container.execInContainer(
- "bash",
- "-c",
- "mkdir -p /tmp/seatunnel/plugins/Jdbc/lib &&
cd /tmp/seatunnel/plugins/Jdbc/lib && wget "
- + driverUrl()
- + " --no-check-certificate");
- Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
- };
-
- private String driverUrl() {
- return
"https://repo1.maven.org/maven2/com/mysql/mysql-connector-j/8.0.32/mysql-connector-j-8.0.32.jar";;
- }
+ AbstractEdgeSocketIT::copyMySQLDriverToJdbcContainer;
@Override
protected void startSinkDependencies() throws Exception {
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-edge-socket-e2e/src/test/java/org/apache/seatunnel/e2e/connector/edgesocket/EdgeSocketMysql8_4IT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-edge-socket-e2e/src/test/java/org/apache/seatunnel/e2e/connector/edgesocket/EdgeSocketMysql8_4IT.java
index ba39d59c1a..0eaf9a13fb 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-edge-socket-e2e/src/test/java/org/apache/seatunnel/e2e/connector/edgesocket/EdgeSocketMysql8_4IT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-edge-socket-e2e/src/test/java/org/apache/seatunnel/e2e/connector/edgesocket/EdgeSocketMysql8_4IT.java
@@ -66,20 +66,7 @@ public class EdgeSocketMysql8_4IT extends
AbstractEdgeSocketIT {
@TestContainerExtension
protected final ContainerExtendedFactory extendedFactory =
- container -> {
- Container.ExecResult extraCommands =
- container.execInContainer(
- "bash",
- "-c",
- "mkdir -p /tmp/seatunnel/plugins/Jdbc/lib &&
cd /tmp/seatunnel/plugins/Jdbc/lib && wget "
- + driverUrl()
- + " --no-check-certificate");
- Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
- };
-
- private String driverUrl() {
- return
"https://repo1.maven.org/maven2/com/mysql/mysql-connector-j/8.0.32/mysql-connector-j-8.0.32.jar";;
- }
+ AbstractEdgeSocketIT::copyMySQLDriverToJdbcContainer;
@Override
protected void startSinkDependencies() throws Exception {
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-sftp-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/fstp/SftpFileIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-sftp-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/fstp/SftpFileIT.java
index e6941d0de6..b12ba764fd 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-sftp-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/fstp/SftpFileIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-sftp-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/fstp/SftpFileIT.java
@@ -387,11 +387,16 @@ public class SftpFileIT extends TestSuiteBase implements
TestResource {
SFTP_CONTAINER_HOME
+
"/tmp/seatunnel/continuous/dst/root.bin")));
- Thread.sleep(3000);
- Assertions.assertFalse(
- isSftpFileExists(
- SFTP_CONTAINER_HOME
- +
"/tmp/seatunnel/continuous/dst/subdir/nested.bin"));
+ Awaitility.await()
+ .during(3, TimeUnit.SECONDS)
+ .atMost(5, TimeUnit.SECONDS)
+ .pollInterval(500, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertFalse(
+ isSftpFileExists(
+ SFTP_CONTAINER_HOME
+ +
"/tmp/seatunnel/continuous/dst/subdir/nested.bin")));
Container.ExecResult cancelResult = container.cancelJob(jobId);
Assertions.assertEquals(0, cancelResult.getExitCode(),
cancelResult.getStderr());
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/src/test/java/org/apache/seatunnel/e2e/connector/http/HttpIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/src/test/java/org/apache/seatunnel/e2e/connector/http/HttpIT.java
index ba2b2fe3e3..daac372164 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/src/test/java/org/apache/seatunnel/e2e/connector/http/HttpIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/src/test/java/org/apache/seatunnel/e2e/connector/http/HttpIT.java
@@ -35,6 +35,7 @@ import org.junit.jupiter.api.TestTemplate;
import org.mockserver.client.MockServerClient;
import org.mockserver.model.ClearType;
import org.mockserver.model.Format;
+import org.postgresql.Driver;
import org.testcontainers.containers.Container;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.PostgreSQLContainer;
@@ -54,6 +55,9 @@ import java.io.File;
import java.io.IOException;
import java.math.BigDecimal;
import java.net.URL;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
@@ -90,22 +94,36 @@ public class HttpIT extends TestSuiteBase implements
TestResource {
private static final String COUNT_QUERY = "select count(*) from sink";
private static final String PG_IMAGE = "postgres:14-alpine";
- private static final String PG_DRIVER_JAR =
-
"https://repo1.maven.org/maven2/org/postgresql/postgresql/42.3.3/postgresql-42.3.3.jar";;
+ private static final String PG_DRIVER_CONTAINER_PATH =
+ "/tmp/seatunnel/plugins/Jdbc/lib/postgresql.jar";
private PostgreSQLContainer<?> postgreSQLContainer;
@TestContainerExtension
private final ContainerExtendedFactory extendedFactory =
container -> {
+ Path driverJarPath = driverJarPath();
+ Assertions.assertTrue(
+ Files.isRegularFile(driverJarPath),
+ "PostgreSQL JDBC driver should be resolved from the
test classpath before E2E runs: "
+ + driverJarPath);
Container.ExecResult extraCommands =
container.execInContainer(
- "bash",
- "-c",
- "mkdir -p /tmp/seatunnel/plugins/Jdbc/lib &&
cd /tmp/seatunnel/plugins/Jdbc/lib && curl -O "
- + PG_DRIVER_JAR);
- Assertions.assertEquals(0, extraCommands.getExitCode());
+ "bash", "-c", "mkdir -p
/tmp/seatunnel/plugins/Jdbc/lib");
+ Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
+ container.copyFileToContainer(
+ MountableFile.forHostPath(driverJarPath),
PG_DRIVER_CONTAINER_PATH);
};
+ private Path driverJarPath() {
+ try {
+ return Paths.get(
+
Driver.class.getProtectionDomain().getCodeSource().getLocation().toURI());
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to resolve PostgreSQL JDBC driver jar from the
test classpath", e);
+ }
+ }
+
@BeforeAll
@Override
public void startUp() throws ClassNotFoundException {
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaFormatIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaFormatIT.java
index 6db41a61d9..1340bcb3b5 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaFormatIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaFormatIT.java
@@ -52,6 +52,7 @@ import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.TestTemplate;
+import org.postgresql.Driver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.Container;
@@ -62,12 +63,16 @@ import org.testcontainers.lifecycle.Startables;
import org.testcontainers.shaded.org.awaitility.Awaitility;
import org.testcontainers.utility.DockerImageName;
import org.testcontainers.utility.DockerLoggerFactory;
+import org.testcontainers.utility.MountableFile;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.math.BigDecimal;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
@@ -282,21 +287,33 @@ public class KafkaFormatIT extends TestSuiteBase
implements TestResource {
// --------------------------- Postgres
Container-------------------------------------
private static final String PG_IMAGE = "postgres:alpine3.16";
- private static final String PG_DRIVER_JAR =
-
"https://repo1.maven.org/maven2/org/postgresql/postgresql/42.3.3/postgresql-42.3.3.jar";;
-
private static PostgreSQLContainer<?> POSTGRESQL_CONTAINER;
+ private Path postgresqlDriverJarPath() {
+ try {
+ return Paths.get(
+
Driver.class.getProtectionDomain().getCodeSource().getLocation().toURI());
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to resolve PostgreSQL JDBC driver jar from the
test classpath", e);
+ }
+ }
+
@TestContainerExtension
private final ContainerExtendedFactory extendedFactory =
container -> {
+ Path driverJarPath = postgresqlDriverJarPath();
+ Assertions.assertTrue(
+ Files.isRegularFile(driverJarPath),
+ "PostgreSQL JDBC driver should be resolved from the
test classpath before E2E runs: "
+ + driverJarPath);
Container.ExecResult extraCommands =
container.execInContainer(
- "bash",
- "-c",
- "mkdir -p /tmp/seatunnel/plugins/Jdbc/lib &&
cd /tmp/seatunnel/plugins/Jdbc/lib && curl -O "
- + PG_DRIVER_JAR);
+ "bash", "-c", "mkdir -p
/tmp/seatunnel/plugins/Jdbc/lib");
Assertions.assertEquals(0, extraCommands.getExitCode());
+ container.copyFileToContainer(
+ MountableFile.forHostPath(driverJarPath),
+ "/tmp/seatunnel/plugins/Jdbc/lib/" +
driverJarPath.getFileName());
};
private void createKafkaContainer() {
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
index 751031dca6..cccefc0455 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java
@@ -680,6 +680,30 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
}
});
+ AtomicInteger partitionProbeId = new AtomicInteger();
+ Awaitility.await()
+ .pollInterval(2, SECONDS)
+ .atMost(2, MINUTES)
+ .untilAsserted(
+ () -> {
+ int currentProbeId =
partitionProbeId.getAndIncrement();
+ String probeMessage =
+ String.format(
+
"{\"id\":%d,\"message\":\"partition_probe_%d\",\"timestamp\":%d}",
+ 10000 + currentProbeId,
+ currentProbeId,
+ System.currentTimeMillis());
+ producer.send(
+ new ProducerRecord<>(
+ sourceTopic, 1, null,
probeMessage.getBytes()));
+ producer.flush();
+
+ Assertions.assertTrue(
+ countKafkaMessagesContaining(outputTopic,
"partition_probe_")
+ > 0,
+ "SeaTunnel should discover and consume the
new Kafka partition");
+ });
+
for (int i = 0; i < 15; i++) {
String message =
String.format(
@@ -692,17 +716,17 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
Awaitility.await()
.pollInterval(2, SECONDS)
.atMost(2, MINUTES)
- .until(
+ .untilAsserted(
() -> {
- try {
- // Check the output topic data count
- List<String> outputData =
getKafkaConsumerListData(outputTopic);
- log.info("Output topic data count: {}",
outputData.size());
- return outputData.size() >= 15 &&
outputData.size() < 25;
- } catch (Exception e) {
- log.error("Error checking output topic data",
e);
- return false;
- }
+ long newPartitionMessageCount =
+ countKafkaMessagesContaining(
+ outputTopic,
"new_partition_message_");
+ log.info(
+ "New partition output message count: {}",
+ newPartitionMessageCount);
+ Assertions.assertTrue(
+ newPartitionMessageCount >= 15,
+ "All messages from the discovered Kafka
partition should be consumed");
});
try (AdminClient adminClient = createKafkaAdmin()) {
@@ -830,6 +854,8 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
final String sourceTopic = "test_topic_restore_latest";
final String sinkTopic = "test_topic_restore_latest_output";
final String payload = "Seatunnel Restore Test Data Latest";
+ final String firstRunPayload = payload + "_additional";
+ final String restorePayload = payload + "_restore";
final String jobId = "18696753645410";
// Write 20 initial records before starting the job.
@@ -854,6 +880,27 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
awaitJobRunning(container, jobId);
+ AtomicInteger latestModeProbeId = new AtomicInteger();
+ Awaitility.await()
+ .pollInterval(2, SECONDS)
+ .atMost(2, MINUTES)
+ .untilAsserted(
+ () -> {
+ int currentProbeId =
latestModeProbeId.getAndIncrement();
+ String probePayload = payload + "_probe_" +
currentProbeId;
+ producer.send(
+ new ProducerRecord<>(
+ sourceTopic,
+ ("key_probe_" +
currentProbeId).getBytes(),
+ probePayload.getBytes()));
+ producer.flush();
+
+ Assertions.assertTrue(
+ countKafkaMessagesContaining(sinkTopic,
payload + "_probe_")
+ > 0,
+ "Latest mode source should consume records
produced after startup");
+ });
+
// Produce 10 records after job start; latest mode should consume only
these 10
// initially.
for (int i = 0; i < 10; i++) {
@@ -861,15 +908,17 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
new ProducerRecord<>(
sourceTopic,
("key_additional_" + i).getBytes(),
- (payload + "_additional").getBytes()));
+ firstRunPayload.getBytes()));
}
producer.flush();
- final long expectedSinkAfterFirstRun = 10;
Awaitility.await()
.pollInterval(2, SECONDS)
.atMost(2, MINUTES)
- .until(() -> visibleCountOnP0(sinkTopic) ==
expectedSinkAfterFirstRun);
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+ 10, countKafkaMessages(sinkTopic,
firstRunPayload)));
container.savepointJob(jobId);
@@ -879,13 +928,13 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
new ProducerRecord<>(
sourceTopic,
("key_restore_" + i).getBytes(),
- (payload + "_restore").getBytes()));
+ restorePayload.getBytes()));
}
producer.flush();
long srcEndAfterAll = endOffsetOnP0(sourceTopic);
Assertions.assertTrue(
- srcEndAfterAll == srcEndBeforeStart + 25,
+ srcEndAfterAll >= srcEndBeforeStart + 25,
"Final end offset should advance by at least 25");
CompletableFuture.runAsync(
@@ -901,7 +950,10 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
Awaitility.await()
.pollInterval(2, SECONDS)
.atMost(5, MINUTES)
- .until(() -> visibleCountOnP0(sinkTopic) ==
expectedSinkAfterFirstRun + 15);
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+ 15, countKafkaMessages(sinkTopic,
restorePayload)));
}
// ---------------------------- TIMESTAMP MODE
-----------------------------
@@ -1099,6 +1151,16 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
}
}
+ private long countKafkaMessages(String topic, String expectedValue) {
+ return getKafkaConsumerListData(topic,
0).stream().filter(expectedValue::equals).count();
+ }
+
+ private long countKafkaMessagesContaining(String topic, String
expectedValue) {
+ return getKafkaConsumerListData(topic, 0).stream()
+ .filter(value -> value.contains(expectedValue))
+ .count();
+ }
+
/** Get the current end offset (LEO) on partition-0. */
private long endOffsetOnP0(String topic) {
try (KafkaConsumer<String, String> c = new
KafkaConsumer<>(kafkaConsumerConfig())) {
@@ -1746,8 +1808,7 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
return null;
});
// wait for data written to kafka
- given().pollDelay(60, SECONDS)
- .pollInterval(5, SECONDS)
+ given().pollInterval(5, SECONDS)
.await()
.atMost(5, MINUTES)
.untilAsserted(
@@ -1780,8 +1841,7 @@ public class KafkaIT extends TestSuiteBase implements
TestResource {
}
});
- given().pollDelay(60, SECONDS)
- .pollInterval(5, SECONDS)
+ given().pollInterval(5, SECONDS)
.await()
.atMost(10, MINUTES)
.untilAsserted(
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/java/org/apache/seatunnel/e2e/connector/pulsar/CanalToPulsarIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/java/org/apache/seatunnel/e2e/connector/pulsar/CanalToPulsarIT.java
index 9e4552a7f3..0ad224c126 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/java/org/apache/seatunnel/e2e/connector/pulsar/CanalToPulsarIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/java/org/apache/seatunnel/e2e/connector/pulsar/CanalToPulsarIT.java
@@ -44,6 +44,7 @@ import org.apache.pulsar.client.api.Schema;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.TestTemplate;
+import org.postgresql.Driver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.Container;
@@ -59,6 +60,9 @@ import org.testcontainers.utility.DockerLoggerFactory;
import org.testcontainers.utility.MountableFile;
import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
@@ -116,8 +120,7 @@ public class CanalToPulsarIT extends TestSuiteBase
implements TestResource {
// postgres
private static final String PG_IMAGE = "postgres:alpine3.16";
- private static final String PG_DRIVER_JAR =
-
"https://repo1.maven.org/maven2/org/postgresql/postgresql/42.3.3/postgresql-42.3.3.jar";;
+ private static final String SEATUNNEL_JDBC_PLUGIN_LIB =
"/tmp/seatunnel/plugins/Jdbc/lib";
private static PostgreSQLContainer<?> POSTGRESQL_CONTAINER;
@@ -126,14 +129,33 @@ public class CanalToPulsarIT extends TestSuiteBase
implements TestResource {
container -> {
Container.ExecResult extraCommands =
container.execInContainer(
- "bash",
- "-c",
- "mkdir -p /tmp/seatunnel/plugins/Jdbc/lib &&
cd /tmp/seatunnel/plugins/Jdbc/lib && curl -O "
- + PG_DRIVER_JAR);
+ "bash", "-c", "mkdir -p " +
SEATUNNEL_JDBC_PLUGIN_LIB);
Assertions.assertEquals(0, extraCommands.getExitCode());
+
+ Path driverJarPath = postgresDriverJarPath();
+ container.copyFileToContainer(
+ MountableFile.forHostPath(driverJarPath),
+ SEATUNNEL_JDBC_PLUGIN_LIB + "/" +
driverJarPath.getFileName());
};
+ private Path postgresDriverJarPath() {
+ try {
+ Path driverJarPath =
+ Paths.get(
+ Driver.class
+ .getProtectionDomain()
+ .getCodeSource()
+ .getLocation()
+ .toURI());
+ Assertions.assertTrue(Files.isRegularFile(driverJarPath));
+ return driverJarPath;
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to resolve PostgreSQL JDBC driver jar from the
test classpath", e);
+ }
+ }
+
private void createPostgreSQLContainer() throws ClassNotFoundException {
POSTGRESQL_CONTAINER =
new PostgreSQLContainer<>(DockerImageName.parse(PG_IMAGE))
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rocketmq/RocketMqIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rocketmq/RocketMqIT.java
index 67616f4440..979f2faa01 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rocketmq/RocketMqIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-rocketmq-e2e/src/test/java/org/apache/seatunnel/e2e/connector/rocketmq/RocketMqIT.java
@@ -453,9 +453,10 @@ public class RocketMqIT extends TestSuiteBase implements
TestResource {
private Map<String, RocketMqConsumerMessage>
getRocketMqConsumerData(String topicName) {
Map<String, RocketMqConsumerMessage> data = new HashMap<>();
+ Map<MessageQueue, Long> consumedOffsets = new HashMap<>();
+ DefaultLitePullConsumer consumer = null;
try {
- DefaultLitePullConsumer consumer =
-
RocketMqAdminUtil.initDefaultLitePullConsumer(newConfiguration(), false);
+ consumer =
RocketMqAdminUtil.initDefaultLitePullConsumer(newConfiguration(), false);
consumer.start();
// assign
Map<MessageQueue, TopicOffset> queueOffsets =
@@ -490,35 +491,68 @@ public class RocketMqIT extends TestSuiteBase implements
TestResource {
break;
}
for (MessageExt message : messages) {
+ MessageQueue messageQueue =
+ new MessageQueue(
+ message.getTopic(),
+ message.getBrokerName(),
+ message.getQueueId());
RocketMqConsumerMessage consumerMessage =
new RocketMqConsumerMessage(
new String(message.getBody(),
StandardCharsets.UTF_8),
message.getTags());
data.put(message.getKeys(), consumerMessage);
+ consumedOffsets.merge(messageQueue,
message.getQueueOffset(), Math::max);
consumer.getOffsetStore()
.updateConsumeOffsetToBroker(
- new MessageQueue(
- message.getTopic(),
- message.getBrokerName(),
- message.getQueueId()),
- message.getQueueOffset(),
- false);
+ messageQueue, message.getQueueOffset(),
false);
}
consumer.commitSync();
}
- if (consumer != null) {
- consumer.shutdown();
- }
log.info("Consumer {} data total {}", topicName, data.size());
- // consumer.commitSync() only submits the offset to the broker,
and NameServer scans the
- // broker to update the offset every 10 seconds
- Thread.sleep(20 * 1000);
} catch (Exception ex) {
throw new RuntimeException(ex);
+ } finally {
+ if (consumer != null) {
+ consumer.shutdown();
+ }
}
+ waitConsumedOffsetsSynced(topicName, consumedOffsets);
return data;
}
+ private void waitConsumedOffsetsSynced(
+ String topicName, Map<MessageQueue, Long> consumedOffsets) {
+ if (consumedOffsets.isEmpty()) {
+ return;
+ }
+ Awaitility.await()
+ .ignoreExceptions()
+ .atMost(30, TimeUnit.SECONDS)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ Map<MessageQueue, Long> currentOffsets =
+ RocketMqAdminUtil.currentOffsets(
+ newConfiguration(),
+ Lists.newArrayList(topicName),
+ consumedOffsets.keySet());
+ for (Map.Entry<MessageQueue, Long> consumedOffset :
+ consumedOffsets.entrySet()) {
+ Long currentOffset =
currentOffsets.get(consumedOffset.getKey());
+ Assertions.assertNotNull(
+ currentOffset,
+ "Consume offset should be visible for "
+ + consumedOffset.getKey());
+ Assertions.assertTrue(
+ currentOffset >=
consumedOffset.getValue(),
+ "Consume offset should be synced to
broker, currentOffset="
+ + currentOffset
+ + ", consumedOffset="
+ + consumedOffset.getValue());
+ }
+ });
+ }
+
private void checkOffsetNoDiff(String topicName, String consumerGroup) {
RocketMqBaseConfiguration config = newConfiguration();
config.setGroupId(consumerGroup);
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/java/org/apache/seatunnel/e2e/connector/starrocks/StarRocksCDCSinkIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/java/org/apache/seatunnel/e2e/connector/starrocks/StarRocksCDCSinkIT.java
index 2824cc5ecc..443b966144 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/java/org/apache/seatunnel/e2e/connector/starrocks/StarRocksCDCSinkIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/java/org/apache/seatunnel/e2e/connector/starrocks/StarRocksCDCSinkIT.java
@@ -34,20 +34,21 @@ import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.output.Slf4jLogConsumer;
import org.testcontainers.lifecycle.Startables;
import org.testcontainers.utility.DockerLoggerFactory;
+import org.testcontainers.utility.MountableFile;
import lombok.extern.slf4j.Slf4j;
-import java.net.URL;
-import java.net.URLClassLoader;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
import java.sql.Connection;
-import java.sql.Driver;
+import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
-import java.util.Properties;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@@ -65,8 +66,8 @@ public class StarRocksCDCSinkIT extends TestSuiteBase
implements TestResource {
private static final String PASSWORD = "";
private static final String DATABASE = "test";
private static final String SINK_TABLE = "e2e_table_sink";
- private static final String SR_DRIVER_JAR =
-
"https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.16/mysql-connector-java-8.0.16.jar";;
+ private static final String SR_DRIVER_CONTAINER_PATH =
+ "/tmp/seatunnel/plugins/Jdbc/lib/mysql-connector-java.jar";
private static final String DDL_SINK =
"create table "
@@ -92,15 +93,37 @@ public class StarRocksCDCSinkIT extends TestSuiteBase
implements TestResource {
@TestContainerExtension
private final ContainerExtendedFactory extendedFactory =
container -> {
+ Path driverJarPath = driverJarPath();
+ Assertions.assertTrue(
+ Files.isRegularFile(driverJarPath),
+ "MySQL JDBC driver should be resolved from the test
classpath before E2E runs: "
+ + driverJarPath);
Container.ExecResult extraCommands =
container.execInContainer(
- "bash",
- "-c",
- "mkdir -p /tmp/seatunnel/plugins/Jdbc/lib &&
cd /tmp/seatunnel/plugins/Jdbc/lib && curl -O "
- + SR_DRIVER_JAR);
- Assertions.assertEquals(0, extraCommands.getExitCode());
+ "bash", "-c", "mkdir -p
/tmp/seatunnel/plugins/Jdbc/lib");
+ Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
+ container.copyFileToContainer(
+ MountableFile.forHostPath(driverJarPath),
SR_DRIVER_CONTAINER_PATH);
};
+ /**
+ * Resolve the MySQL JDBC test dependency from the active test classpath
instead of downloading
+ * it inside the container.
+ */
+ private Path driverJarPath() {
+ try {
+ return Paths.get(
+ com.mysql.cj.jdbc.Driver.class
+ .getProtectionDomain()
+ .getCodeSource()
+ .getLocation()
+ .toURI());
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to resolve MySQL JDBC driver jar from the test
classpath", e);
+ }
+ }
+
@BeforeAll
@Override
public void startUp() {
@@ -162,21 +185,14 @@ public class StarRocksCDCSinkIT extends TestSuiteBase
implements TestResource {
}
private void initializeJdbcConnection() throws Exception {
- URLClassLoader urlClassLoader =
- new URLClassLoader(
- new URL[] {new URL(SR_DRIVER_JAR)},
- StarRocksCDCSinkIT.class.getClassLoader());
- Thread.currentThread().setContextClassLoader(urlClassLoader);
- Driver driver = (Driver)
urlClassLoader.loadClass(DRIVER_CLASS).newInstance();
- Properties props = new Properties();
- props.put("user", USERNAME);
- props.put("password", PASSWORD);
+ Class.forName(DRIVER_CLASS);
jdbcConnection =
- driver.connect(
+ DriverManager.getConnection(
String.format(
"jdbc:mysql://%s:%s",
starRocksServer.getHost(),
starRocksServer.getFirstMappedPort()),
- props);
+ USERNAME,
+ PASSWORD);
}
private void initializeJdbcTable() {
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/java/org/apache/seatunnel/e2e/connector/starrocks/StarRocksIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/java/org/apache/seatunnel/e2e/connector/starrocks/StarRocksIT.java
index 1af0d27988..24fed1ae5c 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/java/org/apache/seatunnel/e2e/connector/starrocks/StarRocksIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/java/org/apache/seatunnel/e2e/connector/starrocks/StarRocksIT.java
@@ -39,16 +39,17 @@ import org.testcontainers.containers.Container;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.output.Slf4jLogConsumer;
import org.testcontainers.lifecycle.Startables;
+import org.testcontainers.utility.MountableFile;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.math.BigDecimal;
-import java.net.MalformedURLException;
-import java.net.URL;
-import java.net.URLClassLoader;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
import java.sql.Connection;
-import java.sql.Driver;
+import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
@@ -57,7 +58,6 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;
-import java.util.Properties;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -81,8 +81,8 @@ public class StarRocksIT extends TestSuiteBase implements
TestResource {
private static final String TABLE_OPTIONS_SINK_TABLE =
"sink_table_options";
private static final String TABLE_OPTIONS_CONFIG_FILE =
"/fake-to-starrocks-with-table-options.conf";
- private static final String SR_DRIVER_JAR =
-
"https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.16/mysql-connector-java-8.0.16.jar";;
+ private static final String SR_DRIVER_CONTAINER_PATH =
+ "/tmp/seatunnel/plugins/Jdbc/lib/mysql-connector-java.jar";
private static final String COLUMN_STRING =
"BIGINT_COL, LARGEINT_COL, SMALLINT_COL, TINYINT_COL, BOOLEAN_COL,
DECIMAL_COL, DOUBLE_COL, FLOAT_COL, INT_COL, CHAR_COL, VARCHAR_11_COL,
STRING_COL, DATETIME_COL, DATE_COL";
@@ -225,15 +225,37 @@ public class StarRocksIT extends TestSuiteBase implements
TestResource {
@TestContainerExtension
private final ContainerExtendedFactory extendedFactory =
container -> {
+ Path driverJarPath = driverJarPath();
+ Assertions.assertTrue(
+ Files.isRegularFile(driverJarPath),
+ "MySQL JDBC driver should be resolved from the test
classpath before E2E runs: "
+ + driverJarPath);
Container.ExecResult extraCommands =
container.execInContainer(
- "bash",
- "-c",
- "mkdir -p /tmp/seatunnel/plugins/Jdbc/lib &&
cd /tmp/seatunnel/plugins/Jdbc/lib && curl -O "
- + SR_DRIVER_JAR);
- Assertions.assertEquals(0, extraCommands.getExitCode());
+ "bash", "-c", "mkdir -p
/tmp/seatunnel/plugins/Jdbc/lib");
+ Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
+ container.copyFileToContainer(
+ MountableFile.forHostPath(driverJarPath),
SR_DRIVER_CONTAINER_PATH);
};
+ /**
+ * Resolve the MySQL JDBC test dependency from the active test classpath
instead of downloading
+ * it inside the container.
+ */
+ private Path driverJarPath() {
+ try {
+ return Paths.get(
+ com.mysql.cj.jdbc.Driver.class
+ .getProtectionDomain()
+ .getCodeSource()
+ .getLocation()
+ .toURI());
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to resolve MySQL JDBC driver jar from the test
classpath", e);
+ }
+ }
+
@BeforeAll
@Override
public void startUp() throws Exception {
@@ -358,18 +380,11 @@ public class StarRocksIT extends TestSuiteBase implements
TestResource {
}
}
- private void initializeJdbcConnection()
- throws SQLException, ClassNotFoundException, MalformedURLException,
- InstantiationException, IllegalAccessException {
- URLClassLoader urlClassLoader =
- new URLClassLoader(
- new URL[] {new URL(SR_DRIVER_JAR)},
StarRocksIT.class.getClassLoader());
- Thread.currentThread().setContextClassLoader(urlClassLoader);
- Driver driver = (Driver)
urlClassLoader.loadClass(DRIVER_CLASS).newInstance();
- Properties props = new Properties();
- props.put("user", USERNAME);
- props.put("password", PASSWORD);
- jdbcConnection = driver.connect(String.format(URL,
starRocksServer.getHost()), props);
+ private void initializeJdbcConnection() throws SQLException,
ClassNotFoundException {
+ Class.forName(DRIVER_CLASS);
+ jdbcConnection =
+ DriverManager.getConnection(
+ String.format(URL, starRocksServer.getHost()),
USERNAME, PASSWORD);
}
private void initializeJdbcTable() {
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/java/org/apache/seatunnel/e2e/connector/starrocks/StarRocksSchemaChangeIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/java/org/apache/seatunnel/e2e/connector/starrocks/StarRocksSchemaChangeIT.java
index 8d5d8a927f..8fa6149a8b 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/java/org/apache/seatunnel/e2e/connector/starrocks/StarRocksSchemaChangeIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/java/org/apache/seatunnel/e2e/connector/starrocks/StarRocksSchemaChangeIT.java
@@ -40,14 +40,15 @@ import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.output.Slf4jLogConsumer;
import org.testcontainers.lifecycle.Startables;
import org.testcontainers.utility.DockerLoggerFactory;
+import org.testcontainers.utility.MountableFile;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
-import java.net.URL;
-import java.net.URLClassLoader;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
import java.sql.Connection;
-import java.sql.Driver;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
@@ -58,7 +59,6 @@ import java.time.OffsetDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
-import java.util.Properties;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
@@ -90,8 +90,8 @@ public class StarRocksSchemaChangeIT extends TestSuiteBase
implements TestResour
private static final String PASSWORD = "";
private static final String SINK_TABLE = "products";
private static final String CREATE_DATABASE = "CREATE DATABASE IF NOT
EXISTS " + DATABASE;
- private static final String SR_DRIVER_JAR =
-
"https://repo1.maven.org/maven2/com/mysql/mysql-connector-j/8.0.32/mysql-connector-j-8.0.32.jar";;
+ private static final String SR_DRIVER_CONTAINER_PATH =
+ "/tmp/seatunnel/plugins/Jdbc/lib/mysql-connector-java.jar";
private Connection starRocksConnection;
private Connection mysqlConnection;
@@ -114,15 +114,37 @@ public class StarRocksSchemaChangeIT extends
TestSuiteBase implements TestResour
@TestContainerExtension
private final ContainerExtendedFactory extendedFactory =
container -> {
+ Path driverJarPath = driverJarPath();
+ Assertions.assertTrue(
+ Files.isRegularFile(driverJarPath),
+ "MySQL JDBC driver should be resolved from the test
classpath before E2E runs: "
+ + driverJarPath);
Container.ExecResult extraCommands =
container.execInContainer(
- "bash",
- "-c",
- "mkdir -p /tmp/seatunnel/plugins/Jdbc/lib &&
cd /tmp/seatunnel/plugins/Jdbc/lib && curl -O "
- + SR_DRIVER_JAR);
- Assertions.assertEquals(0, extraCommands.getExitCode());
+ "bash", "-c", "mkdir -p
/tmp/seatunnel/plugins/Jdbc/lib");
+ Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
+ container.copyFileToContainer(
+ MountableFile.forHostPath(driverJarPath),
SR_DRIVER_CONTAINER_PATH);
};
+ /**
+ * Resolve the MySQL JDBC test dependency from the active test classpath
instead of downloading
+ * it inside the container.
+ */
+ private Path driverJarPath() {
+ try {
+ return Paths.get(
+ com.mysql.cj.jdbc.Driver.class
+ .getProtectionDomain()
+ .getCodeSource()
+ .getLocation()
+ .toURI());
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to resolve MySQL JDBC driver jar from the test
classpath", e);
+ }
+ }
+
private static MySqlContainer createMySqlContainer(MySqlVersion version) {
return new MySqlContainer(version)
.withConfigurationOverride("docker/server-gtids/my.cnf")
@@ -137,19 +159,12 @@ public class StarRocksSchemaChangeIT extends
TestSuiteBase implements TestResour
}
private void initializeJdbcConnection() throws Exception {
- URLClassLoader urlClassLoader =
- new URLClassLoader(
- new URL[] {new URL(SR_DRIVER_JAR)},
- StarRocksCDCSinkIT.class.getClassLoader());
- Thread.currentThread().setContextClassLoader(urlClassLoader);
- Driver driver = (Driver)
urlClassLoader.loadClass(DRIVER_CLASS).newInstance();
- Properties props = new Properties();
- props.put("user", USERNAME);
- props.put("password", PASSWORD);
+ Class.forName(DRIVER_CLASS);
starRocksConnection =
- driver.connect(
+ DriverManager.getConnection(
String.format("jdbc:mysql://%s:%s",
starRocksServer.getHost(), QUERY_PORT),
- props);
+ USERNAME,
+ PASSWORD);
}
private void initializeStarRocksServer() {
