[fix]: CI build failures #173

Author: itinycheng

.github/workflows/ci.yml

@@ -39,3 +39,13 @@ jobs:
         uses: docker/setup-buildx-action@v3.10.0
       - name: Build with Maven
         run: ./mvnw verify
+      - name: Upload test logs on failure
+        if: failure()
+        uses: actions/upload-artifact@v4
+        with:
+          name: test-logs-java-${{ matrix.java-version }}
+          path: |
+            **/target/surefire-reports/
+            **/target/failsafe-reports/
+            **/hs_err_*.log
+          retention-days: 7

flink-connector-clickhouse-e2e-test/pom.xml

@@ -36,14 +36,21 @@ limitations under the License.
 	<dependencies>
 		<dependency>
 			<groupId>org.testcontainers</groupId>
-			<artifactId>clickhouse</artifactId>
+			<artifactId>testcontainers-clickhouse</artifactId>
 			<version>${testcontainer.version}</version>
+			<scope>test</scope>
 		</dependency>
 		<dependency>
 			<groupId>org.apache.flink</groupId>
 			<artifactId>flink-connector-test-utils</artifactId>
 			<version>${flink.version}</version>
 			<scope>test</scope>
+			<exclusions>
+				<exclusion>
+					<artifactId>commons-io</artifactId>
+					<groupId>commons-io</groupId>
+				</exclusion>
+			</exclusions>
 		</dependency>
 		<dependency>
 			<groupId>org.apache.flink</groupId>

flink-connector-clickhouse-e2e-test/src/test/java/org/apache/flink/connector/clickhouse/ClickhouseE2ECase.java

@@ -27,6 +27,8 @@ public void testSink() throws Exception {
                         CLICKHOUSE_CONTAINER.getJdbcUrl(),
                         CLICKHOUSE_CONTAINER.getUsername(),
                         CLICKHOUSE_CONTAINER.getPassword());
+        Thread.sleep(5000);
+
         proxy.execute(
                 "create table test (id Int32, name String, float32_column Float32, date_column Date,datetime_column DateTime, array_column Array(Int32)) engine = Memory");
         proxy.execute(
@@ -52,7 +54,9 @@ public void testSink() throws Exception {
                         + "',\n"
                         + "  'table-name' = 'test',\n"
                         + "  'username'='test_username',\n"
-                        + "  'password'='test_password'\n"
+                        + "  'password'='test_password',\n"
+                        + "  'properties.compress' = 'false',\n"
+                        + "  'properties.decompress' = 'false'\n"
                         + ");");
         sqlLines.add(
                 "create table test (id int, name varchar,float32_column FLOAT,\n"
@@ -63,7 +67,9 @@ public void testSink() throws Exception {
                         + "',\n"
                         + "  'table-name' = 'test_insert',\n"
                         + "  'username'='test_username',\n"
-                        + "  'password'='test_password'\n"
+                        + "  'password'='test_password',\n"
+                        + "  'properties.compress' = 'false',\n"
+                        + "  'properties.decompress' = 'false'\n"
                         + ");");
         sqlLines.add("insert into test select * from clickhouse_test;");
 
@@ -86,7 +92,7 @@ public void testSink() throws Exception {
                 expectedResult,
                 "test_insert",
                 Arrays.asList("id", "name", "float32_column", "datetime_column", "array_column"),
-                60000);
+                600_000);
     }
 
     @After

flink-connector-clickhouse-e2e-test/src/test/java/org/apache/flink/connector/clickhouse/ClickhouseProxy.java

@@ -20,6 +20,7 @@
 
 /** A proxy for Clickhouse to execute SQLs and check results. */
 public class ClickhouseProxy {
+
     private final String jdbcUrl;
     private final String username;
     private final String password;
@@ -41,13 +42,15 @@ public void connect() {
                 Properties properties = new Properties();
                 properties.put("username", username);
                 properties.put("password", password);
+                properties.put("compress", "false");
+                properties.put("decompress", "false");
                 ClickHouseDataSource clickHouseDataSource =
                         new ClickHouseDataSource(jdbcUrl, properties);
                 connection = clickHouseDataSource.getConnection(username, password);
                 statement = connection.createStatement();
             }
         } catch (Exception e) {
-            logger.error("Failed to connect to clickhouse", e);
+            throw new RuntimeException("Cannot establish ClickHouse connection", e);
         }
     }
 

flink-connector-clickhouse-e2e-test/src/test/java/org/apache/flink/connector/clickhouse/FlinkContainerTestEnvironment.java

