You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by lz...@apache.org on 2022/05/09 08:46:39 UTC
[flink-table-store] branch master updated: [FLINK-27542] Refactor E2eTestBase with docker compose
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-table-store.git
The following commit(s) were added to refs/heads/master by this push:
new ad9e09d [FLINK-27542] Refactor E2eTestBase with docker compose
ad9e09d is described below
commit ad9e09dbd9621ff2935f193c53a35b355c76fac6
Author: tsreaper <ts...@gmail.com>
AuthorDate: Mon May 9 16:46:35 2022 +0800
[FLINK-27542] Refactor E2eTestBase with docker compose
This closes #114
---
.../flink/table/store/tests/E2eTestBase.java | 122 ++++++++-------------
.../flink/table/store/tests/LogStoreE2eTest.java | 52 +--------
.../test/resources-filtered/docker-compose.yaml | 97 ++++++++++++++++
.../project.properties => resources/flink.env} | 2 +-
pom.xml | 2 +-
5 files changed, 148 insertions(+), 127 deletions(-)
diff --git a/flink-table-store-e2e-tests/src/test/java/org/apache/flink/table/store/tests/E2eTestBase.java b/flink-table-store-e2e-tests/src/test/java/org/apache/flink/table/store/tests/E2eTestBase.java
index ab63f48..ece6ee6 100644
--- a/flink-table-store-e2e-tests/src/test/java/org/apache/flink/table/store/tests/E2eTestBase.java
+++ b/flink-table-store-e2e-tests/src/test/java/org/apache/flink/table/store/tests/E2eTestBase.java
@@ -20,27 +20,23 @@ package org.apache.flink.table.store.tests;
import org.apache.flink.table.store.tests.utils.TestUtils;
-import com.github.dockerjava.api.model.Volume;
-import org.junit.ClassRule;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.testcontainers.containers.BindMode;
import org.testcontainers.containers.Container;
-import org.testcontainers.containers.GenericContainer;
-import org.testcontainers.containers.Network;
+import org.testcontainers.containers.ContainerState;
+import org.testcontainers.containers.DockerComposeContainer;
import org.testcontainers.containers.output.OutputFrame;
import org.testcontainers.containers.output.Slf4jLogConsumer;
import org.testcontainers.utility.MountableFile;
-import java.io.IOException;
+import java.io.File;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Properties;
import java.util.UUID;
import java.util.function.Function;
@@ -55,95 +51,69 @@ public abstract class E2eTestBase {
private static final Logger LOG = LoggerFactory.getLogger(E2eTestBase.class);
- // ------------------------------------------------------------------------------------------
- // Flink Variables
- // ------------------------------------------------------------------------------------------
- private static final String FLINK_IMAGE_TAG;
-
- static {
- Properties properties = new Properties();
- try {
- properties.load(
- E2eTestBase.class.getClassLoader().getResourceAsStream("project.properties"));
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- // TODO change image tag to official flink image after 1.15 is released
- FLINK_IMAGE_TAG = "tsreaper/flink-test:" + properties.getProperty("flink.version");
+ private final boolean withKafka;
+
+ protected E2eTestBase() {
+ this(false);
+ }
+
+ protected E2eTestBase(boolean withKafka) {
+ this.withKafka = withKafka;
}
- private static final String INTER_CONTAINER_JM_ALIAS = "jobmanager";
- private static final String INTER_CONTAINER_TM_ALIAS = "taskmanager";
- private static final int JOB_MANAGER_REST_PORT = 8081;
- private static final String FLINK_PROPERTIES =
- String.join(
- "\n",
- Arrays.asList(
- "jobmanager.rpc.address: jobmanager",
- "taskmanager.numberOfTaskSlots: 9",
- "parallelism.default: 3",
- "sql-client.execution.result-mode: TABLEAU"));
-
- // ------------------------------------------------------------------------------------------
- // Additional Jars
- // ------------------------------------------------------------------------------------------
private static final String TABLE_STORE_JAR_NAME = "flink-table-store.jar";
private static final String BUNDLED_HADOOP_JAR_NAME = "bundled-hadoop.jar";
-
- protected static final String TEST_DATA_DIR = "/opt/flink/test-data";
-
- @ClassRule public static final Network NETWORK = Network.newNetwork();
- private GenericContainer<?> jobManager;
- protected GenericContainer<?> taskManager;
+ protected static final String TEST_DATA_DIR = "/test-data";
private static final String PRINT_SINK_IDENTIFIER = "table-store-e2e-result";
private static final int CHECK_RESULT_INTERVAL_MS = 1000;
private static final int CHECK_RESULT_RETRIES = 60;
private final List<String> currentResults = new ArrayList<>();
+ private DockerComposeContainer<?> environment;
+ private ContainerState jobManager;
+
@BeforeEach
public void before() throws Exception {
- Volume volume = new Volume(TEST_DATA_DIR);
- jobManager =
- new GenericContainer<>(FLINK_IMAGE_TAG)
- .withCommand("jobmanager")
- .withNetwork(NETWORK)
- .withNetworkAliases(INTER_CONTAINER_JM_ALIAS)
- .withExposedPorts(JOB_MANAGER_REST_PORT)
- .withEnv("FLINK_PROPERTIES", FLINK_PROPERTIES)
- .withLogConsumer(new LogConsumer(LOG))
- .withCreateContainerCmdModifier(
- cmd ->
- cmd.withName("jobmanager-" + UUID.randomUUID().toString())
- .withVolumes(volume));
- jobManager.start();
+ List<String> services = new ArrayList<>();
+ services.add("jobmanager");
+ services.add("taskmanager");
+ environment =
+ new DockerComposeContainer<>(
+ new File(
+ E2eTestBase.class
+ .getClassLoader()
+ .getResource("docker-compose.yaml")
+ .toURI()))
+ .withLogConsumer("jobmanager_1", new LogConsumer(LOG))
+ .withLogConsumer("taskmanager_1", new LogConsumer(LOG));
+ if (withKafka) {
+ services.add("zookeeper");
+ services.add("kafka");
+ environment
+ .withLogConsumer("zookeeper_1", new Slf4jLogConsumer(LOG))
+ .withLogConsumer("kafka_1", new Slf4jLogConsumer(LOG));
+ }
+ environment.withServices(services.toArray(new String[0]));
+
+ synchronized (E2eTestBase.class) {
+ // there are some steps which cannot be executed in parallel when starting the same
+ // docker image, so we should lock here
+ environment.start();
+ }
+ jobManager = environment.getContainerByServiceName("jobmanager_1").get();
jobManager.execInContainer("chown", "-R", "flink:flink", TEST_DATA_DIR);
- LOG.info("Job manager started.");
-
- taskManager =
- new GenericContainer<>(FLINK_IMAGE_TAG)
- .withCommand("taskmanager")
- .withNetwork(NETWORK)
- .withNetworkAliases(INTER_CONTAINER_TM_ALIAS)
- .withEnv("FLINK_PROPERTIES", FLINK_PROPERTIES)
- .withVolumesFrom(jobManager, BindMode.READ_WRITE)
- .dependsOn(jobManager)
- .withLogConsumer(new LogConsumer(LOG));
- taskManager.start();
- LOG.info("Task manager started.");
copyResource(TABLE_STORE_JAR_NAME);
copyResource(BUNDLED_HADOOP_JAR_NAME);
}
@AfterEach
- public void after() throws Exception {
- taskManager.stop();
- jobManager.stop();
- LOG.info("Job manager and task manager stopped.");
+ public void after() {
+ environment.stop();
}
- private void copyResource(String resourceName) throws Exception {
+ private void copyResource(String resourceName) {
jobManager.copyFileToContainer(
MountableFile.forHostPath(TestUtils.getResource(resourceName).toString()),
TEST_DATA_DIR + "/" + resourceName);
@@ -211,7 +181,7 @@ public abstract class E2eTestBase {
return String.format(testDataSinkDdl, sinkName, schema, PRINT_SINK_IDENTIFIER);
}
- protected List<String> getCurrentResults() {
+ private List<String> getCurrentResults() {
synchronized (currentResults) {
return new ArrayList<>(currentResults);
}
diff --git a/flink-table-store-e2e-tests/src/test/java/org/apache/flink/table/store/tests/LogStoreE2eTest.java b/flink-table-store-e2e-tests/src/test/java/org/apache/flink/table/store/tests/LogStoreE2eTest.java
index ef1e5af..e601b28 100644
--- a/flink-table-store-e2e-tests/src/test/java/org/apache/flink/table/store/tests/LogStoreE2eTest.java
+++ b/flink-table-store-e2e-tests/src/test/java/org/apache/flink/table/store/tests/LogStoreE2eTest.java
@@ -18,55 +18,15 @@
package org.apache.flink.table.store.tests;
-import org.apache.flink.util.DockerImageVersions;
-
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.testcontainers.containers.KafkaContainer;
-import org.testcontainers.containers.output.Slf4jLogConsumer;
-import org.testcontainers.utility.DockerImageName;
-import java.time.Duration;
import java.util.UUID;
/** Tests for reading and writing log store in stream jobs. */
public class LogStoreE2eTest extends E2eTestBase {
- private static final Logger LOG = LoggerFactory.getLogger(LogStoreE2eTest.class);
-
- private static final String INTER_CONTAINER_KAFKA_ALIAS = "kafka";
-
- private KafkaContainer kafka;
-
- @BeforeEach
- @Override
- public void before() throws Exception {
- super.before();
- kafka =
- new KafkaContainer(DockerImageName.parse(DockerImageVersions.KAFKA))
- .withEmbeddedZookeeper()
- .withNetwork(NETWORK)
- .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS)
- .withEnv(
- "KAFKA_TRANSACTION_MAX_TIMEOUT_MS",
- String.valueOf(Duration.ofHours(2).toMillis()))
- // Disable log deletion to prevent records from being deleted during test
- // run
- .withEnv("KAFKA_LOG_RETENTION_MS", "-1")
- .withLogConsumer(new Slf4jLogConsumer(LOG));
- kafka.start();
- LOG.info("Kafka started.");
- }
-
- @AfterEach
- @Override
- public void after() throws Exception {
- super.after();
- kafka.stop();
- LOG.info("Kafka stopped.");
+ public LogStoreE2eTest() {
+ super(true);
}
@Test
@@ -118,9 +78,7 @@ public class LogStoreE2eTest extends E2eTestBase {
+ ");";
tableStoreStreamDdl =
String.format(
- tableStoreStreamDdl,
- TEST_DATA_DIR + "/" + tableStoreDir,
- getBootstrapServers());
+ tableStoreStreamDdl, TEST_DATA_DIR + "/" + tableStoreDir, "kafka:9092");
// prepare first part of test data
writeTestData(testDataSourceDir + "/1.csv", "A,10\nC,30\nD,40\n");
@@ -149,10 +107,6 @@ public class LogStoreE2eTest extends E2eTestBase {
checkResult(s -> s.split(",")[0], "A, 100", "B, 2", "C, 30", "D, 400");
}
- private String getBootstrapServers() {
- return INTER_CONTAINER_KAFKA_ALIAS + ":9092";
- }
-
private void runSql(String sql, String... ddls) throws Exception {
runSql(String.join("\n", ddls) + "\n" + sql);
}
diff --git a/flink-table-store-e2e-tests/src/test/resources-filtered/docker-compose.yaml b/flink-table-store-e2e-tests/src/test/resources-filtered/docker-compose.yaml
new file mode 100644
index 0000000..7918c70
--- /dev/null
+++ b/flink-table-store-e2e-tests/src/test/resources-filtered/docker-compose.yaml
@@ -0,0 +1,97 @@
+################################################################################
+# 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.
+################################################################################
+
+version: "3"
+
+services:
+
+ # ----------------------------------------
+ # Flink services
+ # ----------------------------------------
+
+ jobmanager:
+ image: apache/flink:${flink.version}
+ volumes:
+ - testdata:/test-data
+ command: jobmanager
+ env_file:
+ - ./flink.env
+ networks:
+ testnetwork:
+ aliases:
+ - jobmanager
+ ports:
+ - "8081:8081"
+
+ taskmanager:
+ image: apache/flink:${flink.version}
+ volumes:
+ - testdata:/test-data
+ command: taskmanager
+ env_file:
+ - ./flink.env
+ networks:
+ testnetwork:
+ aliases:
+ - taskmanager
+ depends_on:
+ - jobmanager
+
+ # ----------------------------------------
+ # Kafka services
+ # ----------------------------------------
+
+ zookeeper:
+ image: confluentinc/cp-zookeeper:7.0.1
+ networks:
+ testnetwork:
+ aliases:
+ - zookeeper
+ environment:
+ ZOOKEEPER_CLIENT_PORT: 2181
+ ZOOKEEPER_TICK_TIME: 2000
+ ports:
+ - "2181:2181"
+
+ kafka:
+ image: confluentinc/cp-kafka:7.0.1
+ networks:
+ testnetwork:
+ aliases:
+ - kafka
+ environment:
+ KAFKA_BROKER_ID: 1
+ KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
+ KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_INTERNAL:PLAINTEXT
+ KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_INTERNAL://kafka:29092
+ KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+ KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+ KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+ KAFKA_TRANSACTION_MAX_TIMEOUT_MS: 7200000
+ # Disable log deletion to prevent records from being deleted during test run
+ KAFKA_LOG_RETENTION_MS: -1
+ ports:
+ - "9092:9092"
+ depends_on:
+ - zookeeper
+
+volumes:
+ testdata:
+
+networks:
+ testnetwork:
diff --git a/flink-table-store-e2e-tests/src/test/resources-filtered/project.properties b/flink-table-store-e2e-tests/src/test/resources/flink.env
similarity index 86%
rename from flink-table-store-e2e-tests/src/test/resources-filtered/project.properties
rename to flink-table-store-e2e-tests/src/test/resources/flink.env
index 439929d..d604200 100644
--- a/flink-table-store-e2e-tests/src/test/resources-filtered/project.properties
+++ b/flink-table-store-e2e-tests/src/test/resources/flink.env
@@ -16,4 +16,4 @@
# limitations under the License.
################################################################################
-flink.version=${flink.version}
+FLINK_PROPERTIES="jobmanager.rpc.address: jobmanager\ntaskmanager.numberOfTaskSlots: 9\nparallelism.default: 3\nsql-client.execution.result-mode: TABLEAU"
diff --git a/pom.xml b/pom.xml
index a2205f0..411c3b7 100644
--- a/pom.xml
+++ b/pom.xml
@@ -64,7 +64,7 @@ under the License.
</modules>
<properties>
- <flink.version>1.15-SNAPSHOT</flink.version>
+ <flink.version>1.15.0</flink.version>
<hadoop.version>2.8.5</hadoop.version>
<scala.binary.version>2.12</scala.binary.version>
<slf4j.version>1.7.15</slf4j.version>