This is an automated email from the ASF dual-hosted git repository.
renqs pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 45ad242beb7 [FLINK-27854][tests] Add support for FlinkContainers to
work with both existing Docker images and flink-dist builds
45ad242beb7 is described below
commit 45ad242beb7a6d87df239a9069728c1594c905c4
Author: Alexander Fedulov <[email protected]>
AuthorDate: Tue May 10 13:50:00 2022 +0200
[FLINK-27854][tests] Add support for FlinkContainers to work with both
existing Docker images and flink-dist builds
- Refactors FlinkContainers initialization prpocess by introducing
centralized configuration holders (FlinkContainersSettings and
TestContainersSettings).
- Previously FlinkContainers workflow was based on always recreating images
from local 'flink-dist'. This contribution adds support for external Docker
images (required for externalized connectors repositories, where 'flink-dist'
is not present).
---
.../table/test/KinesisFirehoseTableITTest.java | 16 +-
.../table/test/KinesisStreamsTableApiIT.java | 16 +-
.../tests/ElasticsearchSinkE2ECaseBase.java | 2 +-
.../flink/tests/util/kafka/KafkaSinkE2ECase.java | 2 +-
.../flink/tests/util/kafka/KafkaSourceE2ECase.java | 2 +-
.../util/kafka/SQLClientSchemaRegistryITCase.java | 6 +-
.../flink/tests/util/kafka/SmokeKafkaITCase.java | 11 +-
.../FlinkContainerTestEnvironment.java | 80 ++-
.../util/flink/container/FlinkContainers.java | 102 +++-
.../flink/container/FlinkContainersBuilder.java | 298 ----------
.../flink/container/FlinkContainersSettings.java | 599 +++++++++++++++++++++
.../util/flink/container/FlinkImageBuilder.java | 103 ++--
.../container/FlinkTestcontainersConfigurator.java | 183 +++++++
.../flink/container/TestcontainersSettings.java | 169 ++++++
.../FlinkContainerWithPulsarEnvironment.java | 2 +-
.../kinesis/test/KinesisTableApiITCase.java | 16 +-
16 files changed, 1194 insertions(+), 413 deletions(-)
diff --git
a/flink-end-to-end-tests/flink-end-to-end-tests-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/table/test/KinesisFirehoseTableITTest.java
b/flink-end-to-end-tests/flink-end-to-end-tests-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/table/test/KinesisFirehoseTableITTest.java
index 0876953cf68..7334f1403a9 100644
---
a/flink-end-to-end-tests/flink-end-to-end-tests-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/table/test/KinesisFirehoseTableITTest.java
+++
b/flink-end-to-end-tests/flink-end-to-end-tests-aws-kinesis-firehose/src/test/java/org/apache/flink/connector/firehose/table/test/KinesisFirehoseTableITTest.java
@@ -23,6 +23,7 @@ import
org.apache.flink.connector.aws.testutils.LocalstackContainer;
import org.apache.flink.tests.util.TestUtils;
import org.apache.flink.tests.util.flink.SQLJobSubmission;
import org.apache.flink.tests.util.flink.container.FlinkContainers;
+import org.apache.flink.tests.util.flink.container.TestcontainersSettings;
import org.apache.flink.util.DockerImageVersions;
import org.apache.flink.util.TestLogger;
@@ -96,17 +97,20 @@ public class KinesisFirehoseTableITTest extends TestLogger {
.withNetwork(network)
.withNetworkAliases("localstack");
- public static final FlinkContainers FLINK =
- FlinkContainers.builder()
- .setEnvironmentVariable("AWS_CBOR_DISABLE", "1")
- .setEnvironmentVariable(
+ public static final TestcontainersSettings TESTCONTAINERS_SETTINGS =
+ TestcontainersSettings.builder()
+ .environmentVariable("AWS_CBOR_DISABLE", "1")
+ .environmentVariable(
"FLINK_ENV_JAVA_OPTS",
"-Dorg.apache.flink.kinesis-firehose.shaded.com.amazonaws.sdk.disableCertChecking
-Daws.cborEnabled=false")
- .setNetwork(network)
- .setLogger(LOG)
+ .network(network)
+ .logger(LOG)
.dependsOn(mockFirehoseContainer)
.build();
+ public static final FlinkContainers FLINK =
+
FlinkContainers.builder().withTestcontainersSettings(TESTCONTAINERS_SETTINGS).build();
+
@Before
public void setup() throws Exception {
System.setProperty(SdkSystemSetting.CBOR_ENABLED.property(), "false");
diff --git
a/flink-end-to-end-tests/flink-end-to-end-tests-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/table/test/KinesisStreamsTableApiIT.java
b/flink-end-to-end-tests/flink-end-to-end-tests-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/table/test/KinesisStreamsTableApiIT.java
index 9b90dd9128c..8f8886f244a 100644
---
a/flink-end-to-end-tests/flink-end-to-end-tests-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/table/test/KinesisStreamsTableApiIT.java
+++
b/flink-end-to-end-tests/flink-end-to-end-tests-aws-kinesis-streams/src/test/java/org/apache/flink/connector/kinesis/table/test/KinesisStreamsTableApiIT.java
@@ -25,6 +25,7 @@ import
org.apache.flink.connectors.kinesis.testutils.KinesaliteContainer;
import org.apache.flink.tests.util.TestUtils;
import org.apache.flink.tests.util.flink.SQLJobSubmission;
import org.apache.flink.tests.util.flink.container.FlinkContainers;
+import org.apache.flink.tests.util.flink.container.TestcontainersSettings;
import org.apache.flink.util.DockerImageVersions;
import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList;
@@ -91,17 +92,20 @@ public class KinesisStreamsTableApiIT {
.withNetwork(network)
.withNetworkAliases(INTER_CONTAINER_KINESALITE_ALIAS);
- public static final FlinkContainers FLINK =
- FlinkContainers.builder()
- .setEnvironmentVariable("AWS_CBOR_DISABLE", "1")
- .setEnvironmentVariable(
+ public static final TestcontainersSettings TESTCONTAINERS_SETTINGS =
+ TestcontainersSettings.builder()
+ .environmentVariable("AWS_CBOR_DISABLE", "1")
+ .environmentVariable(
"FLINK_ENV_JAVA_OPTS",
"-Dorg.apache.flink.kinesis-streams.shaded.com.amazonaws.sdk.disableCertChecking
-Daws.cborEnabled=false")
- .setNetwork(network)
- .setLogger(LOGGER)
+ .network(network)
+ .logger(LOGGER)
.dependsOn(KINESALITE)
.build();
+ public static final FlinkContainers FLINK =
+
FlinkContainers.builder().withTestcontainersSettings(TESTCONTAINERS_SETTINGS).build();
+
@BeforeClass
public static void setupFlink() throws Exception {
FLINK.start();
diff --git
a/flink-end-to-end-tests/flink-end-to-end-tests-common-elasticsearch/src/main/java/org/apache/flink/streaming/tests/ElasticsearchSinkE2ECaseBase.java
b/flink-end-to-end-tests/flink-end-to-end-tests-common-elasticsearch/src/main/java/org/apache/flink/streaming/tests/ElasticsearchSinkE2ECaseBase.java
index a897644e9db..1020cb450e6 100644
---
a/flink-end-to-end-tests/flink-end-to-end-tests-common-elasticsearch/src/main/java/org/apache/flink/streaming/tests/ElasticsearchSinkE2ECaseBase.java
+++
b/flink-end-to-end-tests/flink-end-to-end-tests-common-elasticsearch/src/main/java/org/apache/flink/streaming/tests/ElasticsearchSinkE2ECaseBase.java
@@ -25,7 +25,7 @@ import
org.apache.flink.connector.testframe.junit.annotations.TestExternalSystem
import org.apache.flink.connector.testframe.junit.annotations.TestSemantics;
import org.apache.flink.connector.testframe.testsuites.SinkTestSuiteBase;
import org.apache.flink.streaming.api.CheckpointingMode;
-import org.apache.flink.tests.util.flink.FlinkContainerTestEnvironment;
+import
org.apache.flink.tests.util.flink.container.FlinkContainerTestEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git
a/flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/KafkaSinkE2ECase.java
b/flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/KafkaSinkE2ECase.java
index 0c75a00adba..334a50bbb68 100644
---
a/flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/KafkaSinkE2ECase.java
+++
b/flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/KafkaSinkE2ECase.java
@@ -27,7 +27,7 @@ import
org.apache.flink.connector.testframe.junit.annotations.TestSemantics;
import org.apache.flink.connector.testframe.testsuites.SinkTestSuiteBase;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.tests.util.TestUtils;
-import org.apache.flink.tests.util.flink.FlinkContainerTestEnvironment;
+import
org.apache.flink.tests.util.flink.container.FlinkContainerTestEnvironment;
import org.apache.flink.util.DockerImageVersions;
import org.testcontainers.containers.KafkaContainer;
diff --git
a/flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/KafkaSourceE2ECase.java
b/flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/KafkaSourceE2ECase.java
index eae1eda5b4e..a71c3283517 100644
---
a/flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/KafkaSourceE2ECase.java
+++
b/flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/KafkaSourceE2ECase.java
@@ -27,7 +27,7 @@ import
org.apache.flink.connector.testframe.junit.annotations.TestSemantics;
import org.apache.flink.connector.testframe.testsuites.SourceTestSuiteBase;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.tests.util.TestUtils;
-import org.apache.flink.tests.util.flink.FlinkContainerTestEnvironment;
+import
org.apache.flink.tests.util.flink.container.FlinkContainerTestEnvironment;
import org.apache.flink.util.DockerImageVersions;
import org.testcontainers.containers.KafkaContainer;
diff --git
a/flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/SQLClientSchemaRegistryITCase.java
b/flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/SQLClientSchemaRegistryITCase.java
index 437d3982a18..b62f52ff3d7 100644
---
a/flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/SQLClientSchemaRegistryITCase.java
+++
b/flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/SQLClientSchemaRegistryITCase.java
@@ -22,6 +22,7 @@ import org.apache.flink.api.common.time.Deadline;
import org.apache.flink.tests.util.TestUtils;
import org.apache.flink.tests.util.flink.SQLJobSubmission;
import org.apache.flink.tests.util.flink.container.FlinkContainers;
+import org.apache.flink.tests.util.flink.container.TestcontainersSettings;
import org.apache.flink.tests.util.kafka.containers.SchemaRegistryContainer;
import org.apache.flink.util.DockerImageVersions;
@@ -88,8 +89,11 @@ public class SQLClientSchemaRegistryITCase {
.withNetworkAliases(INTER_CONTAINER_REGISTRY_ALIAS)
.dependsOn(KAFKA);
+ public final TestcontainersSettings testcontainersSettings =
+
TestcontainersSettings.builder().network(NETWORK).logger(LOG).dependsOn(KAFKA).build();
+
public final FlinkContainers flink =
-
FlinkContainers.builder().setNetwork(NETWORK).setLogger(LOG).dependsOn(KAFKA).build();
+
FlinkContainers.builder().withTestcontainersSettings(testcontainersSettings).build();
private KafkaContainerClient kafkaClient;
private CachedSchemaRegistryClient registryClient;
diff --git
a/flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/SmokeKafkaITCase.java
b/flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/SmokeKafkaITCase.java
index c1a706fc33a..d926fdd5e55 100644
---
a/flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/SmokeKafkaITCase.java
+++
b/flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/SmokeKafkaITCase.java
@@ -25,6 +25,8 @@ import
org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.tests.util.TestUtils;
import org.apache.flink.tests.util.flink.JobSubmission;
import org.apache.flink.tests.util.flink.container.FlinkContainers;
+import org.apache.flink.tests.util.flink.container.FlinkContainersSettings;
+import org.apache.flink.tests.util.flink.container.TestcontainersSettings;
import org.apache.flink.testutils.junit.FailsOnJava11;
import org.apache.flink.util.TestLoggerExtension;
@@ -85,12 +87,15 @@ public class SmokeKafkaITCase {
.withNetwork(NETWORK)
.withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS);
+ public static final TestcontainersSettings TESTCONTAINERS_SETTINGS =
+
TestcontainersSettings.builder().logger(LOG).dependsOn(KAFKA_CONTAINER).build();
+
@RegisterExtension
public static final FlinkContainers FLINK =
FlinkContainers.builder()
- .setConfiguration(getConfiguration())
- .setLogger(LOG)
- .dependsOn(KAFKA_CONTAINER)
+ .withFlinkContainersSettings(
+
FlinkContainersSettings.basedOn(getConfiguration()))
+ .withTestcontainersSettings(TESTCONTAINERS_SETTINGS)
.build();
private static AdminClient admin;
diff --git
a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/FlinkContainerTestEnvironment.java
b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/container/FlinkContainerTestEnvironment.java
similarity index 71%
rename from
flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/FlinkContainerTestEnvironment.java
rename to
flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/container/FlinkContainerTestEnvironment.java
index 9b78dec6176..3409a538c7e 100644
---
a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/FlinkContainerTestEnvironment.java
+++
b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/container/FlinkContainerTestEnvironment.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.flink.tests.util.flink;
+package org.apache.flink.tests.util.flink.container;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.testframe.environment.ClusterControllable;
@@ -27,8 +27,6 @@ import
org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
import org.apache.flink.runtime.testutils.CommonTestUtils;
import org.apache.flink.streaming.api.environment.RemoteStreamEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.tests.util.flink.container.FlinkContainers;
-import org.apache.flink.tests.util.flink.container.FlinkContainersBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -36,30 +34,21 @@ import org.slf4j.LoggerFactory;
import java.net.URL;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.List;
+import java.util.Collection;
import java.util.stream.Collectors;
-import static
org.apache.flink.configuration.HeartbeatManagerOptions.HEARTBEAT_INTERVAL;
-import static
org.apache.flink.configuration.HeartbeatManagerOptions.HEARTBEAT_TIMEOUT;
-import static
org.apache.flink.configuration.JobManagerOptions.SLOT_REQUEST_TIMEOUT;
-import static
org.apache.flink.configuration.MetricOptions.METRIC_FETCHER_UPDATE_INTERVAL;
-import static org.apache.flink.configuration.TaskManagerOptions.NUM_TASK_SLOTS;
-import static
org.apache.flink.connector.testframe.utils.ConnectorTestConstants.HEARTBEAT_INTERVAL_MS;
-import static
org.apache.flink.connector.testframe.utils.ConnectorTestConstants.HEARTBEAT_TIMEOUT_MS;
-import static
org.apache.flink.connector.testframe.utils.ConnectorTestConstants.METRIC_FETCHER_UPDATE_INTERVAL_MS;
-import static
org.apache.flink.connector.testframe.utils.ConnectorTestConstants.SLOT_REQUEST_TIMEOUT_MS;
-
/** Test environment running job on {@link FlinkContainers}. */
public class FlinkContainerTestEnvironment implements TestEnvironment,
ClusterControllable {
private static final Logger LOG =
LoggerFactory.getLogger(FlinkContainerTestEnvironment.class);
private final FlinkContainers flinkContainers;
- private final List<String> jarPaths = new ArrayList<>();
+ private final Collection<String> jarPaths = new ArrayList<>();
+ private String checkpointPath =
FlinkContainersSettings.getDefaultCheckpointPath();
public FlinkContainerTestEnvironment(
int numTaskManagers, int numSlotsPerTaskManager, String...
jarPaths) {
- this(new Configuration(), numTaskManagers, numSlotsPerTaskManager,
jarPaths);
+ this(new Configuration(), numTaskManagers, numSlotsPerTaskManager,
Arrays.asList(jarPaths));
}
public FlinkContainerTestEnvironment(
@@ -67,21 +56,54 @@ public class FlinkContainerTestEnvironment implements
TestEnvironment, ClusterCo
int numTaskManagers,
int numSlotsPerTaskManager,
String... jarPaths) {
- Configuration config = new Configuration();
- config.addAll(clusterConfiguration);
- config.set(NUM_TASK_SLOTS, numSlotsPerTaskManager);
- config.set(HEARTBEAT_INTERVAL, HEARTBEAT_INTERVAL_MS);
- config.set(HEARTBEAT_TIMEOUT, HEARTBEAT_TIMEOUT_MS);
- config.set(SLOT_REQUEST_TIMEOUT, SLOT_REQUEST_TIMEOUT_MS);
- config.set(METRIC_FETCHER_UPDATE_INTERVAL,
METRIC_FETCHER_UPDATE_INTERVAL_MS);
+ this(
+ clusterConfiguration,
+ numTaskManagers,
+ numSlotsPerTaskManager,
+ Arrays.asList(jarPaths));
+ }
+
+ public FlinkContainerTestEnvironment(
+ Configuration clusterConfiguration,
+ int numTaskManagers,
+ int numSlotsPerTaskManager,
+ Collection<String> jarPaths) {
+
+ final TestcontainersSettings testcontainersSettings =
+ TestcontainersSettings.builder().logger(LOG).build();
+
+ final FlinkContainersSettings flinkContainersSettings =
+ FlinkContainersSettings.builder()
+ .basedOn(clusterConfiguration)
+ .numTaskManagers(numTaskManagers)
+ .numSlotsPerTaskManager(numSlotsPerTaskManager)
+ .enableZookeeperHA()
+ .jarPaths(jarPaths)
+ .build();
+
flinkContainers =
FlinkContainers.builder()
- .setNumTaskManagers(numTaskManagers)
- .setConfiguration(config)
- .setLogger(LOG)
- .enableZookeeperHA()
+ .withTestcontainersSettings(testcontainersSettings)
+ .withFlinkContainersSettings(flinkContainersSettings)
+ .build();
+ this.jarPaths.addAll(jarPaths);
+ }
+
+ public static FlinkContainerTestEnvironment
fromSettings(FlinkContainersSettings settings) {
+ return new FlinkContainerTestEnvironment(settings);
+ }
+
+ public FlinkContainerTestEnvironment(FlinkContainersSettings settings) {
+ TestcontainersSettings testcontainersSettings =
+ TestcontainersSettings.builder().logger(LOG).build();
+ flinkContainers =
+ FlinkContainers.builder()
+ .withFlinkContainersSettings(settings)
+ .withTestcontainersSettings(testcontainersSettings)
.build();
- this.jarPaths.addAll(Arrays.asList(jarPaths));
+
+ checkpointPath = settings.getCheckpointPath();
+ jarPaths.addAll(settings.getJarPaths());
}
@Override
@@ -128,7 +150,7 @@ public class FlinkContainerTestEnvironment implements
TestEnvironment, ClusterCo
@Override
public String getCheckpointUri() {
- return FlinkContainersBuilder.CHECKPOINT_PATH.toString();
+ return checkpointPath;
}
@Override
diff --git
a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/container/FlinkContainers.java
b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/container/FlinkContainers.java
index 47cd1d9b187..352add83337 100644
---
a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/container/FlinkContainers.java
+++
b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/container/FlinkContainers.java
@@ -79,32 +79,54 @@ import static
org.apache.flink.util.Preconditions.checkState;
* <pre>{@code
* public class E2ETest {
* // Create a Flink cluster using default configurations.
- * // Remember to declare it as "static" as required by JUnit 5.
+ * // Remember to declare it as "static" as required by
+ * // JUnit 5.
* @RegisterExtension
- * static FlinkContainers flink = FlinkContainers.builder().build();
+ * static FlinkContainers flink =
+ * FlinkContainers.builder().build();
+ *
+ * ---
*
* // To work together with other containers
+ * static TestcontainersSettings testcontainersSettings =
+ * TestcontainersSettings.builder()
+ * .dependsOn(kafkaContainer)
+ * .build());
+ *
* @RegisterExtension
* static FlinkContainers flink =
- * FlinkContainers.builder()
- * .dependsOn(kafkaContainer)
- * .build();
+ * FlinkContainers.builder()
+ * .withTestcontainersSettings(testcontainersSettings)
+ * .build();
*
+ * ---
* // Customize a Flink cluster
- * // Remember to declare it as "static" as required by JUnit 5.
+ * static FlinkContainersSettings flinkContainersSettings =
+ * FlinkContainersSettings.builder()
+ * .numTaskManagers(3)
+ * .numSlotsPerTaskManager(6)
+ * .baseImage(
+ *
"ghcr.io/apache/flink-docker:1.16-snapshot-scala_2.12-java11-debian")
+ * .enableZookeeperHA()
+ * .build();
+ *
+ * static TestcontainersSettings testcontainersSettings =
+ * TestcontainersSettings.builder()
+ * .setLogger(
+ * LoggerFactory.getLogger(E2ETest.class))
+ * .build());
+ *
* @RegisterExtension
* static FlinkContainers flink =
* FlinkContainers.builder()
- * .setNumTaskManagers(3)
- * .setConfiguration(TaskManagerOptions.NUM_TASK_SLOTS, 6)
- * .setLogger(LoggerFactory.getLogger(E2ETest.class))
- * .enableZookeeperHA()
- * .build();
+ * .withFlinkContainersSettings(flinkContainersSettings)
+ * .withTestcontainersSettings(testcontainersSettings)
+ * .build();
* ...
* }
* }</pre>
*
- * <p>Detailed usages can be found in the JavaDoc of {@link
FlinkContainersBuilder}.
+ * <p>See {@link FlinkContainersSettings} and {@link TestcontainersSettings}
for available options.
*
* <h2>Prerequisites</h2>
*
@@ -128,9 +150,57 @@ public class FlinkContainers implements BeforeAllCallback,
AfterAllCallback {
@Nullable private RestClusterClient<StandaloneClusterId> restClusterClient;
private boolean isStarted = false;
+ /** The {@link FlinkContainers} builder. */
+ public static final class Builder {
+ private FlinkContainersSettings flinkContainersSettings =
+ FlinkContainersSettings.defaultConfig();
+ private TestcontainersSettings testcontainersSettings =
+ TestcontainersSettings.defaultSettings();
+
+ private Builder() {}
+
+ /**
+ * Allows to optionally provide Flink containers settings. {@link
FlinkContainersSettings}
+ * based on defaults will be used otherwise.
+ *
+ * @param flinkContainersSettings The Flink containers settings.
+ * @return A reference to this Builder.
+ */
+ public Builder withFlinkContainersSettings(
+ FlinkContainersSettings flinkContainersSettings) {
+ this.flinkContainersSettings = flinkContainersSettings;
+ return this;
+ }
+
+ /**
+ * Allows to optionally provide Testcontainers settings. {@link
TestcontainersSettings}
+ * based on defaults will be used otherwise.
+ *
+ * @param testcontainersSettings The Testcontainers settings.
+ * @return A reference to this Builder.
+ */
+ public Builder withTestcontainersSettings(TestcontainersSettings
testcontainersSettings) {
+ this.testcontainersSettings = testcontainersSettings;
+ return this;
+ }
+
+ /**
+ * Returns {@code FlinkContainers} built from the provided settings.
+ *
+ * @return {@code FlinkContainers} built with parameters of this {@code
+ * FlinkContainers.Builder}.
+ */
+ public FlinkContainers build() {
+ FlinkTestcontainersConfigurator configurator =
+ new FlinkTestcontainersConfigurator(
+ flinkContainersSettings, testcontainersSettings);
+ return configurator.configure();
+ }
+ }
+
/** Creates a builder for {@link FlinkContainers}. */
- public static FlinkContainersBuilder builder() {
- return new FlinkContainersBuilder();
+ public static Builder builder() {
+ return new Builder();
}
FlinkContainers(
@@ -243,7 +313,7 @@ public class FlinkContainers implements BeforeAllCallback,
AfterAllCallback {
// Construct SQL client command
commands.add("cat /tmp/script.sql | ");
- commands.add("flink/bin/sql-client.sh");
+ commands.add("bin/sql-client.sh");
for (String jar : job.getJars()) {
commands.add("--jar");
Path path = Paths.get(jar);
@@ -269,7 +339,7 @@ public class FlinkContainers implements BeforeAllCallback,
AfterAllCallback {
*/
public JobID submitJob(JobSubmission job) throws IOException,
InterruptedException {
final List<String> commands = new ArrayList<>();
- commands.add("flink/bin/flink");
+ commands.add("bin/flink");
commands.add("run");
if (job.isDetached()) {
diff --git
a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/container/FlinkContainersBuilder.java
b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/container/FlinkContainersBuilder.java
deleted file mode 100644
index 80308e0848e..00000000000
---
a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/container/FlinkContainersBuilder.java
+++ /dev/null
@@ -1,298 +0,0 @@
-/*
- * 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.flink.tests.util.flink.container;
-
-import org.apache.flink.configuration.CheckpointingOptions;
-import org.apache.flink.configuration.ConfigOption;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.HighAvailabilityOptions;
-import org.apache.flink.configuration.JobManagerOptions;
-import org.apache.flink.configuration.RestOptions;
-import org.apache.flink.configuration.TaskManagerOptions;
-
-import org.slf4j.Logger;
-import org.testcontainers.containers.GenericContainer;
-import org.testcontainers.containers.Network;
-import org.testcontainers.containers.output.Slf4jLogConsumer;
-import org.testcontainers.images.builder.ImageFromDockerfile;
-import org.testcontainers.utility.DockerImageName;
-
-import java.io.IOException;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.UUID;
-
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
-/** A builder class for {@link FlinkContainers}. */
-public class FlinkContainersBuilder {
-
- // Hostname of components within container network
- // Hostname of JobManager container
- public static final String JOB_MANAGER_HOSTNAME = "jobmanager";
- // Hostname prefix of TaskManagers. Use "taskmanager-0" for example to
access TM with index 0
- public static final String TASK_MANAGER_HOSTNAME_PREFIX = "taskmanager-";
- // Hostname of Zookeeper container
- public static final String ZOOKEEPER_HOSTNAME = "zookeeper";
-
- // Directories for storing states
- // Path in the JM container for storing checkpoints and savepoints
- public static final Path CHECKPOINT_PATH = Paths.get("/flink/checkpoint");
- // Path in the JM container for persisting HA metadata
- public static final Path HA_STORAGE_PATH = Paths.get("/flink/recovery");
-
- private final List<GenericContainer<?>> dependentContainers = new
ArrayList<>();
- private final Configuration conf = new Configuration();
- private final Map<String, String> envVars = new HashMap<>();
- private final Properties logProperties = new Properties();
-
- private int numTaskManagers = 1;
- private Network network = Network.newNetwork();
- private Logger logger;
- private boolean enableZookeeperHA = false;
-
- /** Sets number of TaskManagers. */
- public FlinkContainersBuilder setNumTaskManagers(int numTaskManagers) {
- this.numTaskManagers = numTaskManagers;
- return this;
- }
-
- /** Sets a single configuration of the cluster. */
- public <T> FlinkContainersBuilder setConfiguration(ConfigOption<T> option,
T value) {
- this.conf.set(option, value);
- return this;
- }
-
- /** Sets configurations of the cluster. */
- public FlinkContainersBuilder setConfiguration(Configuration conf) {
- this.conf.addAll(conf);
- return this;
- }
-
- /**
- * Lets Flink cluster depending on another container, and bind the network
of Flink cluster to
- * the dependent one.
- */
- public FlinkContainersBuilder dependsOn(GenericContainer<?> container) {
- container.withNetwork(this.network);
- this.dependentContainers.add(container);
- return this;
- }
-
- /** Sets environment variable to containers. */
- public FlinkContainersBuilder setEnvironmentVariable(String key, String
value) {
- this.envVars.put(key, value);
- return this;
- }
-
- /** Sets network of the Flink cluster. */
- public FlinkContainersBuilder setNetwork(Network network) {
- this.network = network;
- return this;
- }
-
- /** Sets a logger to the cluster in order to consume STDOUT of containers
to the logger. */
- public FlinkContainersBuilder setLogger(Logger logger) {
- this.logger = logger;
- return this;
- }
-
- /**
- * Sets log4j property.
- *
- * <p>Containers will use "log4j-console.properties" under flink-dist as
the base configuration
- * of loggers. Properties specified by this method will be appended to the
config file, or
- * overwrite the property if already exists in the base config file.
- */
- public FlinkContainersBuilder setLogProperty(String key, String value) {
- this.logProperties.setProperty(key, value);
- return this;
- }
-
- /**
- * Sets log4j properties.
- *
- * <p>Containers will use "log4j-console.properties" under flink-dist as
the base configuration
- * of loggers. Properties specified by this method will be appended to the
config file, or
- * overwrite the property if already exists in the base config file.
- */
- public FlinkContainersBuilder setLogProperties(Properties logProperties) {
- this.logProperties.putAll(logProperties);
- return this;
- }
-
- /** Enables high availability service. */
- public FlinkContainersBuilder enableZookeeperHA() {
- this.enableZookeeperHA = true;
- return this;
- }
-
- /** Builds {@link FlinkContainers}. */
- public FlinkContainers build() {
- // Setup Zookeeper HA
- GenericContainer<?> zookeeper = null;
- if (enableZookeeperHA) {
- enableZookeeperHAConfigurations();
- zookeeper = buildZookeeperContainer();
- }
-
- // Add common configurations
- this.conf.set(JobManagerOptions.ADDRESS, JOB_MANAGER_HOSTNAME);
- this.conf.set(
- CheckpointingOptions.CHECKPOINTS_DIRECTORY,
- CHECKPOINT_PATH.toAbsolutePath().toUri().toString());
- this.conf.set(RestOptions.BIND_ADDRESS, "0.0.0.0");
-
- this.conf.set(JobManagerOptions.BIND_HOST, "0.0.0.0");
- this.conf.set(TaskManagerOptions.BIND_HOST, "0.0.0.0");
- this.conf.removeConfig(TaskManagerOptions.HOST);
-
- // Create temporary directory for building Flink image
- final Path imageBuildingTempDir;
- try {
- imageBuildingTempDir =
Files.createTempDirectory("flink-image-build");
- } catch (IOException e) {
- throw new RuntimeException("Failed to create temporary directory",
e);
- }
-
- // Build JobManager
- final GenericContainer<?> jobManager =
buildJobManagerContainer(imageBuildingTempDir);
-
- // Build TaskManagers
- final List<GenericContainer<?>> taskManagers =
- buildTaskManagerContainers(imageBuildingTempDir);
-
- // Mount HA storage to JobManager
- if (enableZookeeperHA) {
- createTempDirAndMountToContainer("flink-recovery",
HA_STORAGE_PATH, jobManager);
- }
-
- // Mount checkpoint storage to JobManager
- createTempDirAndMountToContainer("flink-checkpoint", CHECKPOINT_PATH,
jobManager);
-
- return new FlinkContainers(jobManager, taskManagers, zookeeper, conf);
- }
-
- // --------------------------- Helper Functions
-------------------------------------
-
- private GenericContainer<?> buildJobManagerContainer(Path tempDirectory) {
- // Configure JobManager
- final Configuration jobManagerConf = new Configuration();
- jobManagerConf.addAll(this.conf);
- // Build JobManager container
- final ImageFromDockerfile jobManagerImage;
- try {
- jobManagerImage =
- new FlinkImageBuilder()
- .setTempDirectory(tempDirectory)
- .setConfiguration(jobManagerConf)
- .setLogProperties(logProperties)
- .asJobManager()
- .build();
- } catch (ImageBuildException e) {
- throw new RuntimeException("Failed to build JobManager image", e);
- }
- return configureContainer(
- new GenericContainer<>(jobManagerImage),
JOB_MANAGER_HOSTNAME, "JobManager")
- .withExposedPorts(jobManagerConf.get(RestOptions.PORT));
- }
-
- private List<GenericContainer<?>> buildTaskManagerContainers(Path
tempDirectory) {
- List<GenericContainer<?>> taskManagers = new ArrayList<>();
- for (int i = 0; i < numTaskManagers; i++) {
- // Configure TaskManager
- final Configuration taskManagerConf = new Configuration();
- taskManagerConf.addAll(this.conf);
- final String taskManagerHostName = TASK_MANAGER_HOSTNAME_PREFIX +
i;
- taskManagerConf.set(TaskManagerOptions.HOST, taskManagerHostName);
- // Build TaskManager container
- final ImageFromDockerfile taskManagerImage;
- try {
- taskManagerImage =
- new FlinkImageBuilder()
- .setTempDirectory(tempDirectory)
- .setConfiguration(taskManagerConf)
- .setLogProperties(logProperties)
- .asTaskManager()
- .build();
- } catch (ImageBuildException e) {
- throw new RuntimeException("Failed to build TaskManager
image", e);
- }
- taskManagers.add(
- configureContainer(
- new GenericContainer<>(taskManagerImage),
- taskManagerHostName,
- "TaskManager-" + i));
- }
- return taskManagers;
- }
-
- private GenericContainer<?> buildZookeeperContainer() {
- return configureContainer(
- new
GenericContainer<>(DockerImageName.parse("zookeeper").withTag("3.5.9")),
- ZOOKEEPER_HOSTNAME,
- "Zookeeper");
- }
-
- private GenericContainer<?> configureContainer(
- GenericContainer<?> container, String networkAlias, String
logPrefix) {
- // Set dependent containers
- for (GenericContainer<?> dependentContainer : dependentContainers) {
- container.dependsOn(dependentContainer);
- }
- // Setup network
- container.withNetwork(network);
- container.withNetworkAliases(networkAlias);
- // Setup logger
- if (logger != null) {
- container.withLogConsumer(new
Slf4jLogConsumer(logger).withPrefix(logPrefix));
- }
- // Add environment variables
- container.withEnv(envVars);
- return container;
- }
-
- private void enableZookeeperHAConfigurations() {
- // Set HA related configurations
- checkNotNull(this.conf, "Configuration should not be null for setting
HA service");
- conf.set(HighAvailabilityOptions.HA_MODE, "zookeeper");
- conf.set(HighAvailabilityOptions.HA_ZOOKEEPER_QUORUM,
ZOOKEEPER_HOSTNAME + ":" + "2181");
- conf.set(HighAvailabilityOptions.HA_ZOOKEEPER_ROOT, "/flink");
- conf.set(HighAvailabilityOptions.HA_CLUSTER_ID, "flink-container-" +
UUID.randomUUID());
- conf.set(HighAvailabilityOptions.HA_STORAGE_PATH,
HA_STORAGE_PATH.toUri().toString());
- }
-
- private void createTempDirAndMountToContainer(
- String tempDirPrefix, Path containerPath, GenericContainer<?>
container) {
- try {
- Path tempDirPath = Files.createTempDirectory(tempDirPrefix);
- container.withFileSystemBind(
- tempDirPath.toAbsolutePath().toString(),
- containerPath.toAbsolutePath().toString());
- } catch (IOException e) {
- throw new IllegalStateException("Failed to create temporary
recovery directory", e);
- }
- }
-}
diff --git
a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/container/FlinkContainersSettings.java
b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/container/FlinkContainersSettings.java
new file mode 100644
index 00000000000..05a63b5d131
--- /dev/null
+++
b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/container/FlinkContainersSettings.java
@@ -0,0 +1,599 @@
+/*
+ * 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.flink.tests.util.flink.container;
+
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.HighAvailabilityOptions;
+import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.configuration.MemorySize;
+import org.apache.flink.configuration.RestOptions;
+import org.apache.flink.configuration.TaskManagerOptions;
+
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Properties;
+import java.util.UUID;
+
+import static
org.apache.flink.configuration.HeartbeatManagerOptions.HEARTBEAT_INTERVAL;
+import static
org.apache.flink.configuration.HeartbeatManagerOptions.HEARTBEAT_TIMEOUT;
+import static
org.apache.flink.configuration.JobManagerOptions.SLOT_REQUEST_TIMEOUT;
+import static
org.apache.flink.configuration.MetricOptions.METRIC_FETCHER_UPDATE_INTERVAL;
+
+/** The central configuration holder for Flink container-based test
environments. */
+public class FlinkContainersSettings {
+ private final String baseImage;
+ private final int numTaskManagers;
+ private final int numSlotsPerTaskManager;
+ private final Collection<String> jarPaths;
+ private final Configuration flinkConfig;
+ private final String taskManagerHostnamePrefix;
+ private final Boolean buildFromFlinkDist;
+ private final String flinkDistLocation;
+ private final String flinkHome;
+ private final String checkpointPath;
+ private final String haStoragePath;
+ private final Boolean zookeeperHA;
+ private final String zookeeperHostname;
+ private final Properties logProperties;
+
+ // Defaults
+ private static final long DEFAULT_METRIC_FETCHER_UPDATE_INTERVAL_MS =
1000L;
+ private static final long DEFAULT_SLOT_REQUEST_TIMEOUT_MS = 10_000L;
+ private static final long DEFAULT_HEARTBEAT_TIMEOUT_MS = 5_000L;
+ private static final long DEFAULT_HEARTBEAT_INTERVAL_MS = 1000L;
+ private static final MemorySize DEFAULT_TM_TOTAL_PROCESS_MEMORY =
MemorySize.ofMebiBytes(1728);
+ private static final MemorySize DEFAULT_JM_TOTAL_PROCESS_MEMORY =
MemorySize.ofMebiBytes(1600);
+
+ private static final int DEFAULT_NUM_TASK_MANAGERS = 1;
+ private static final int DEFAULT_NUM_SLOTS_PER_TASK_MANAGER = 1;
+ private static final String DEFAULT_TASK_MANAGERS_HOSTNAME_PREFIX =
"taskmanager-";
+ private static final String DEFAULT_JOB_MANAGER_HOSTNAME = "jobmanager";
+ private static final String DEFAULT_BIND_ADDRESS = "0.0.0.0";
+ private static final String DEFAULT_FLINK_HOME = "/opt/flink";
+ private static final String DEFAULT_CHECKPOINT_PATH = DEFAULT_FLINK_HOME +
"/checkpoint";
+ private static final String DEFAULT_HA_STORAGE_PATH = DEFAULT_FLINK_HOME +
"/recovery";
+
+ private static final String DEFAULT_ZOOKEEPER_HOSTNAME = "zookeeper";
+ private static final String DEFAULT_ZOOKEEPER_QUORUM =
DEFAULT_ZOOKEEPER_HOSTNAME + ":2181";
+ // --
+
+ private FlinkContainersSettings(Builder builder) {
+ baseImage = builder.baseImage;
+ numTaskManagers = builder.numTaskManagers;
+ numSlotsPerTaskManager = builder.numSlotsPerTaskManager;
+ jarPaths = builder.jarPaths;
+ flinkConfig = builder.flinkConfiguration;
+ taskManagerHostnamePrefix = builder.taskManagerHostnamePrefix;
+ buildFromFlinkDist = builder.buildFromFlinkDist;
+ flinkDistLocation = builder.flinkDistLocation;
+ flinkHome = builder.flinkHome;
+ checkpointPath = builder.checkpointPath;
+ haStoragePath = builder.haStoragePath;
+ zookeeperHA = builder.zookeeperHA;
+ zookeeperHostname = builder.zookeeperHostname;
+ logProperties = builder.logProperties;
+ }
+
+ /**
+ * A new builder for {@code FlinkContainersConfig}.
+ *
+ * @return The builder.
+ */
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ /**
+ * {@code FlinkContainersConfig} based on defaults.
+ *
+ * @return The Flink containers config.
+ */
+ public static FlinkContainersSettings defaultConfig() {
+ return builder().build();
+ }
+
+ /**
+ * {@code FlinkContainersConfig} based on provided Flink configuration.
+ *
+ * @param config The config.
+ * @return The flink containers config.
+ */
+ public static FlinkContainersSettings basedOn(Configuration config) {
+ return builder().basedOn(config).build();
+ }
+
+ /** {@code FlinkContainersConfig} builder static inner class. */
+ public static final class Builder {
+ private String baseImage;
+ private int numTaskManagers = DEFAULT_NUM_TASK_MANAGERS;
+ private int numSlotsPerTaskManager =
DEFAULT_NUM_SLOTS_PER_TASK_MANAGER;
+ private Collection<String> jarPaths = new ArrayList<>();
+ private Configuration flinkConfiguration = defaultFlinkConfig();
+ private String taskManagerHostnamePrefix =
DEFAULT_TASK_MANAGERS_HOSTNAME_PREFIX;
+ private Boolean buildFromFlinkDist = true;
+ private String flinkDistLocation;
+ private String flinkHome = DEFAULT_FLINK_HOME;
+ private String checkpointPath = DEFAULT_CHECKPOINT_PATH;
+ private String haStoragePath = DEFAULT_HA_STORAGE_PATH;
+ private Boolean zookeeperHA = false;
+ private String zookeeperHostname = DEFAULT_ZOOKEEPER_HOSTNAME;
+ private Properties logProperties = defaultLoggingProperties();
+
+ private Builder() {}
+
+ /**
+ * Sets the {@code baseImage} and returns a reference to this Builder
enabling method
+ * chaining.
+ *
+ * @param baseImage The {@code baseImage} to set.
+ * @return A reference to this Builder.
+ */
+ public Builder baseImage(String baseImage) {
+ this.baseImage = baseImage;
+ this.buildFromFlinkDist = false;
+ return this;
+ }
+
+ /**
+ * Sets the {@code flinkDistLocation} and returns a reference to this
Builder enabling
+ * method chaining.
+ *
+ * @param flinkDistLocation The {@code flinkDistLocation} to set.
+ * @return A reference to this Builder.
+ */
+ public Builder flinkDistLocation(String flinkDistLocation) {
+ this.flinkDistLocation = flinkDistLocation;
+ this.buildFromFlinkDist = true;
+ return this;
+ }
+
+ /**
+ * Sets the path of the Flink distribution inside the container.
Returns a reference to this
+ * Builder enabling method chaining.
+ *
+ * @param flinkHome The {@code flinkHome} to set.
+ * @return A reference to this Builder.
+ */
+ public Builder flinkHome(String flinkHome) {
+ this.flinkHome = flinkHome;
+ return this;
+ }
+
+ /**
+ * Sets the {@code checkpointPath} and returns a reference to this
Builder enabling method
+ * chaining.
+ *
+ * @param checkpointPath The checkpoint path to set.
+ * @return A reference to this Builder.
+ */
+ public Builder checkpointPath(String checkpointPath) {
+ this.checkpointPath = checkpointPath;
+ return setConfigOption(
+ CheckpointingOptions.CHECKPOINTS_DIRECTORY,
toUri(checkpointPath));
+ }
+
+ /**
+ * Sets the {@code haStoragePath} and returns a reference to this
Builder enabling method
+ * chaining.
+ *
+ * @param haStoragePath The path for storing HA data.
+ * @return A reference to this Builder.
+ */
+ public Builder haStoragePath(String haStoragePath) {
+ this.haStoragePath = haStoragePath;
+ return setConfigOption(HighAvailabilityOptions.HA_STORAGE_PATH,
toUri(haStoragePath));
+ }
+
+ /**
+ * Sets the {@code zookeeperHostname} and returns a reference to this
Builder enabling
+ * method chaining.
+ *
+ * @param zookeeperHostname The Zookeeper hostname.
+ * @return A reference to this Builder.
+ */
+ public Builder zookeeperHostname(String zookeeperHostname) {
+ this.zookeeperHostname = zookeeperHostname;
+ return this;
+ }
+
+ /**
+ * Enables Zookeeper HA. NOTE: this option uses default HA
configuration. If you want to use
+ * non-default configuration, you should provide all settings,
including the HA_MODE
+ * directly via the {@code basedOn()} method instead.
+ *
+ * @return A reference to this Builder.
+ */
+ public Builder enableZookeeperHA() {
+ zookeeperHA = true;
+ setConfigOption(HighAvailabilityOptions.HA_MODE, "zookeeper");
+ setConfigOption(HighAvailabilityOptions.HA_ZOOKEEPER_QUORUM,
DEFAULT_ZOOKEEPER_QUORUM);
+ setConfigOption(
+ HighAvailabilityOptions.HA_CLUSTER_ID, "flink-container-"
+ UUID.randomUUID());
+ setConfigOption(
+ HighAvailabilityOptions.HA_STORAGE_PATH,
toUri(DEFAULT_HA_STORAGE_PATH));
+ return this;
+ }
+
+ /**
+ * Sets the {@code numTaskManagers} and returns a reference to this
Builder enabling method
+ * chaining.
+ *
+ * @param numTaskManagers The {@code numTaskManagers} to set.
+ * @return A reference to this Builder.
+ */
+ public Builder numTaskManagers(int numTaskManagers) {
+ this.numTaskManagers = numTaskManagers;
+ return this;
+ }
+
+ /**
+ * Sets the {@code numSlotsPerTaskManager} and returns a reference to
this Builder enabling
+ * method chaining. It also adds this property into the {@code
flinkConfiguration} field.
+ *
+ * @param numSlotsPerTaskManager The {@code numSlotsPerTaskManager} to
set.
+ * @return A reference to this Builder.
+ */
+ public Builder numSlotsPerTaskManager(int numSlotsPerTaskManager) {
+ this.numSlotsPerTaskManager = numSlotsPerTaskManager;
+ return setConfigOption(TaskManagerOptions.NUM_TASK_SLOTS,
numSlotsPerTaskManager);
+ }
+
+ /**
+ * Sets the {@code jarPaths} and returns a reference to this Builder
enabling method
+ * chaining.
+ *
+ * @param jarPaths The {@code jarPaths} to set.
+ * @return A reference to this Builder.
+ */
+ public Builder jarPaths(String... jarPaths) {
+ this.jarPaths = Arrays.asList(jarPaths);
+ return this;
+ }
+
+ /**
+ * Sets the {@code jarPaths} and returns a reference to this Builder
enabling method
+ * chaining.
+ *
+ * @param jarPaths The {@code jarPaths} to set.
+ * @return A reference to this Builder.
+ */
+ public Builder jarPaths(Collection<String> jarPaths) {
+ this.jarPaths = jarPaths;
+ return this;
+ }
+
+ /**
+ * Sets a single Flink configuration parameter (the options for
flink-conf.yaml) and returns
+ * a reference to this Builder enabling method chaining.
+ *
+ * @param <T> The type parameter.
+ * @param option The option.
+ * @param value The value.
+ * @return A reference to this Builder.
+ */
+ public <T> Builder setConfigOption(ConfigOption<T> option, T value) {
+ this.flinkConfiguration.set(option, value);
+ return this;
+ }
+
+ /**
+ * Sets a single Flink logging configuration property in the log4j
format and returns a
+ * reference to this Builder enabling method chaining.
+ *
+ * @param key The property key.
+ * @param value The property value.
+ * @return A reference to this Builder.
+ */
+ public Builder setLogProperty(String key, String value) {
+ this.logProperties.setProperty(key, value);
+ return this;
+ }
+
+ /**
+ * Merges the provided {@code config} with the default config,
potentially overwriting the
+ * defaults in case of collisions. Returns a reference to this Builder
enabling method
+ * chaining.
+ *
+ * @param <T> the type parameter
+ * @param config The {@code config} to add.
+ * @return A reference to this Builder.
+ */
+ public <T> Builder basedOn(Configuration config) {
+ this.flinkConfiguration.addAll(config);
+ return this;
+ }
+
+ /**
+ * Sets the {@code flinkConfiguration} value to {@code config} and
returns a reference to
+ * this Builder enabling method chaining.
+ *
+ * @param <T> the type parameter
+ * @param config The {@code config} to set.
+ * @return A reference to this Builder.
+ */
+ public <T> Builder fullConfiguration(Configuration config) {
+ this.flinkConfiguration = config;
+ return this;
+ }
+
+ /**
+ * Sets the {@code taskManagerHostnamePrefix} and returns a reference
to this Builder
+ * enabling method chaining.
+ *
+ * @param taskManagerHostnamePrefix The {@code
taskManagerHostnamePrefix} to set.
+ * @return A reference to this Builder.
+ */
+ public Builder taskManagerHostnamePrefix(String
taskManagerHostnamePrefix) {
+ this.taskManagerHostnamePrefix = taskManagerHostnamePrefix;
+ return this;
+ }
+
+ /**
+ * Sets the job manager hostname and returns a reference to this
Builder enabling method
+ * chaining.
+ *
+ * @param jobManagerHostname The job manager hostname to set.
+ * @return A reference to this Builder.
+ */
+ public Builder jobManagerHostname(String jobManagerHostname) {
+ return setConfigOption(JobManagerOptions.ADDRESS,
jobManagerHostname);
+ }
+
+ private Configuration defaultFlinkConfig() {
+ final Configuration config = new Configuration();
+ config.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY,
DEFAULT_TM_TOTAL_PROCESS_MEMORY);
+ config.set(JobManagerOptions.TOTAL_PROCESS_MEMORY,
DEFAULT_JM_TOTAL_PROCESS_MEMORY);
+ config.set(HEARTBEAT_INTERVAL, DEFAULT_HEARTBEAT_INTERVAL_MS);
+ config.set(HEARTBEAT_TIMEOUT, DEFAULT_HEARTBEAT_TIMEOUT_MS);
+ config.set(SLOT_REQUEST_TIMEOUT, DEFAULT_SLOT_REQUEST_TIMEOUT_MS);
+ config.set(METRIC_FETCHER_UPDATE_INTERVAL,
DEFAULT_METRIC_FETCHER_UPDATE_INTERVAL_MS);
+ config.set(JobManagerOptions.ADDRESS,
DEFAULT_JOB_MANAGER_HOSTNAME);
+
+ config.set(RestOptions.BIND_ADDRESS, DEFAULT_BIND_ADDRESS);
+ config.set(TaskManagerOptions.BIND_HOST, DEFAULT_BIND_ADDRESS);
+ config.set(JobManagerOptions.BIND_HOST, DEFAULT_BIND_ADDRESS);
+
+ config.set(CheckpointingOptions.CHECKPOINTS_DIRECTORY,
toUri(DEFAULT_CHECKPOINT_PATH));
+
+ return config;
+ }
+
+ private Properties defaultLoggingProperties() {
+ final Properties logProperties = new Properties();
+ logProperties.setProperty(
+ "appender.rolling.strategy.max",
"${env:MAX_LOG_FILE_NUMBER:-10}");
+ logProperties.setProperty("appender.rolling.policies.size.size",
"100MB");
+ logProperties.setProperty("logger.zookeeper.name",
"org.apache.zookeeper");
+ logProperties.setProperty(
+ "logger.shaded_zookeeper.name",
"org.apache.flink.shaded.zookeeper3");
+ logProperties.setProperty("logger.zookeeper.level", "INFO");
+ logProperties.setProperty("appender.console.name",
"ConsoleAppender");
+ logProperties.setProperty("appender.rolling.fileName",
"${sys:log.file}");
+ logProperties.setProperty("appender.rolling.filePattern",
"${sys:log.file}.%i");
+ logProperties.setProperty("rootLogger.appenderRef.rolling.ref",
"RollingFileAppender");
+ logProperties.setProperty("logger.akka.name", "akka");
+ logProperties.setProperty("appender.console.type", "CONSOLE");
+ logProperties.setProperty("appender.rolling.append", "true");
+ logProperties.setProperty("appender.console.layout.type",
"PatternLayout");
+ logProperties.setProperty("appender.rolling.name",
"RollingFileAppender");
+ logProperties.setProperty("rootLogger.appenderRef.console.ref",
"ConsoleAppender");
+ logProperties.setProperty(
+ "appender.rolling.policies.size.type",
"SizeBasedTriggeringPolicy");
+ logProperties.setProperty(
+ "appender.console.layout.pattern",
+ "%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n");
+ logProperties.setProperty("rootLogger.level", "INFO");
+ logProperties.setProperty("logger.hadoop.name",
"org.apache.hadoop");
+ logProperties.setProperty("logger.shaded_zookeeper.level", "INFO");
+ logProperties.setProperty("appender.rolling.layout.type",
"PatternLayout");
+ logProperties.setProperty("logger.kafka.name", "org.apache.kafka");
+ logProperties.setProperty("logger.netty.level", "OFF");
+ logProperties.setProperty("appender.rolling.type", "RollingFile");
+ logProperties.setProperty("logger.akka.level", "INFO");
+ logProperties.setProperty("logger.hadoop.level", "INFO");
+ logProperties.setProperty(
+ "appender.rolling.layout.pattern",
+ "%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n");
+ logProperties.setProperty(
+ "appender.rolling.policies.startup.type",
"OnStartupTriggeringPolicy");
+ logProperties.setProperty(
+ "logger.netty.name",
"org.jboss.netty.channel.DefaultChannelPipeline");
+ logProperties.setProperty("appender.rolling.strategy.type",
"DefaultRolloverStrategy");
+ logProperties.setProperty("appender.rolling.policies.type",
"Policies");
+ logProperties.setProperty("monitorInterval", "30");
+ logProperties.setProperty("logger.kafka.level", "INFO");
+ return logProperties;
+ }
+
+ /**
+ * Returns a {@code FlinkContainersConfig} built from the parameters
previously set.
+ *
+ * @return A {@code FlinkContainersConfig} built with parameters of
this {@code
+ * FlinkContainersConfig.Builder}.
+ */
+ public FlinkContainersSettings build() {
+ return new FlinkContainersSettings(this);
+ }
+ }
+
+ /**
+ * Gets base image.
+ *
+ * @return The base image.
+ */
+ public String getBaseImage() {
+ return baseImage;
+ }
+
+ /**
+ * Gets number of task managers.
+ *
+ * @return The number task managers.
+ */
+ public int getNumTaskManagers() {
+ return numTaskManagers;
+ }
+
+ /**
+ * Gets number slots per task manager.
+ *
+ * @return The number slots per task manager.
+ */
+ public int getNumSlotsPerTaskManager() {
+ return numSlotsPerTaskManager;
+ }
+
+ /**
+ * Gets jar paths.
+ *
+ * @return The jar paths.
+ */
+ public Collection<String> getJarPaths() {
+ return jarPaths;
+ }
+
+ /**
+ * Gets flink configuration.
+ *
+ * @return The flink configuration.
+ */
+ public Configuration getFlinkConfig() {
+ return flinkConfig;
+ }
+
+ /**
+ * Gets task manager hostname prefix.
+ *
+ * @return The task manager hostname prefix.
+ */
+ public String getTaskManagerHostnamePrefix() {
+ return taskManagerHostnamePrefix;
+ }
+
+ /**
+ * Gets job manager hostname.
+ *
+ * @return The job manager hostname.
+ */
+ public String getJobManagerHostname() {
+ return flinkConfig.getString(JobManagerOptions.ADDRESS);
+ }
+
+ /**
+ * Returns whether to build from flink-dist or from an existing base
container. Also see the
+ * {@code baseImage} property.
+ */
+ public Boolean isBuildFromFlinkDist() {
+ return buildFromFlinkDist;
+ }
+
+ /**
+ * Gets flink dist location.
+ *
+ * @return The flink dist location.
+ */
+ public String getFlinkDistLocation() {
+ return flinkDistLocation;
+ }
+
+ /**
+ * Gets flink home.
+ *
+ * @return The flink home path.
+ */
+ public String getFlinkHome() {
+ return flinkHome;
+ }
+
+ /**
+ * Gets checkpoint path.
+ *
+ * @return The checkpoint path.
+ */
+ public String getCheckpointPath() {
+ return checkpointPath;
+ }
+
+ /**
+ * Gets HA storage path.
+ *
+ * @return The ha storage path.
+ */
+ public String getHaStoragePath() {
+ return haStoragePath;
+ }
+
+ /**
+ * Gets ha storage path uri.
+ *
+ * @return The HA storage path as URI (prefixed with file://).
+ */
+ public String getHaStoragePathUri() {
+ return Paths.get(getHaStoragePath()).toUri().toString();
+ }
+
+ /**
+ * Gets default flink home.
+ *
+ * @return The default flink home path.
+ */
+ public static String getDefaultFlinkHome() {
+ return DEFAULT_FLINK_HOME;
+ }
+
+ /**
+ * Gets default checkpoint path.
+ *
+ * @return The default checkpoint path.
+ */
+ public static String getDefaultCheckpointPath() {
+ return DEFAULT_CHECKPOINT_PATH;
+ }
+
+ /** Is zookeeper HA boolean. */
+ public Boolean isZookeeperHA() {
+ return zookeeperHA;
+ }
+
+ /**
+ * Gets Zookeeper hostname.
+ *
+ * @return The Zookeeper hostname.
+ */
+ public String getZookeeperHostname() {
+ return zookeeperHostname;
+ }
+
+ /**
+ * Gets logging properties.
+ *
+ * @return The logging properties.
+ */
+ public Properties getLogProperties() {
+ return logProperties;
+ }
+
+ private static String toUri(String path) {
+ return Paths.get(path).toUri().toString();
+ }
+}
diff --git
a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/container/FlinkImageBuilder.java
b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/container/FlinkImageBuilder.java
index 25fb0af821e..cbe52c35f93 100644
---
a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/container/FlinkImageBuilder.java
+++
b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/container/FlinkImageBuilder.java
@@ -28,10 +28,8 @@ import org.slf4j.LoggerFactory;
import org.testcontainers.DockerClientFactory;
import org.testcontainers.images.builder.ImageFromDockerfile;
-import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
-import java.io.InputStream;
import java.io.OutputStream;
import java.nio.file.Files;
import java.nio.file.Path;
@@ -51,8 +49,8 @@ import static org.apache.flink.util.Preconditions.checkState;
public class FlinkImageBuilder {
private static final Logger LOG =
LoggerFactory.getLogger(FlinkImageBuilder.class);
- private static final String FLINK_BASE_IMAGE_NAME = "flink-dist-base";
- private static final String DEFAULT_IMAGE_NAME = "flink-dist-configured";
+ private static final String FLINK_BASE_IMAGE_BUILD_NAME = "flink-base";
+ private static final String DEFAULT_IMAGE_NAME_BUILD_PREFIX =
"flink-configured";
private static final Duration DEFAULT_TIMEOUT = Duration.ofMinutes(5);
private static final String LOG4J_PROPERTIES_FILENAME =
"log4j-console.properties";
@@ -60,13 +58,15 @@ public class FlinkImageBuilder {
private final Properties logProperties = new Properties();
private Path tempDirectory;
- private String imageName = DEFAULT_IMAGE_NAME;
+ private String imageNamePrefix = DEFAULT_IMAGE_NAME_BUILD_PREFIX;
private String imageNameSuffix;
- private Path flinkDist = FileUtils.findFlinkDist();
+ private Path flinkDist;
private String javaVersion;
private Configuration conf;
private Duration timeout = DEFAULT_TIMEOUT;
private String startupCommand;
+ private String baseImage;
+ private String flinkHome = FlinkContainersSettings.getDefaultFlinkHome();
/**
* Sets temporary path for holding temp files when building the image.
@@ -80,12 +80,23 @@ public class FlinkImageBuilder {
}
/**
- * Sets the name of building image.
+ * Sets flink home.
*
- * <p>If the name is not specified, {@link #DEFAULT_IMAGE_NAME} will be
used.
+ * @param flinkHome The flink home.
+ * @return The flink home.
*/
- public FlinkImageBuilder setImageName(String imageName) {
- this.imageName = imageName;
+ public FlinkImageBuilder setFlinkHome(String flinkHome) {
+ this.flinkHome = flinkHome;
+ return this;
+ }
+
+ /**
+ * Sets the prefix name of building image.
+ *
+ * <p>If the name is not specified, {@link
#DEFAULT_IMAGE_NAME_BUILD_PREFIX} will be used.
+ */
+ public FlinkImageBuilder setImageNamePrefix(String imageNamePrefix) {
+ this.imageNamePrefix = imageNamePrefix;
return this;
}
@@ -149,7 +160,7 @@ public class FlinkImageBuilder {
/** Use this image for building a JobManager. */
public FlinkImageBuilder asJobManager() {
checkStartupCommandNotSet();
- this.startupCommand = "flink/bin/jobmanager.sh start-foreground &&
tail -f /dev/null";
+ this.startupCommand = "bin/jobmanager.sh start-foreground && tail -f
/dev/null";
this.imageNameSuffix = "jobmanager";
return this;
}
@@ -157,7 +168,7 @@ public class FlinkImageBuilder {
/** Use this image for building a TaskManager. */
public FlinkImageBuilder asTaskManager() {
checkStartupCommandNotSet();
- this.startupCommand = "flink/bin/taskmanager.sh start-foreground &&
tail -f /dev/null";
+ this.startupCommand = "bin/taskmanager.sh start-foreground && tail -f
/dev/null";
this.imageNameSuffix = "taskmanager";
return this;
}
@@ -170,28 +181,47 @@ public class FlinkImageBuilder {
return this;
}
+ /**
+ * Sets base image.
+ *
+ * @param baseImage The base image.
+ * @return A reference to this Builder.
+ */
+ public FlinkImageBuilder setBaseImage(String baseImage) {
+ this.baseImage = baseImage;
+ return this;
+ }
+
/** Build the image. */
public ImageFromDockerfile build() throws ImageBuildException {
sanityCheck();
- final String finalImageName = imageName + "-" + imageNameSuffix;
+ final String finalImageName = imageNamePrefix + "-" + imageNameSuffix;
try {
- // Build base image first
- buildBaseImage(flinkDist);
- final Path flinkConfFile =
createTemporaryFlinkConfFile(tempDirectory);
+ if (baseImage == null) {
+ baseImage = FLINK_BASE_IMAGE_BUILD_NAME;
+ if (flinkDist == null) {
+ flinkDist = FileUtils.findFlinkDist();
+ }
+ // Build base image first
+ buildBaseImage(flinkDist);
+ }
+
+ final Path flinkConfFile = createTemporaryFlinkConfFile(conf,
tempDirectory);
+
final Path log4jPropertiesFile =
createTemporaryLog4jPropertiesFile(tempDirectory);
// Copy flink-conf.yaml into image
filesToCopy.put(
flinkConfFile,
- Paths.get("flink", "conf",
GlobalConfiguration.FLINK_CONF_FILENAME));
+ Paths.get(flinkHome, "conf",
GlobalConfiguration.FLINK_CONF_FILENAME));
filesToCopy.put(
- log4jPropertiesFile, Paths.get("flink", "conf",
LOG4J_PROPERTIES_FILENAME));
+ log4jPropertiesFile, Paths.get(flinkHome, "conf",
LOG4J_PROPERTIES_FILENAME));
final ImageFromDockerfile image =
new ImageFromDockerfile(finalImageName)
.withDockerfileFromBuilder(
builder -> {
// Build from base image
- builder.from(FLINK_BASE_IMAGE_NAME);
+ builder.from(baseImage);
// Copy files into image
filesToCopy.forEach(
(from, to) ->
@@ -212,19 +242,22 @@ public class FlinkImageBuilder {
return;
}
LOG.info("Building Flink base image with flink-dist at {}", flinkDist);
- new ImageFromDockerfile(FLINK_BASE_IMAGE_NAME)
+ new ImageFromDockerfile(FLINK_BASE_IMAGE_BUILD_NAME)
.withDockerfileFromBuilder(
builder ->
builder.from("openjdk:" +
getJavaVersionSuffix())
- .copy("flink", "flink")
+ .copy(flinkHome, flinkHome)
.build())
- .withFileFromPath("flink", flinkDist)
+ .withFileFromPath(flinkHome, flinkDist)
.get(timeout.toMillis(), TimeUnit.MILLISECONDS);
}
private boolean baseImageExists() {
try {
-
DockerClientFactory.instance().client().inspectImageCmd(FLINK_BASE_IMAGE_NAME).exec();
+ DockerClientFactory.instance()
+ .client()
+ .inspectImageCmd(FLINK_BASE_IMAGE_BUILD_NAME)
+ .exec();
return true;
} catch (NotFoundException e) {
return false;
@@ -250,17 +283,8 @@ public class FlinkImageBuilder {
}
}
- private Path createTemporaryFlinkConfFile(Path tempDirectory) throws
IOException {
- // Load flink-conf.yaml under flink-dist as a base configuration
- final Configuration finalConfiguration =
- GlobalConfiguration.loadConfiguration(
- flinkDist.resolve("conf").toAbsolutePath().toString());
-
- // Merge user specific configurations in builder
- if (this.conf != null) {
- finalConfiguration.addAll(this.conf);
- }
-
+ private Path createTemporaryFlinkConfFile(Configuration
finalConfiguration, Path tempDirectory)
+ throws IOException {
// Create a temporary flink-conf.yaml file and write merged
configurations into it
Path flinkConfFile =
tempDirectory.resolve(GlobalConfiguration.FLINK_CONF_FILENAME);
Files.write(
@@ -273,19 +297,10 @@ public class FlinkImageBuilder {
}
private Path createTemporaryLog4jPropertiesFile(Path tempDirectory) throws
IOException {
- // Load log4j-console.properties under flink-dist as base properties
- final Path logPropertiesPath =
flinkDist.resolve("conf").resolve(LOG4J_PROPERTIES_FILENAME);
- final Properties mergedLogProperties = new Properties();
- try (InputStream input = new
FileInputStream(logPropertiesPath.toFile())) {
- mergedLogProperties.load(input);
- // Merge all log properties
- mergedLogProperties.putAll(this.logProperties);
- }
-
// Create a temporary log4j.properties file and write merged
properties into it
Path log4jPropFile = tempDirectory.resolve(LOG4J_PROPERTIES_FILENAME);
try (OutputStream output = new
FileOutputStream(log4jPropFile.toFile())) {
- mergedLogProperties.store(output, null);
+ logProperties.store(output, null);
}
return log4jPropFile;
diff --git
a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/container/FlinkTestcontainersConfigurator.java
b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/container/FlinkTestcontainersConfigurator.java
new file mode 100644
index 00000000000..7b5ac72508d
--- /dev/null
+++
b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/container/FlinkTestcontainersConfigurator.java
@@ -0,0 +1,183 @@
+/*
+ * 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.flink.tests.util.flink.container;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.RestOptions;
+import org.apache.flink.configuration.TaskManagerOptions;
+
+import org.slf4j.Logger;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.images.builder.ImageFromDockerfile;
+import org.testcontainers.utility.DockerImageName;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+
+/** Orchestrates configuration of Flink containers within Testcontainers
framework. */
+class FlinkTestcontainersConfigurator {
+
+ private final TestcontainersSettings testcontainersSettings;
+ private final FlinkContainersSettings flinkContainersSettings;
+
+ FlinkTestcontainersConfigurator(
+ FlinkContainersSettings flinkContainersSettings,
+ TestcontainersSettings testcontainersSettings) {
+ this.testcontainersSettings = testcontainersSettings;
+ this.flinkContainersSettings = flinkContainersSettings;
+ }
+
+ private GenericContainer<?> configureJobManagerContainer(Path
tempDirectory) {
+ // Configure JobManager
+ final Configuration jobManagerConf = new Configuration();
+ jobManagerConf.addAll(flinkContainersSettings.getFlinkConfig());
+ // Build JobManager container
+ final ImageFromDockerfile jobManagerImage;
+ try {
+ jobManagerImage =
+ new FlinkImageBuilder()
+ .setTempDirectory(tempDirectory)
+ .setConfiguration(jobManagerConf)
+
.setLogProperties(flinkContainersSettings.getLogProperties())
+
.setBaseImage(flinkContainersSettings.getBaseImage())
+ .asJobManager()
+ .build();
+ } catch (ImageBuildException e) {
+ throw new RuntimeException("Failed to build JobManager image", e);
+ }
+ return configureContainer(
+ new GenericContainer<>(jobManagerImage),
+ flinkContainersSettings.getJobManagerHostname(),
+ "JobManager")
+ .withExposedPorts(jobManagerConf.get(RestOptions.PORT));
+ }
+
+ private List<GenericContainer<?>> configureTaskManagerContainers(Path
tempDirectory) {
+ List<GenericContainer<?>> taskManagers = new ArrayList<>();
+ for (int i = 0; i < flinkContainersSettings.getNumTaskManagers(); i++)
{
+ // Configure TaskManager
+ final Configuration taskManagerConf = new Configuration();
+ taskManagerConf.addAll(flinkContainersSettings.getFlinkConfig());
+ final String taskManagerHostName =
+ flinkContainersSettings.getTaskManagerHostnamePrefix() + i;
+ taskManagerConf.set(TaskManagerOptions.HOST, taskManagerHostName);
+ // Build TaskManager container
+ final ImageFromDockerfile taskManagerImage;
+ try {
+ taskManagerImage =
+ new FlinkImageBuilder()
+ .setTempDirectory(tempDirectory)
+ .setConfiguration(taskManagerConf)
+
.setLogProperties(flinkContainersSettings.getLogProperties())
+
.setBaseImage(flinkContainersSettings.getBaseImage())
+ .asTaskManager()
+ .build();
+ } catch (ImageBuildException e) {
+ throw new RuntimeException("Failed to build TaskManager
image", e);
+ }
+ taskManagers.add(
+ configureContainer(
+ new GenericContainer<>(taskManagerImage),
+ taskManagerHostName,
+ "TaskManager-" + i));
+ }
+ return taskManagers;
+ }
+
+ private GenericContainer<?> configureZookeeperContainer() {
+ return configureContainer(
+ new
GenericContainer<>(DockerImageName.parse("zookeeper").withTag("3.5.9")),
+ flinkContainersSettings.getZookeeperHostname(),
+ "Zookeeper");
+ }
+
+ private GenericContainer<?> configureContainer(
+ GenericContainer<?> container, String networkAlias, String
logPrefix) {
+ // Set dependent containers
+ for (GenericContainer<?> dependentContainer :
testcontainersSettings.getDependencies()) {
+
dependentContainer.withNetwork(testcontainersSettings.getNetwork());
+ container.dependsOn(dependentContainer);
+ }
+ // Setup network
+ container.withNetwork(testcontainersSettings.getNetwork());
+ container.withNetworkAliases(networkAlias);
+ // Setup logger
+ Logger logger = testcontainersSettings.getLogger();
+ if (logger != null) {
+ container.withLogConsumer(new
Slf4jLogConsumer(logger).withPrefix(logPrefix));
+ }
+ // Add environment variables
+ container.withEnv(testcontainersSettings.getEnvVars());
+ container.withWorkingDirectory(flinkContainersSettings.getFlinkHome());
+ return container;
+ }
+
+ /** Configures and creates {@link FlinkContainers}. */
+ public FlinkContainers configure() {
+ // Create temporary directory for building Flink image
+ final Path imageBuildingTempDir;
+ try {
+ imageBuildingTempDir =
Files.createTempDirectory("flink-image-build");
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to create temporary directory",
e);
+ }
+
+ // Build JobManager
+ final GenericContainer<?> jobManager =
configureJobManagerContainer(imageBuildingTempDir);
+
+ // Build TaskManagers
+ final List<GenericContainer<?>> taskManagers =
+ configureTaskManagerContainers(imageBuildingTempDir);
+
+ // Setup Zookeeper HA
+ GenericContainer<?> zookeeper = null;
+ // Mount HA storage to JobManager
+ if (flinkContainersSettings.isZookeeperHA()) {
+ zookeeper = configureZookeeperContainer();
+ createTempDirAndMountToContainer(
+ "flink-recovery",
flinkContainersSettings.getHaStoragePath(), jobManager);
+ }
+
+ // Mount checkpoint storage to JobManager
+ createTempDirAndMountToContainer(
+ "flink-checkpoint",
flinkContainersSettings.getCheckpointPath(), jobManager);
+
+ return new FlinkContainers(
+ jobManager, taskManagers, zookeeper,
flinkContainersSettings.getFlinkConfig());
+ }
+
+ void createTempDirAndMountToContainer(
+ String tempDirPrefix, String containerPath, GenericContainer<?>
container) {
+ try {
+ Path tempDirPath = Files.createTempDirectory(tempDirPrefix);
+ File file = tempDirPath.toFile();
+ file.setReadable(true, false);
+ file.setWritable(true, false);
+ file.setExecutable(true, false);
+
container.withFileSystemBind(tempDirPath.toAbsolutePath().toString(),
containerPath);
+ } catch (IOException e) {
+ throw new IllegalStateException("Failed to create temporary
recovery directory", e);
+ }
+ }
+}
diff --git
a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/container/TestcontainersSettings.java
b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/container/TestcontainersSettings.java
new file mode 100644
index 00000000000..bb0880cbc58
--- /dev/null
+++
b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/container/TestcontainersSettings.java
@@ -0,0 +1,169 @@
+/*
+ * 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.flink.tests.util.flink.container;
+
+import org.slf4j.Logger;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.Network;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+
+/** The configuration holder for Testcontainers. */
+public class TestcontainersSettings {
+
+ private final Network network;
+ private final Logger logger;
+ private final String baseImage;
+ private final Map<String, String> envVars;
+ private final Collection<GenericContainer<?>> dependencies;
+
+ private TestcontainersSettings(Builder builder) {
+ network = builder.network;
+ baseImage = builder.baseImage;
+ logger = builder.logger;
+ envVars = builder.envVars;
+ dependencies = builder.dependingContainers;
+ }
+
+ /**
+ * A new builder for {@code TestcontainersSettings}.
+ *
+ * @return The builder.
+ */
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ /**
+ * {@code TestcontainersSettings} based on defaults.
+ *
+ * @return The Testcontainers config.
+ */
+ public static TestcontainersSettings defaultSettings() {
+ return builder().build();
+ }
+
+ /** The {@code TestContainersSettings} builder. */
+ public static final class Builder {
+ private Network network = Network.newNetwork();
+ private String baseImage;
+ private Logger logger;
+ private final Map<String, String> envVars = new HashMap<>();
+ private final Collection<GenericContainer<?>> dependingContainers =
new ArrayList<>();
+
+ private Builder() {}
+
+ /**
+ * Sets an environment variable and returns a reference to this
Builder enabling method
+ * chaining.
+ *
+ * @param name The name of the environment variable.
+ * @param value The value of the environment variable.
+ * @return A reference to this Builder.
+ */
+ public Builder environmentVariable(String name, String value) {
+ this.envVars.put(name, value);
+ return this;
+ }
+
+ /**
+ * Lets Flink cluster depending on another container, and bind the
network of Flink cluster
+ * to the dependent one.
+ */
+ public Builder dependsOn(GenericContainer<?> container) {
+ container.withNetwork(this.network);
+ this.dependingContainers.add(container);
+ return this;
+ }
+
+ /**
+ * Sets the {@code network} and returns a reference to this Builder
enabling method
+ * chaining.
+ *
+ * @param network The {@code network} to set.
+ * @return A reference to this Builder.
+ */
+ public Builder network(Network network) {
+ this.network = network;
+ return this;
+ }
+
+ /**
+ * Sets the {@code baseImage} and returns a reference to this Builder
enabling method
+ * chaining.
+ *
+ * @param baseImage The {@code baseImage} to set.
+ * @return A reference to this Builder.
+ */
+ public Builder baseImage(String baseImage) {
+ this.baseImage = baseImage;
+ return this;
+ }
+
+ /**
+ * Sets the {@code baseImage} and returns a reference to this Builder
enabling method
+ * chaining.
+ *
+ * @param logger The {@code logger} to set.
+ * @return A reference to this Builder.
+ */
+ public Builder logger(Logger logger) {
+ this.logger = logger;
+ return this;
+ }
+
+ /**
+ * Returns a {@code TestContainersSettings} built from the parameters
previously set.
+ *
+ * @return A {@code TestContainersSettings} built with parameters of
this {@code
+ * TestContainersSettings.Builder}
+ */
+ public TestcontainersSettings build() {
+ return new TestcontainersSettings(this);
+ }
+ }
+
+ /** @return The network. */
+ public Network getNetwork() {
+ return network;
+ }
+
+ /** @return The logger. */
+ public Logger getLogger() {
+ return logger;
+ }
+
+ /** @return The base image. */
+ public String getBaseImage() {
+ return baseImage;
+ }
+
+ /** @return The environment variables. */
+ public Map<String, String> getEnvVars() {
+ return envVars;
+ }
+
+ /** @return The dependencies (other containers). */
+ public Collection<GenericContainer<?>> getDependencies() {
+ return dependencies;
+ }
+}
diff --git
a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
index 9a1c6dd6f0d..d2f9ab44c2a 100644
---
a/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
+++
b/flink-end-to-end-tests/flink-end-to-end-tests-pulsar/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
@@ -23,7 +23,7 @@ import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.tests.util.TestUtils;
-import org.apache.flink.tests.util.flink.FlinkContainerTestEnvironment;
+import
org.apache.flink.tests.util.flink.container.FlinkContainerTestEnvironment;
/** A Flink Container which would bundles pulsar connector in its classpath. */
public class FlinkContainerWithPulsarEnvironment extends
FlinkContainerTestEnvironment {
diff --git
a/flink-end-to-end-tests/flink-streaming-kinesis-test/src/test/java/org/apache/flink/streaming/kinesis/test/KinesisTableApiITCase.java
b/flink-end-to-end-tests/flink-streaming-kinesis-test/src/test/java/org/apache/flink/streaming/kinesis/test/KinesisTableApiITCase.java
index 246dcfce2e8..7b44088ad55 100644
---
a/flink-end-to-end-tests/flink-streaming-kinesis-test/src/test/java/org/apache/flink/streaming/kinesis/test/KinesisTableApiITCase.java
+++
b/flink-end-to-end-tests/flink-streaming-kinesis-test/src/test/java/org/apache/flink/streaming/kinesis/test/KinesisTableApiITCase.java
@@ -25,6 +25,7 @@ import org.apache.flink.streaming.kinesis.test.model.Order;
import org.apache.flink.tests.util.TestUtils;
import org.apache.flink.tests.util.flink.SQLJobSubmission;
import org.apache.flink.tests.util.flink.container.FlinkContainers;
+import org.apache.flink.tests.util.flink.container.TestcontainersSettings;
import org.apache.flink.util.DockerImageVersions;
import org.apache.flink.util.TestLogger;
@@ -79,17 +80,20 @@ public class KinesisTableApiITCase extends TestLogger {
private KinesisPubsubClient kinesisClient;
- public static final FlinkContainers FLINK =
- FlinkContainers.builder()
- .setEnvironmentVariable("AWS_CBOR_DISABLE", "1")
- .setEnvironmentVariable(
+ public static final TestcontainersSettings TESTCONTAINERS_SETTINGS =
+ TestcontainersSettings.builder()
+ .environmentVariable("AWS_CBOR_DISABLE", "1")
+ .environmentVariable(
"FLINK_ENV_JAVA_OPTS",
"-Dorg.apache.flink.kinesis.shaded.com.amazonaws.sdk.disableCertChecking
-Daws.cborEnabled=false")
- .setNetwork(network)
- .setLogger(LOGGER)
+ .network(network)
+ .logger(LOGGER)
.dependsOn(KINESALITE)
.build();
+ public static final FlinkContainers FLINK =
+
FlinkContainers.builder().withTestcontainersSettings(TESTCONTAINERS_SETTINGS).build();
+
@BeforeClass
public static void setupFlink() throws Exception {
FLINK.start();