@@ -6,7 +6,6 @@
 import org.apache.flink.client.program.rest.RestClusterClient;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.RestOptions;
-import org.apache.flink.connector.testframe.container.TestcontainersSettings;
 import org.apache.flink.runtime.client.JobStatusMessage;
 import org.apache.flink.table.api.ValidationException;
 import org.apache.flink.test.resources.ResourceTestUtils;
@@ -22,6 +21,7 @@
 import org.testcontainers.containers.GenericContainer;
 import org.testcontainers.containers.Network;
 import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.Wait;
 import org.testcontainers.lifecycle.Startables;
 import org.testcontainers.utility.DockerImageName;
 import org.testcontainers.utility.MountableFile;
@@ -54,15 +54,15 @@ public class FlinkContainerTestEnvironment {
                     .withExposedPorts(8123, 9000)
                     .withUsername("test_username")
                     .withPassword("test_password")
+                    .withStartupTimeout(Duration.ofMinutes(10))
+                    .withStartupAttempts(3)
+                    .waitingFor(
+                            Wait.forHttp("/ping")
+                                    .forPort(8123)
+                                    .forStatusCode(200)
+                                    .withStartupTimeout(Duration.ofMinutes(10)))
                     .withLogConsumer(new Slf4jLogConsumer(logger));
 
-    private static final TestcontainersSettings TESTCONTAINERS_SETTINGS =
-            TestcontainersSettings.builder()
-                    .logger(logger)
-                    .network(NETWORK)
-                    .dependsOn(CLICKHOUSE_CONTAINER)
-                    .build();
-
     public static final Path SQL_CONNECTOR_CLICKHOUSE_JAR =
             ResourceTestUtils.getResource("flink-connector-clickhouse-1.0.0-SNAPSHOT.jar");
     public static final Path CLICKHOUSE_JDBC_JAR =
@@ -81,6 +81,7 @@ public class FlinkContainerTestEnvironment {
     @Before
     public void setUp() throws Exception {
         CLICKHOUSE_CONTAINER.start();
+        Thread.sleep(30000);
 
         String properties =
                 String.join(
@@ -90,7 +91,7 @@ public void setUp() throws Exception {
                                 "heartbeat.timeout: 60000",
                                 "parallelism.default: 1"));
         jobManager =
-                new GenericContainer<>(new DockerImageName("flink:2.1-scala_2.12"))
+                new GenericContainer<>(DockerImageName.parse("flink:2.1-scala_2.12"))
                         .withCommand("jobmanager")
                         .withNetwork(NETWORK)
                         .withExtraHost("host.docker.internal", "host-gateway")
@@ -101,7 +102,7 @@ public void setUp() throws Exception {
                         .withEnv("FLINK_PROPERTIES", properties)
                         .withLogConsumer(new Slf4jLogConsumer(logger));
         taskManager =
-                new GenericContainer<>(new DockerImageName("flink:2.1-scala_2.12"))
+                new GenericContainer<>(DockerImageName.parse("flink:2.1-scala_2.12"))
                         .withCommand("taskmanager")
                         .withExtraHost("host.docker.internal", "host-gateway")
                         .withNetwork(NETWORK)
@@ -184,11 +185,6 @@ private String copyAndGetContainerPath(GenericContainer<?> container, String fil
         return containerPath;
     }
 
-    private static List<String> readSqlFile(final String resourceName) throws Exception {
-        return Files.readAllLines(
-                Paths.get(ClickhouseE2ECase.class.getResource("/" + resourceName).toURI()));
-    }
-
     public void waitUntilJobRunning(Duration timeout) {
         RestClusterClient<?> clusterClient = getRestClusterClient();
         Deadline deadline = Deadline.fromNow(timeout);

pom.xml

@@ -61,8 +61,8 @@ limitations under the License.
 		<clickhouse-jdbc.version>0.6.4</clickhouse-jdbc.version>
 		<scala.binary.version>2.12</scala.binary.version>
 		<flink.version>2.1.0</flink.version>
-		<commons-lang3.version>3.13.0</commons-lang3.version>
-		<testcontainer.version>1.19.8</testcontainer.version>
+		<commons-lang3.version>3.18.0</commons-lang3.version>
+		<testcontainer.version>2.0.2</testcontainer.version>
 		<httpclient5.version>5.2.1</httpclient5.version>
 		<httpcore5.version>5.2</httpcore5.version>
 		<shade.base>org.apache.flink.shaded.clickhouse</shade.base>