[hotfix][e2e][jdbc] fix some error when docker version is old

seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-jdbc-flink-e2e/src/test/java/org/apache/seatunnel/e2e/flink/v2/jdbc/JdbcMysqlIT.java

@@ -39,6 +39,8 @@
 import org.testcontainers.utility.DockerImageName;
 
 import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
 import java.nio.file.Paths;
 import java.sql.Connection;
 import java.sql.DriverManager;
@@ -50,6 +52,7 @@
 import java.util.Arrays;
 import java.util.Comparator;
 import java.util.List;
+import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -62,9 +65,9 @@ public class JdbcMysqlIT extends FlinkContainer {
 
     @SuppressWarnings("checkstyle:MagicNumber")
     @BeforeEach
-    public void startPostgreSqlContainer() throws Exception {
+    public void startMySqlContainer() throws Exception {
         // Non-root users need to grant XA_RECOVER_ADMIN permission on is_exactly_once = "true"
-        mc = new MySQLContainer<>(DockerImageName.parse("mysql:8.0.29"))
+        mc = new MySQLContainer<>(DockerImageName.parse("bitnami/mysql:8.0.29").asCompatibleSubstituteFor("mysql"))
             .withNetwork(NETWORK)
             .withNetworkAliases("mysql")
             .withUsername("root")
@@ -77,17 +80,14 @@ public void startPostgreSqlContainer() throws Exception {
             .atLeast(100, TimeUnit.MILLISECONDS)
             .pollInterval(500, TimeUnit.MILLISECONDS)
             .atMost(5, TimeUnit.SECONDS)
-            .untilAsserted(() -> initializeJdbcTable());
+            .untilAsserted(this::initializeJdbcTable);
         batchInsertData();
     }
 
-    private void initializeJdbcTable() {
-        java.net.URL resource = FlinkContainer.class.getResource("/jdbc/init_sql/mysql_init.conf");
-        if (resource == null) {
-            throw new IllegalArgumentException("can't find find file");
-        }
+    private void initializeJdbcTable() throws URISyntaxException {
+        URI resource = Objects.requireNonNull(FlinkContainer.class.getResource("/jdbc/init_sql/mysql_init.conf")).toURI();
 
-        config = new ConfigBuilder(Paths.get(resource.getPath())).getConfig();
+        config = new ConfigBuilder(Paths.get(resource)).getConfig();
 
         CheckConfigUtil.checkAllExists(this.config, "source_table", "sink_table", "type_source_table",
             "type_sink_table", "insert_type_source_table_sql", "check_type_sink_table_sql");
@@ -125,15 +125,15 @@ private void batchInsertData() {
     @Test
     public void testJdbcMysqlSourceAndSink() throws Exception {
         Container.ExecResult execResult = executeSeaTunnelFlinkJob("/jdbc/jdbc_mysql_source_and_sink.conf");
-        Assertions.assertEquals(0, execResult.getExitCode());
+        Assertions.assertEquals(0, execResult.getExitCode(), execResult.getStderr());
 
         Assertions.assertIterableEquals(generateTestDataset(), queryResult());
     }
 
     @Test
     public void testJdbcMysqlSourceAndSinkParallel() throws Exception {
         Container.ExecResult execResult = executeSeaTunnelFlinkJob("/jdbc/jdbc_mysql_source_and_sink_parallel.conf");
-        Assertions.assertEquals(0, execResult.getExitCode());
+        Assertions.assertEquals(0, execResult.getExitCode(), execResult.getStderr());
 
         //Sorting is required, because it is read in parallel, so there will be out of order
         List<List> sortedResult = queryResult().stream().sorted(Comparator.comparing(list -> (Integer) list.get(1)))
@@ -145,7 +145,7 @@ public void testJdbcMysqlSourceAndSinkParallel() throws Exception {
     public void testJdbcMysqlSourceAndSinkParallelUpperLower() throws Exception {
         Container.ExecResult execResult =
             executeSeaTunnelFlinkJob("/jdbc/jdbc_mysql_source_and_sink_parallel_upper_lower.conf");
-        Assertions.assertEquals(0, execResult.getExitCode());
+        Assertions.assertEquals(0, execResult.getExitCode(), execResult.getStderr());
 
         //Sorting is required, because it is read in parallel, so there will be out of order
         List<List> sortedResult = queryResult().stream().sorted(Comparator.comparing(list -> (Integer) list.get(1)))
@@ -167,7 +167,7 @@ public void testJdbcMysqlSourceAndSinkXA() throws Exception {
     @Test
     public void testJdbcMysqlSourceAndSinkDataType() throws Exception {
         Container.ExecResult execResult = executeSeaTunnelFlinkJob("/jdbc/jdbc_mysql_source_and_sink_datatype.conf");
-        Assertions.assertEquals(0, execResult.getExitCode());
+        Assertions.assertEquals(0, execResult.getExitCode(), execResult.getStderr());
         checkSinkDataTypeTable();
     }
 
@@ -213,7 +213,7 @@ private static List<List> generateTestDataset() {
     }
 
     @AfterEach
-    public void closePostgreSqlContainer() {
+    public void closeMySqlContainer() {
         if (mc != null) {
             mc.stop();
         }

seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-jdbc-flink-e2e/src/test/java/org/apache/seatunnel/e2e/flink/v2/jdbc/JdbcPostgresIT.java

@@ -21,6 +21,7 @@
 
 import org.apache.seatunnel.e2e.flink.FlinkContainer;
 
+import lombok.extern.slf4j.Slf4j;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
@@ -49,6 +50,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+@Slf4j
 public class JdbcPostgresIT extends FlinkContainer {
     private static final Logger LOGGER = LoggerFactory.getLogger(JdbcPostgresIT.class);
     private PostgreSQLContainer<?> pg;
@@ -57,10 +59,11 @@ public class JdbcPostgresIT extends FlinkContainer {
     @SuppressWarnings("checkstyle:MagicNumber")
     @BeforeEach
     public void startPostgreSqlContainer() throws Exception {
-        pg = new PostgreSQLContainer<>(DockerImageName.parse("postgres:14.3"))
+        pg = new PostgreSQLContainer<>(DockerImageName.parse("postgres:14-alpine"))
             .withNetwork(NETWORK)
             .withNetworkAliases("postgresql")
             .withCommand("postgres -c max_prepared_transactions=100")
+            .withUsername("root")
             .withLogConsumer(new Slf4jLogConsumer(LOGGER));
         Startables.deepStart(Stream.of(pg)).join();
         LOGGER.info("Postgres container started");
@@ -115,14 +118,14 @@ private void batchInsertData() {
     @Test
     public void testJdbcPostgresSourceAndSink() throws Exception {
         Container.ExecResult execResult = executeSeaTunnelFlinkJob("/jdbc/jdbc_postgres_source_and_sink.conf");
-        Assertions.assertEquals(0, execResult.getExitCode());
+        Assertions.assertEquals(0, execResult.getExitCode(), execResult.getStderr());
         Assertions.assertIterableEquals(generateTestDataset(), queryResult());
     }
 
     @Test
     public void testJdbcPostgresSourceAndSinkParallel() throws Exception {
         Container.ExecResult execResult = executeSeaTunnelFlinkJob("/jdbc/jdbc_postgres_source_and_sink_parallel.conf");
-        Assertions.assertEquals(0, execResult.getExitCode());
+        Assertions.assertEquals(0, execResult.getExitCode(), execResult.getStderr());
 
         //Sorting is required, because it is read in parallel, so there will be out of order
         List<List> sortedResult = queryResult().stream().sorted(Comparator.comparing(list -> (Integer) list.get(1)))
@@ -134,7 +137,7 @@ public void testJdbcPostgresSourceAndSinkParallel() throws Exception {
     public void testJdbcPostgresSourceAndSinkParallelUpperLower() throws Exception {
         Container.ExecResult execResult =
             executeSeaTunnelFlinkJob("/jdbc/jdbc_postgres_source_and_sink_parallel_upper_lower.conf");
-        Assertions.assertEquals(0, execResult.getExitCode());
+        Assertions.assertEquals(0, execResult.getExitCode(), execResult.getStderr());
 
         //Sorting is required, because it is read in parallel, so there will be out of order
         List<List> sortedResult = queryResult().stream().sorted(Comparator.comparing(list -> (Integer) list.get(1)))
@@ -148,7 +151,7 @@ public void testJdbcPostgresSourceAndSinkParallelUpperLower() throws Exception {
     @Test
     public void testJdbcPostgresSourceAndSinkXA() throws Exception {
         Container.ExecResult execResult = executeSeaTunnelFlinkJob("/jdbc/jdbc_postgres_source_and_sink_xa.conf");
-        Assertions.assertEquals(0, execResult.getExitCode());
+        Assertions.assertEquals(0, execResult.getExitCode(), execResult.getStderr());
 
         Assertions.assertIterableEquals(generateTestDataset(), queryResult());
     }

seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-jdbc-flink-e2e/src/test/resources/jdbc/jdbc_postgres_source_and_sink.conf

@@ -24,7 +24,7 @@ source{
     jdbc{
         url = "jdbc:postgresql://postgresql:5432/test"
         driver = "org.postgresql.Driver"
-        user = "test"
+        user = "root"
         password = "test"
         query = "select name , age from source"
     }
@@ -37,8 +37,8 @@ sink {
     jdbc {
         url = "jdbc:postgresql://postgresql:5432/test"
         driver = "org.postgresql.Driver"
-        user = "test"
+        user = "root"
         password = "test"
         query = "insert into sink(name,age) values(?,?)"
     }
-}
+}

seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-jdbc-flink-e2e/src/test/resources/jdbc/jdbc_postgres_source_and_sink_parallel.conf

@@ -24,7 +24,7 @@ source{
     jdbc{
         url = "jdbc:postgresql://postgresql:5432/test"
         driver = "org.postgresql.Driver"
-        user = "test"
+        user = "root"
         password = "test"
         query = "select user_id,  name , age from source"
         partition_column= "user_id"
@@ -44,9 +44,9 @@ sink {
 
         url = "jdbc:postgresql://postgresql:5432/test"
         driver = "org.postgresql.Driver"
-        user = "test"
+        user = "root"
         password = "test"
         connection_check_timeout_sec = 100
         query = "insert into sink(name,age) values(?,?)"
     }
-}
+}

seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-jdbc-flink-e2e/src/test/resources/jdbc/jdbc_postgres_source_and_sink_parallel_upper_lower.conf

@@ -24,7 +24,7 @@ source{
     jdbc{
         url = "jdbc:postgresql://postgresql:5432/test"
         driver = "org.postgresql.Driver"
-        user = "test"
+        user = "root"
         password = "test"
         query = "select user_id,  name , age from source"
         partition_column= "user_id"
@@ -46,9 +46,9 @@ sink {
         url = "jdbc:postgresql://postgresql:5432/test"
         driver = "org.postgresql.Driver"
 
-        user = "test"
+        user = "root"
         password = "test"
         connection_check_timeout_sec = 100
         query = "insert into sink(name,age) values(?,?)"
     }
-}
+}

seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-jdbc-flink-e2e/src/test/resources/jdbc/jdbc_postgres_source_and_sink_xa.conf

@@ -25,7 +25,7 @@ source {
     jdbc{
         url = "jdbc:postgresql://postgresql:5432/test"
         driver = "org.postgresql.Driver"
-        user = "test"
+        user = "root"
         password = "test"
         query = "select name , age from source"
     }
@@ -38,7 +38,7 @@ sink {
     jdbc {
         url = "jdbc:postgresql://postgresql:5432/test"
         driver = "org.postgresql.Driver"
-        user = "test"
+        user = "root"
         password = "test"
 
         max_retries = 0
@@ -50,4 +50,4 @@ sink {
         max_commit_attempts = 3
         transaction_timeout_sec = 86400
     }
-}
+}

seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-jdbc-spark-e2e/src/test/java/org/apache/seatunnel/e2e/spark/v2/jdbc/JdbcMysqlIT.java

@@ -39,7 +39,8 @@
 import org.testcontainers.utility.DockerImageName;
 
 import java.io.IOException;
-import java.net.URL;
+import java.net.URI;
+import java.net.URISyntaxException;
 import java.nio.file.Paths;
 import java.sql.Connection;
 import java.sql.DriverManager;
@@ -51,6 +52,7 @@
 import java.util.Arrays;
 import java.util.Comparator;
 import java.util.List;
+import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -63,9 +65,9 @@ public class JdbcMysqlIT extends SparkContainer {
 
     @SuppressWarnings("checkstyle:MagicNumber")
     @BeforeEach
-    public void startPostgreSqlContainer() throws Exception {
+    public void startMySqlContainer() throws Exception {
         // Non-root users need to grant XA_RECOVER_ADMIN permission on is_exactly_once = "true"
-        mc = new MySQLContainer<>(DockerImageName.parse("mysql:8.0.29"))
+        mc = new MySQLContainer<>(DockerImageName.parse("bitnami/mysql:8.0.29").asCompatibleSubstituteFor("mysql"))
             .withNetwork(NETWORK)
             .withNetworkAliases("mysql")
             .withUsername("root")
@@ -78,18 +80,14 @@ public void startPostgreSqlContainer() throws Exception {
             .atLeast(100, TimeUnit.MILLISECONDS)
             .pollInterval(500, TimeUnit.MILLISECONDS)
             .atMost(5, TimeUnit.SECONDS)
-            .untilAsserted(() -> initializeJdbcTable());
+            .untilAsserted(this::initializeJdbcTable);
         batchInsertData();
     }
 
-    private void initializeJdbcTable() {
-        URL resource = JdbcMysqlIT.class.getResource("/jdbc/init_sql/mysql_init.conf");
-        if (resource == null) {
-            throw new IllegalArgumentException("can't find find file");
-        }
-
-        config = new ConfigBuilder(Paths.get(resource.getPath())).getConfig();
+    private void initializeJdbcTable() throws URISyntaxException {
 
+        URI resource = Objects.requireNonNull(JdbcMysqlIT.class.getResource("/jdbc/init_sql/mysql_init.conf")).toURI();
+        config = new ConfigBuilder(Paths.get(resource)).getConfig();
         CheckConfigUtil.checkAllExists(this.config, "source_table", "sink_table", "type_source_table",
             "type_sink_table", "insert_type_source_table_sql", "check_type_sink_table_sql");
 
@@ -213,7 +211,7 @@ private static List<List> generateTestDataset() {
     }
 
     @AfterEach
-    public void closePostgreSqlContainer() {
+    public void closeMySqlContainer() {
         if (mc != null) {
             mc.stop();
         }

seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-jdbc-spark-e2e/src/test/java/org/apache/seatunnel/e2e/spark/v2/jdbc/JdbcPostgresIT.java

@@ -57,7 +57,7 @@ public class JdbcPostgresIT extends SparkContainer {
     @SuppressWarnings("checkstyle:MagicNumber")
     @BeforeEach
     public void startPostgreSqlContainer() throws Exception {
-        pg = new PostgreSQLContainer<>(DockerImageName.parse("postgres:14.3"))
+        pg = new PostgreSQLContainer<>(DockerImageName.parse("postgres:14-alpine"))
             .withNetwork(NETWORK)
             .withNetworkAliases("postgresql")
             .withCommand("postgres -c max_prepared_transactions=100")

seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-jdbc-spark-e2e/src/test/resources/jdbc/jdbc_postgres_source_and_sink.conf

@@ -41,4 +41,4 @@ sink {
         password = "test"
         query = "insert into sink(name,age) values(?,?)"
     }
-}
+}

seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-jdbc-spark-e2e/src/test/resources/jdbc/jdbc_postgres_source_and_sink_parallel.conf

@@ -49,4 +49,4 @@ sink {
         connection_check_timeout_sec = 100
         query = "insert into sink(name,age) values(?,?)"
     }
-}
+}

seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-jdbc-spark-e2e/src/test/resources/jdbc/jdbc_postgres_source_and_sink_parallel_upper_lower.conf

@@ -51,4 +51,4 @@ sink {
         connection_check_timeout_sec = 100
         query = "insert into sink(name,age) values(?,?)"
     }
-}
+}

seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-jdbc-spark-e2e/src/test/resources/jdbc/jdbc_postgres_source_and_sink_xa.conf

@@ -50,4 +50,4 @@ sink {
         max_commit_attempts = 3
         transaction_timeout_sec = 86400
     }
-}
+}