Merge pull request #20 from ClickHouse/embedded-for-tests

Author: mzitnik

.github/workflows/tests.yaml

@@ -0,0 +1,20 @@
+name: Apache Flink ClickHouse Connector Tests CI
+
+on: [push]
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    name: Apache Flink ClickHouse Connector tests
+    steps:
+      - uses: actions/checkout@v3
+      - name: Set up JDK 17
+        uses: actions/setup-java@v3
+        with:
+          java-version: '21'
+          distribution: 'adopt'
+          architecture: x64
+      - name: Setup and execute Gradle 'test' task
+        uses: gradle/gradle-build-action@v2
+        with:
+          arguments: test

flink-connector-clickhouse-base/build.gradle.kts

@@ -3,6 +3,11 @@
  * 
  */
 
+plugins {
+    kotlin("jvm") version "1.9.0"
+    java
+}
+
 repositories {
     // Use Maven Central for resolving dependencies.
     mavenCentral()

flink-connector-clickhouse-base/src/test/java/org/apache/flink/connector/test/DummyFlinkClusterTest.java

@@ -0,0 +1,49 @@
+package org.apache.flink.connector.test;
+
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.streaming.api.functions.sink.legacy.SinkFunction;
+import org.apache.flink.streaming.util.TestStreamEnvironment;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.junit.jupiter.api.*;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+class DummyFlinkClusterTest extends EmbeddedFlinkClusterForTests {
+    // A simple Collection Sink
+    private static class CollectSink implements SinkFunction<Long> {
+
+        // must be static
+        public static final List<Long> values = Collections.synchronizedList(new ArrayList<>());
+
+        @Override
+        public void invoke(Long value, SinkFunction.Context context) throws Exception {
+            values.add(value);
+        }
+    }
+
+    public static class IncrementMapFunction implements MapFunction<Long, Long> {
+
+        @Override
+        public Long map(Long record) throws Exception {
+            return record + 1;
+        }
+    }
+
+    @Test
+    void testDummyFlinkCluster() throws Exception {
+        MiniClusterWithClientResource flinkCluster = EmbeddedFlinkClusterForTests.getMiniCluster();
+        TestStreamEnvironment env = flinkCluster.getTestStreamEnvironment();
+
+        CollectSink.values.clear();
+
+        env.fromData(1L, 2L)
+            .map(new IncrementMapFunction())
+            .addSink(new CollectSink());
+        env.execute("testDummyFlinkCluster");
+        Assertions.assertEquals(2, CollectSink.values.size());
+        Assertions.assertEquals(2L, CollectSink.values.get(0));
+        Assertions.assertEquals(3L, CollectSink.values.get(1));
+    }
+}

flink-connector-clickhouse-base/src/test/java/org/apache/flink/connector/test/EmbeddedFlinkClusterForTests.java

@@ -0,0 +1,49 @@
+package org.apache.flink.connector.test;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.RestOptions;
+import org.apache.flink.configuration.TaskManagerOptions;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+
+public class EmbeddedFlinkClusterForTests {
+
+    static MiniClusterWithClientResource flinkCluster = null;
+    static int REST_PORT = getFromEnvOrDefault("REST_PORT", 8081);
+    static int NUM_TASK_SLOTS = getFromEnvOrDefault("NUM_TASK_SLOTS", 2);
+    static int NUM_TASK_SLOTS_PER_TASK = getFromEnvOrDefault("NUM_TASK_SLOTS_PER_TASK" , 2);
+    static int NUM_TASK_MANAGERS = getFromEnvOrDefault("NUM_TASK_MANAGERS",3);
+
+    static int getFromEnvOrDefault(String key, int defaultValue) {
+        String value = System.getenv().getOrDefault(key, String.valueOf(defaultValue));
+        return Integer.parseInt(value);
+    }
+
+    @BeforeAll
+    static void setUp() throws Exception {
+        Configuration config = new Configuration();
+        config.set(RestOptions.PORT, REST_PORT); // web UI port (optional)
+        config.set(TaskManagerOptions.NUM_TASK_SLOTS, NUM_TASK_SLOTS);
+        flinkCluster = new MiniClusterWithClientResource(
+                new MiniClusterResourceConfiguration.Builder()
+                        .setNumberSlotsPerTaskManager(NUM_TASK_SLOTS_PER_TASK)
+                        .setNumberTaskManagers(NUM_TASK_MANAGERS)
+                        .setConfiguration(config)
+                        .build());
+        flinkCluster.before();
+    }
+    @AfterAll
+    static void tearDown() {
+        if (flinkCluster != null) {
+            flinkCluster.after();
+        }
+    }
+
+    protected static MiniClusterWithClientResource getMiniCluster() {
+        if (flinkCluster == null)
+            throw new RuntimeException("No MiniCluster available");
+        return flinkCluster;
+    }
+}