You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ar...@apache.org on 2021/08/13 18:17:01 UTC
[flink] branch master updated (091a201 -> adee561)
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 091a201 [FLINK-14482][rocksdb] Bump FRocksDB version to 6.20.3
new 4fd3323 [hotfix][test/testcontainers] Use Testcontainer BOM to manage versions in the project
new 613a824 [FLINK-19554][connector/testing-framework] Basic abstractions of connector testing framework
new 43066fe [FLINK-19554][testutil/container] Support TM restarting in FlinkContainer and network accessibility from other containers
new b9e1e4d [FLINK-19554][connector/testing-framework] Implementations of TestEnvironment
new 249b281 [FLINK-19554][connector/testing-framework] Connector testing framework utilities based on JUnit 5
new b6bf1aa [FLINK-19554][connector/testing-framework] Base test class for source with fundamental test cases
new adee561 [FLINK-19554][connector/testing-framework] KafkaSource IT and E2E case based on connector testing framework
The 7 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../flink-connector-elasticsearch6/pom.xml | 1 -
.../flink-connector-elasticsearch7/pom.xml | 1 -
flink-connectors/flink-connector-jdbc/pom.xml | 7 +-
flink-connectors/flink-connector-kafka/pom.xml | 13 +-
.../connector/kafka/source/KafkaSourceITCase.java | 313 ++++++++++++---------
.../source/enumerator/KafkaEnumeratorTest.java | 2 +-
.../initializer/OffsetsInitializerTest.java | 2 +-
.../enumerator/subscriber/KafkaSubscriberTest.java | 2 +-
.../reader/KafkaPartitionSplitReaderTest.java | 2 +-
.../kafka/source/reader/KafkaSourceReaderTest.java | 2 +-
.../KafkaMultipleTopicExternalContext.java | 124 ++++++++
.../source/testutils/KafkaPartitionDataWriter.java | 60 ++++
.../testutils/KafkaSingleTopicExternalContext.java | 242 ++++++++++++++++
.../source/{ => testutils}/KafkaSourceTestEnv.java | 2 +-
flink-connectors/flink-connector-kinesis/pom.xml | 1 -
flink-connectors/flink-connector-rabbitmq/pom.xml | 1 -
.../flink-end-to-end-tests-common-kafka/pom.xml | 30 +-
.../flink/tests/util/kafka/KafkaSourceE2ECase.java | 69 +++++
.../flink-end-to-end-tests-common/pom.xml | 6 +-
.../flink/tests/util/flink/FlinkContainer.java | 99 ++++++-
.../util/flink/FlinkContainerTestEnvironment.java | 121 ++++++++
.../flink-streaming-kinesis-test/pom.xml | 17 ++
.../flink/runtime/testutils/CommonTestUtils.java | 41 +++
.../flink-connector-testing/README.md | 44 +++
.../flink-connector-testing}/pom.xml | 59 ++--
.../flink/connectors/test/common/TestResource.java | 53 ++++
.../common/environment/ClusterControllable.java | 33 +++
.../environment/MiniClusterTestEnvironment.java | 135 +++++++++
.../environment/RemoteClusterTestEnvironment.java | 71 +++++
.../test/common/environment/TestEnvironment.java | 35 +++
.../DefaultContainerizedExternalSystem.java | 108 +++++++
.../test/common/external/ExternalContext.java | 75 +++++
.../common/external/SourceSplitDataWriter.java | 34 +++
.../junit/annotations/ExternalContextFactory.java | 43 +++
.../common/junit/annotations/ExternalSystem.java | 39 +++
.../test/common/junit/annotations/TestEnv.java | 40 +++
.../extensions/ConnectorTestingExtension.java | 144 ++++++++++
.../TestCaseInvocationContextProvider.java | 203 +++++++++++++
.../junit/extensions/TestLoggerExtension.java | 83 ++++++
.../common/testsuites/SourceTestSuiteBase.java | 299 ++++++++++++++++++++
.../test/common/utils/TestDataMatchers.java | 246 ++++++++++++++++
.../test/common/utils/IteratorWithCurrentTest.java | 77 +++++
.../test/common/utils/TestDataMatchersTest.java | 166 +++++++++++
flink-test-utils-parent/pom.xml | 1 +
pom.xml | 8 +
.../modules-skipping-deployment.modulelist | 1 +
46 files changed, 2959 insertions(+), 196 deletions(-)
create mode 100644 flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/testutils/KafkaMultipleTopicExternalContext.java
create mode 100644 flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/testutils/KafkaPartitionDataWriter.java
create mode 100644 flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/testutils/KafkaSingleTopicExternalContext.java
rename flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/{ => testutils}/KafkaSourceTestEnv.java (99%)
create mode 100644 flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/KafkaSourceE2ECase.java
create mode 100644 flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/FlinkContainerTestEnvironment.java
create mode 100644 flink-test-utils-parent/flink-connector-testing/README.md
copy {flink-connectors/flink-connector-rabbitmq => flink-test-utils-parent/flink-connector-testing}/pom.xml (62%)
create mode 100644 flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/TestResource.java
create mode 100644 flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/environment/ClusterControllable.java
create mode 100644 flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/environment/MiniClusterTestEnvironment.java
create mode 100644 flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/environment/RemoteClusterTestEnvironment.java
create mode 100644 flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/environment/TestEnvironment.java
create mode 100644 flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/external/DefaultContainerizedExternalSystem.java
create mode 100644 flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/external/ExternalContext.java
create mode 100644 flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/external/SourceSplitDataWriter.java
create mode 100644 flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/junit/annotations/ExternalContextFactory.java
create mode 100644 flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/junit/annotations/ExternalSystem.java
create mode 100644 flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/junit/annotations/TestEnv.java
create mode 100644 flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/junit/extensions/ConnectorTestingExtension.java
create mode 100644 flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/junit/extensions/TestCaseInvocationContextProvider.java
create mode 100644 flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/junit/extensions/TestLoggerExtension.java
create mode 100644 flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/testsuites/SourceTestSuiteBase.java
create mode 100644 flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/utils/TestDataMatchers.java
create mode 100644 flink-test-utils-parent/flink-connector-testing/src/test/java/org/apache/flink/connectors/test/common/utils/IteratorWithCurrentTest.java
create mode 100644 flink-test-utils-parent/flink-connector-testing/src/test/java/org/apache/flink/connectors/test/common/utils/TestDataMatchersTest.java
[flink] 05/07: [FLINK-19554][connector/testing-framework] Connector
testing framework utilities based on JUnit 5
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 249b2818d5e2c6c59b57e722706c5f29ba169712
Author: Qingsheng Ren <re...@gmail.com>
AuthorDate: Thu Aug 12 17:58:43 2021 +0800
[FLINK-19554][connector/testing-framework] Connector testing framework utilities based on JUnit 5
---
.../DefaultContainerizedExternalSystem.java | 108 +++++++++++
.../junit/annotations/ExternalContextFactory.java | 43 +++++
.../common/junit/annotations/ExternalSystem.java | 39 ++++
.../test/common/junit/annotations/TestEnv.java | 40 ++++
.../extensions/ConnectorTestingExtension.java | 144 +++++++++++++++
.../TestCaseInvocationContextProvider.java | 203 +++++++++++++++++++++
.../junit/extensions/TestLoggerExtension.java | 83 +++++++++
7 files changed, 660 insertions(+)
diff --git a/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/external/DefaultContainerizedExternalSystem.java b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/external/DefaultContainerizedExternalSystem.java
new file mode 100644
index 0000000..ef71c8e
--- /dev/null
+++ b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/external/DefaultContainerizedExternalSystem.java
@@ -0,0 +1,108 @@
+/*
+ * 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.connectors.test.common.external;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.connectors.test.common.TestResource;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.GenericContainer;
+
+/**
+ * Default implementation of external system based on container.
+ *
+ * @param <C> Type of underlying container
+ */
+@Experimental
+public class DefaultContainerizedExternalSystem<C extends GenericContainer<C>>
+ implements TestResource {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(DefaultContainerizedExternalSystem.class);
+
+ private final C container;
+
+ /**
+ * Get a builder for {@link DefaultContainerizedExternalSystem}.
+ *
+ * @param <C> Type of underlying container
+ * @return An instance of builder
+ */
+ public static <C extends GenericContainer<C>> Builder<C> builder() {
+ return new Builder<>();
+ }
+
+ private DefaultContainerizedExternalSystem(C container) {
+ this.container = container;
+ }
+
+ @Override
+ public void startUp() throws Exception {
+ if (container.isRunning()) {
+ return;
+ }
+ container.start();
+ }
+
+ @Override
+ public void tearDown() throws Exception {
+ if (!container.isRunning()) {
+ return;
+ }
+ container.stop();
+ }
+
+ public C getContainer() {
+ return this.container;
+ }
+
+ /**
+ * Builder for {@link DefaultContainerizedExternalSystem}.
+ *
+ * @param <C> Type of underlying container
+ */
+ public static class Builder<C extends GenericContainer<C>> {
+
+ private C container;
+ private GenericContainer<?> flinkContainer;
+
+ public <T extends GenericContainer<T>> Builder<T> fromContainer(T container) {
+ @SuppressWarnings("unchecked")
+ Builder<T> self = (Builder<T>) this;
+ self.container = container;
+ return self;
+ }
+
+ public Builder<C> bindWithFlinkContainer(GenericContainer<?> flinkContainer) {
+ this.flinkContainer = flinkContainer;
+ container.dependsOn(flinkContainer).withNetwork(flinkContainer.getNetwork());
+ return this;
+ }
+
+ public DefaultContainerizedExternalSystem<C> build() {
+ if (flinkContainer == null) {
+ LOG.warn(
+ "External system container is not bound with Flink container. "
+ + "This might lead to network isolation between external system and Flink");
+ }
+ return new DefaultContainerizedExternalSystem<>(container);
+ }
+ }
+}
diff --git a/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/junit/annotations/ExternalContextFactory.java b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/junit/annotations/ExternalContextFactory.java
new file mode 100644
index 0000000..0b34555
--- /dev/null
+++ b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/junit/annotations/ExternalContextFactory.java
@@ -0,0 +1,43 @@
+/*
+ * 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.connectors.test.common.junit.annotations;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.connectors.test.common.external.ExternalContext;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Marks the field in test class defining a {@link ExternalContext.Factory} for constructing {@link
+ * ExternalContext} before invocation of each test case.
+ *
+ * <p>Multiple fields can be annotated as external context factory, and these external contexts will
+ * be provided as different parameters of test cases.
+ *
+ * <p>The lifecycle of a {@link ExternalContext} will be PER-CASE, which means an instance of {@link
+ * ExternalContext} will be constructed before invocation of each test case, and closed right after
+ * the execution of the case for isolation between test cases.
+ */
+@Target(ElementType.FIELD)
+@Retention(RetentionPolicy.RUNTIME)
+@Experimental
+public @interface ExternalContextFactory {}
diff --git a/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/junit/annotations/ExternalSystem.java b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/junit/annotations/ExternalSystem.java
new file mode 100644
index 0000000..d6f9569
--- /dev/null
+++ b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/junit/annotations/ExternalSystem.java
@@ -0,0 +1,39 @@
+/*
+ * 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.connectors.test.common.junit.annotations;
+
+import org.apache.flink.annotation.Experimental;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Marks the field in test class defining external system.
+ *
+ * <p>Only one field can be annotated as external system in test class.
+ *
+ * <p>The lifecycle of external system will be PER-CLASS for performance, because launching and
+ * tearing down external system could be relatively a heavy operation.
+ */
+@Target(ElementType.FIELD)
+@Retention(RetentionPolicy.RUNTIME)
+@Experimental
+public @interface ExternalSystem {}
diff --git a/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/junit/annotations/TestEnv.java b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/junit/annotations/TestEnv.java
new file mode 100644
index 0000000..574c87a
--- /dev/null
+++ b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/junit/annotations/TestEnv.java
@@ -0,0 +1,40 @@
+/*
+ * 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.connectors.test.common.junit.annotations;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.connectors.test.common.environment.TestEnvironment;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Marks the field in test class defining {@link TestEnvironment}.
+ *
+ * <p>Only one field can be annotated as test environment in a test class.
+ *
+ * <p>The lifecycle of {@link TestEnvironment} will be PER-CLASS for performance, because launching
+ * and tearing down Flink cluster could be relatively a heavy operation.
+ */
+@Target(ElementType.FIELD)
+@Retention(RetentionPolicy.RUNTIME)
+@Experimental
+public @interface TestEnv {}
diff --git a/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/junit/extensions/ConnectorTestingExtension.java b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/junit/extensions/ConnectorTestingExtension.java
new file mode 100644
index 0000000..6652de7
--- /dev/null
+++ b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/junit/extensions/ConnectorTestingExtension.java
@@ -0,0 +1,144 @@
+/*
+ * 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.connectors.test.common.junit.extensions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connectors.test.common.TestResource;
+import org.apache.flink.connectors.test.common.environment.TestEnvironment;
+import org.apache.flink.connectors.test.common.external.ExternalContext;
+import org.apache.flink.connectors.test.common.junit.annotations.ExternalContextFactory;
+import org.apache.flink.connectors.test.common.junit.annotations.ExternalSystem;
+import org.apache.flink.connectors.test.common.junit.annotations.TestEnv;
+
+import org.junit.jupiter.api.extension.AfterAllCallback;
+import org.junit.jupiter.api.extension.BeforeAllCallback;
+import org.junit.jupiter.api.extension.Extension;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.junit.platform.commons.support.AnnotationSupport;
+
+import java.lang.annotation.Annotation;
+import java.util.Collection;
+import java.util.List;
+
+/**
+ * A JUnit 5 {@link Extension} for supporting running of connector testing framework.
+ *
+ * <p>This extension is responsible for searching test resources annotated by {@link TestEnv},
+ * {@link ExternalSystem} and {@link ExternalContextFactory}, storing them into storage provided by
+ * JUnit, and manage lifecycle of these resources.
+ *
+ * <p>The extension uses {@link ExtensionContext.Store} for handing over test resources to {@link
+ * TestCaseInvocationContextProvider}, which will inject these resources into test cases as
+ * parameters.
+ *
+ * <p>The order of initialization is promised to be:
+ *
+ * <ol>
+ * <li>Test environment annotated by {@link TestEnv}, before all test cases in this extension
+ * <li>External system annotated by {@link ExternalSystem}, before all test cases in this
+ * extension
+ * <li>External contexts annotated by {@link ExternalContextFactory}, before each test case in
+ * {@link TestCaseInvocationContextProvider}
+ * </ol>
+ */
+@Internal
+public class ConnectorTestingExtension implements BeforeAllCallback, AfterAllCallback {
+
+ public static final ExtensionContext.Namespace TEST_RESOURCE_NAMESPACE =
+ ExtensionContext.Namespace.create("testResourceNamespace");
+ public static final String TEST_ENV_STORE_KEY = "testEnvironment";
+ public static final String EXTERNAL_SYSTEM_STORE_KEY = "externalSystem";
+ public static final String EXTERNAL_CONTEXT_FACTORIES_STORE_KEY = "externalContext";
+
+ private TestEnvironment testEnvironment;
+ private TestResource externalSystem;
+
+ @SuppressWarnings("rawtypes")
+ @Override
+ public void beforeAll(ExtensionContext context) throws Exception {
+
+ // Setup test environment and store
+ final List<TestEnvironment> testEnvironments =
+ AnnotationSupport.findAnnotatedFieldValues(
+ context.getRequiredTestInstance(), TestEnv.class, TestEnvironment.class);
+ checkExactlyOneAnnotatedField(testEnvironments, TestEnv.class);
+ testEnvironment = testEnvironments.get(0);
+ testEnvironment.startUp();
+ context.getStore(TEST_RESOURCE_NAMESPACE).put(TEST_ENV_STORE_KEY, testEnvironment);
+
+ // Setup external system and store
+ final List<TestResource> externalSystems =
+ AnnotationSupport.findAnnotatedFieldValues(
+ context.getRequiredTestInstance(),
+ ExternalSystem.class,
+ TestResource.class);
+ checkExactlyOneAnnotatedField(externalSystems, ExternalSystem.class);
+ externalSystem = externalSystems.get(0);
+ externalSystem.startUp();
+ context.getStore(TEST_RESOURCE_NAMESPACE).put(EXTERNAL_SYSTEM_STORE_KEY, externalSystem);
+
+ // Search external context factories
+ final List<ExternalContext.Factory> externalContextFactories =
+ AnnotationSupport.findAnnotatedFieldValues(
+ context.getRequiredTestInstance(),
+ ExternalContextFactory.class,
+ ExternalContext.Factory.class);
+ checkAtLeastOneAnnotationField(externalContextFactories, ExternalContextFactory.class);
+ context.getStore(TEST_RESOURCE_NAMESPACE)
+ .put(EXTERNAL_CONTEXT_FACTORIES_STORE_KEY, externalContextFactories);
+ }
+
+ @Override
+ public void afterAll(ExtensionContext context) throws Exception {
+ // Tear down test environment
+ testEnvironment.tearDown();
+
+ // Tear down external system
+ externalSystem.tearDown();
+
+ // Clear store
+ context.getStore(TEST_RESOURCE_NAMESPACE).remove(TEST_ENV_STORE_KEY);
+ context.getStore(TEST_RESOURCE_NAMESPACE).remove(EXTERNAL_SYSTEM_STORE_KEY);
+ context.getStore(TEST_RESOURCE_NAMESPACE).remove(EXTERNAL_CONTEXT_FACTORIES_STORE_KEY);
+ }
+
+ private void checkExactlyOneAnnotatedField(
+ Collection<?> fields, Class<? extends Annotation> annotation) {
+ if (fields.size() > 1) {
+ throw new IllegalStateException(
+ String.format(
+ "Multiple fields are annotated with '@%s'",
+ annotation.getSimpleName()));
+ }
+ if (fields.isEmpty()) {
+ throw new IllegalStateException(
+ String.format(
+ "No fields are annotated with '@%s'", annotation.getSimpleName()));
+ }
+ }
+
+ private void checkAtLeastOneAnnotationField(
+ Collection<?> fields, Class<? extends Annotation> annotation) {
+ if (fields.isEmpty()) {
+ throw new IllegalStateException(
+ String.format(
+ "No fields are annotated with '@%s'", annotation.getSimpleName()));
+ }
+ }
+}
diff --git a/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/junit/extensions/TestCaseInvocationContextProvider.java b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/junit/extensions/TestCaseInvocationContextProvider.java
new file mode 100644
index 0000000..b28dc94
--- /dev/null
+++ b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/junit/extensions/TestCaseInvocationContextProvider.java
@@ -0,0 +1,203 @@
+/*
+ * 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.connectors.test.common.junit.extensions;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connectors.test.common.environment.ClusterControllable;
+import org.apache.flink.connectors.test.common.environment.TestEnvironment;
+import org.apache.flink.connectors.test.common.external.ExternalContext;
+
+import org.junit.jupiter.api.extension.AfterTestExecutionCallback;
+import org.junit.jupiter.api.extension.Extension;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.junit.jupiter.api.extension.ParameterContext;
+import org.junit.jupiter.api.extension.ParameterResolutionException;
+import org.junit.jupiter.api.extension.ParameterResolver;
+import org.junit.jupiter.api.extension.TestTemplateInvocationContext;
+import org.junit.jupiter.api.extension.TestTemplateInvocationContextProvider;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Stream;
+
+import static org.apache.flink.connectors.test.common.junit.extensions.ConnectorTestingExtension.EXTERNAL_CONTEXT_FACTORIES_STORE_KEY;
+import static org.apache.flink.connectors.test.common.junit.extensions.ConnectorTestingExtension.TEST_ENV_STORE_KEY;
+import static org.apache.flink.connectors.test.common.junit.extensions.ConnectorTestingExtension.TEST_RESOURCE_NAMESPACE;
+
+/**
+ * A helper class for injecting test resources into test case as parameters.
+ *
+ * <p>This provider will resolve {@link TestEnvironment} and {@link ExternalContext.Factory} from
+ * the storage in JUnit's {@link ExtensionContext}, inject them into test method, and register a
+ * {@link AfterTestExecutionCallback} for closing the external context after the execution of test
+ * case.
+ */
+@Internal
+public class TestCaseInvocationContextProvider implements TestTemplateInvocationContextProvider {
+
+ @Override
+ public boolean supportsTestTemplate(ExtensionContext context) {
+ // Only support test cases with TestEnvironment and ExternalContext as parameter
+ return Arrays.stream(context.getRequiredTestMethod().getParameterTypes())
+ .anyMatch(
+ (type) ->
+ TestEnvironment.class.isAssignableFrom(type)
+ || ExternalContext.class.isAssignableFrom(type));
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public Stream<TestTemplateInvocationContext> provideTestTemplateInvocationContexts(
+ ExtensionContext context) {
+
+ // Fetch test environment from store
+ TestEnvironment testEnv =
+ context.getStore(TEST_RESOURCE_NAMESPACE)
+ .get(TEST_ENV_STORE_KEY, TestEnvironment.class);
+
+ // Fetch external context factories from store
+ List<ExternalContext.Factory<?>> externalContextFactories =
+ (List<ExternalContext.Factory<?>>)
+ context.getStore(TEST_RESOURCE_NAMESPACE)
+ .get(EXTERNAL_CONTEXT_FACTORIES_STORE_KEY);
+
+ // Create a invocation context for each external context factory
+ return externalContextFactories.stream()
+ .map(
+ factory ->
+ new TestResourceProvidingInvocationContext(
+ testEnv, factory.createExternalContext()));
+ }
+
+ /**
+ * Invocation context for injecting {@link TestEnvironment} and {@link ExtensionContext} into
+ * test cases as method parameters.
+ */
+ static class TestResourceProvidingInvocationContext implements TestTemplateInvocationContext {
+
+ private final TestEnvironment testEnvironment;
+ private final ExternalContext<?> externalContext;
+
+ public TestResourceProvidingInvocationContext(
+ TestEnvironment testEnvironment, ExternalContext<?> externalContext) {
+ this.testEnvironment = testEnvironment;
+ this.externalContext = externalContext;
+ }
+
+ @Override
+ public String getDisplayName(int invocationIndex) {
+ return String.format(
+ "TestEnvironment: [%s], ExternalContext: [%s]",
+ testEnvironment, externalContext);
+ }
+
+ @Override
+ public List<Extension> getAdditionalExtensions() {
+ return Arrays.asList(
+ // Extension for injecting parameters
+ new TestEnvironmentResolver(testEnvironment),
+ new ExternalContextProvider(externalContext),
+ new ClusterControllableProvider(testEnvironment),
+ // Extension for closing external context
+ (AfterTestExecutionCallback) ignore -> externalContext.close());
+ }
+ }
+
+ private static class TestEnvironmentResolver implements ParameterResolver {
+
+ private final TestEnvironment testEnvironment;
+
+ private TestEnvironmentResolver(TestEnvironment testEnvironment) {
+ this.testEnvironment = testEnvironment;
+ }
+
+ @Override
+ public boolean supportsParameter(
+ ParameterContext parameterContext, ExtensionContext extensionContext)
+ throws ParameterResolutionException {
+ return isAssignableFromParameterType(
+ TestEnvironment.class, parameterContext.getParameter().getType());
+ }
+
+ @Override
+ public Object resolveParameter(
+ ParameterContext parameterContext, ExtensionContext extensionContext)
+ throws ParameterResolutionException {
+ return this.testEnvironment;
+ }
+ }
+
+ private static class ExternalContextProvider implements ParameterResolver {
+
+ private final ExternalContext<?> externalContext;
+
+ private ExternalContextProvider(ExternalContext<?> externalContext) {
+ this.externalContext = externalContext;
+ }
+
+ @Override
+ public boolean supportsParameter(
+ ParameterContext parameterContext, ExtensionContext extensionContext)
+ throws ParameterResolutionException {
+ return isAssignableFromParameterType(
+ ExternalContext.class, parameterContext.getParameter().getType());
+ }
+
+ @Override
+ public Object resolveParameter(
+ ParameterContext parameterContext, ExtensionContext extensionContext)
+ throws ParameterResolutionException {
+ return this.externalContext;
+ }
+ }
+
+ private static class ClusterControllableProvider implements ParameterResolver {
+
+ private final TestEnvironment testEnvironment;
+
+ private ClusterControllableProvider(TestEnvironment testEnvironment) {
+ this.testEnvironment = testEnvironment;
+ }
+
+ @Override
+ public boolean supportsParameter(
+ ParameterContext parameterContext, ExtensionContext extensionContext)
+ throws ParameterResolutionException {
+ return isAssignableFromParameterType(
+ ClusterControllable.class, parameterContext.getParameter().getType())
+ && isTestEnvironmentControllable(this.testEnvironment);
+ }
+
+ @Override
+ public Object resolveParameter(
+ ParameterContext parameterContext, ExtensionContext extensionContext)
+ throws ParameterResolutionException {
+ return testEnvironment;
+ }
+
+ private boolean isTestEnvironmentControllable(TestEnvironment testEnvironment) {
+ return ClusterControllable.class.isAssignableFrom(testEnvironment.getClass());
+ }
+ }
+
+ private static boolean isAssignableFromParameterType(
+ Class<?> requiredType, Class<?> parameterType) {
+ return requiredType.isAssignableFrom(parameterType);
+ }
+}
diff --git a/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/junit/extensions/TestLoggerExtension.java b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/junit/extensions/TestLoggerExtension.java
new file mode 100644
index 0000000..aaac2de
--- /dev/null
+++ b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/junit/extensions/TestLoggerExtension.java
@@ -0,0 +1,83 @@
+/*
+ * 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.connectors.test.common.junit.extensions;
+
+import org.apache.flink.annotation.Internal;
+
+import org.junit.jupiter.api.extension.BeforeEachCallback;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.junit.jupiter.api.extension.TestWatcher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.PrintWriter;
+import java.io.StringWriter;
+
+/** A JUnit-5-style test logger. */
+@Internal
+public class TestLoggerExtension implements TestWatcher, BeforeEachCallback {
+ private static final Logger LOG = LoggerFactory.getLogger(TestLoggerExtension.class);
+
+ @Override
+ public void beforeEach(ExtensionContext context) {
+ LOG.info(
+ "\n================================================================================"
+ + "\n🧪 Test {}.{} is running."
+ + "\n--------------------------------------------------------------------------------",
+ context.getRequiredTestClass().getCanonicalName(),
+ context.getRequiredTestMethod().getName());
+ }
+
+ @Override
+ public void testSuccessful(ExtensionContext context) {
+ LOG.info(
+ "\n--------------------------------------------------------------------------------"
+ + "\nTest {}.{} successfully run."
+ + "\n================================================================================",
+ context.getRequiredTestClass().getCanonicalName(),
+ context.getRequiredTestMethod().getName());
+ }
+
+ @Override
+ public void testFailed(ExtensionContext context, Throwable cause) {
+ LOG.error(
+ "\n--------------------------------------------------------------------------------"
+ + "\nTest {}.{} failed with:\n{}"
+ + "\n================================================================================",
+ context.getRequiredTestClass().getCanonicalName(),
+ context.getRequiredTestMethod().getName(),
+ exceptionToString(cause));
+ }
+
+ private static String exceptionToString(Throwable t) {
+ if (t == null) {
+ return "(null)";
+ }
+
+ try {
+ StringWriter stm = new StringWriter();
+ PrintWriter wrt = new PrintWriter(stm);
+ t.printStackTrace(wrt);
+ wrt.close();
+ return stm.toString();
+ } catch (Throwable ignored) {
+ return t.getClass().getName() + " (error while printing stack trace)";
+ }
+ }
+}
[flink] 04/07: [FLINK-19554][connector/testing-framework]
Implementations of TestEnvironment
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit b9e1e4db20beca757e958a1f7a1797e59e5cc8cd
Author: Qingsheng Ren <re...@gmail.com>
AuthorDate: Thu Aug 12 17:56:55 2021 +0800
[FLINK-19554][connector/testing-framework] Implementations of TestEnvironment
---
.../flink-end-to-end-tests-common/pom.xml | 5 +
.../util/flink/FlinkContainerTestEnvironment.java | 121 ++++++++++++++++++
.../flink-streaming-kinesis-test/pom.xml | 17 +++
.../flink/runtime/testutils/CommonTestUtils.java | 41 +++++++
.../environment/MiniClusterTestEnvironment.java | 135 +++++++++++++++++++++
.../environment/RemoteClusterTestEnvironment.java | 71 +++++++++++
6 files changed, 390 insertions(+)
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-common/pom.xml b/flink-end-to-end-tests/flink-end-to-end-tests-common/pom.xml
index 702fe8d..3ed5d81 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-common/pom.xml
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-common/pom.xml
@@ -61,6 +61,11 @@ under the License.
<scope>compile</scope>
</dependency>
<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-testing_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers</artifactId>
</dependency>
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/FlinkContainerTestEnvironment.java
new file mode 100644
index 0000000..fc1b5d9
--- /dev/null
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/FlinkContainerTestEnvironment.java
@@ -0,0 +1,121 @@
+/*
+ * 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;
+
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.time.Deadline;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connectors.test.common.environment.ClusterControllable;
+import org.apache.flink.connectors.test.common.environment.TestEnvironment;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.runtime.testutils.CommonTestUtils;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import java.time.Duration;
+import java.util.Arrays;
+
+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.TaskManagerOptions.NUM_TASK_SLOTS;
+
+/** Test environment running job on {@link FlinkContainer}. */
+public class FlinkContainerTestEnvironment implements TestEnvironment, ClusterControllable {
+
+ private final FlinkContainer flinkContainer;
+ private final String[] jarPath;
+
+ public FlinkContainerTestEnvironment(
+ int numTaskManagers, int numSlotsPerTaskManager, String... jarPath) {
+
+ Configuration flinkConfiguration = new Configuration();
+ flinkConfiguration.set(HEARTBEAT_INTERVAL, 1000L);
+ flinkConfiguration.set(HEARTBEAT_TIMEOUT, 5000L);
+ flinkConfiguration.set(SLOT_REQUEST_TIMEOUT, 10000L);
+ flinkConfiguration.set(NUM_TASK_SLOTS, numSlotsPerTaskManager);
+
+ this.flinkContainer =
+ FlinkContainer.builder()
+ .numTaskManagers(numTaskManagers)
+ .withFlinkConfiguration(flinkConfiguration)
+ .build();
+ this.jarPath = jarPath;
+ }
+
+ @Override
+ public void startUp() {
+ if (!flinkContainer.isRunning()) {
+ this.flinkContainer.start();
+ }
+ }
+
+ @Override
+ public void tearDown() {
+ if (flinkContainer.isRunning()) {
+ this.flinkContainer.stop();
+ }
+ }
+
+ @Override
+ public StreamExecutionEnvironment createExecutionEnvironment() {
+ return StreamExecutionEnvironment.createRemoteEnvironment(
+ this.flinkContainer.getHost(),
+ this.flinkContainer.getMappedPort(FlinkContainer.JOB_MANAGER_REST_PORT),
+ this.jarPath);
+ }
+
+ @Override
+ public void triggerJobManagerFailover(JobClient jobClient, Runnable afterFailAction) {}
+
+ @Override
+ public void triggerTaskManagerFailover(JobClient jobClient, Runnable afterFailAction)
+ throws Exception {
+ flinkContainer.restartTaskManager(
+ () -> {
+ try {
+ CommonTestUtils.waitForJobStatus(
+ jobClient,
+ Arrays.asList(
+ JobStatus.FAILING, JobStatus.FAILED, JobStatus.RESTARTING),
+ Deadline.fromNow(Duration.ofSeconds(30)));
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Error waiting for job entering failure status", e);
+ }
+ afterFailAction.run();
+ });
+ }
+
+ @Override
+ public void isolateNetwork(JobClient jobClient, Runnable afterFailAction) {}
+
+ @Override
+ public String toString() {
+ return "FlinkContainer";
+ }
+
+ /**
+ * Get instance of Flink containers for cluster controlling.
+ *
+ * @return Flink cluster on Testcontainers
+ */
+ public FlinkContainer getFlinkContainer() {
+ return this.flinkContainer;
+ }
+}
diff --git a/flink-end-to-end-tests/flink-streaming-kinesis-test/pom.xml b/flink-end-to-end-tests/flink-streaming-kinesis-test/pom.xml
index edd6f61..534879e 100644
--- a/flink-end-to-end-tests/flink-streaming-kinesis-test/pom.xml
+++ b/flink-end-to-end-tests/flink-streaming-kinesis-test/pom.xml
@@ -145,6 +145,23 @@ under the License.
</systemPropertyVariables>
</configuration>
</plugin>
+
+ <!-- Skip dependency convergence check because of guava version -->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-enforcer-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>dependency-convergence</id>
+ <goals>
+ <goal>enforce</goal>
+ </goals>
+ <configuration>
+ <skip>true</skip>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
</plugins>
</build>
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/CommonTestUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/CommonTestUtils.java
index 7933dbe..ea1d55b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/CommonTestUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/CommonTestUtils.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.testutils;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.api.common.time.Deadline;
+import org.apache.flink.core.execution.JobClient;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.executiongraph.AccessExecutionGraph;
import org.apache.flink.runtime.minicluster.MiniCluster;
@@ -39,6 +40,7 @@ import java.lang.management.RuntimeMXBean;
import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.Arrays;
+import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -200,6 +202,45 @@ public class CommonTestUtils {
waitUntilCondition(() -> jobStatusSupplier.get() != JobStatus.INITIALIZING, timeout, 20L);
}
+ public static void waitForJobStatus(
+ JobClient client, List<JobStatus> expectedStatus, Deadline deadline) throws Exception {
+ waitUntilCondition(
+ () -> {
+ final JobStatus currentStatus = client.getJobStatus().get();
+
+ // Entered an expected status
+ if (expectedStatus.contains(currentStatus)) {
+ return true;
+ }
+
+ // Entered a terminal status but not expected
+ if (currentStatus.isTerminalState()) {
+ try {
+ // Exception will be exposed here if job failed
+ client.getJobExecutionResult().get();
+ } catch (Exception e) {
+ throw new IllegalStateException(
+ String.format(
+ "Job has entered %s state, but expecting %s",
+ currentStatus, expectedStatus),
+ e);
+ }
+ throw new IllegalStateException(
+ String.format(
+ "Job has entered a terminal state %s, but expecting %s",
+ currentStatus, expectedStatus));
+ }
+
+ // Continue waiting for expected status
+ return false;
+ },
+ deadline);
+ }
+
+ public static void terminateJob(JobClient client, Duration timeout) throws Exception {
+ client.cancel().get(timeout.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
/** Utility class to read the output of a process stream and forward it into a StringWriter. */
public static class PipeForwarder extends Thread {
diff --git a/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/environment/MiniClusterTestEnvironment.java b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/environment/MiniClusterTestEnvironment.java
new file mode 100644
index 0000000..ab1c4fc
--- /dev/null
+++ b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/environment/MiniClusterTestEnvironment.java
@@ -0,0 +1,135 @@
+/*
+ * 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.connectors.test.common.environment;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.time.Deadline;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.runtime.highavailability.nonha.embedded.HaLeadershipControl;
+import org.apache.flink.runtime.minicluster.RpcServiceSharing;
+import org.apache.flink.runtime.testutils.CommonTestUtils;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+
+/** Test environment for running jobs on Flink mini-cluster. */
+@Experimental
+public class MiniClusterTestEnvironment implements TestEnvironment, ClusterControllable {
+
+ private final MiniClusterWithClientResource miniCluster;
+
+ private int latestTMIndex = 0;
+
+ private static final Logger LOG = LoggerFactory.getLogger(MiniClusterTestEnvironment.class);
+
+ private boolean isStarted = false;
+
+ public MiniClusterTestEnvironment() {
+ this.miniCluster =
+ new MiniClusterWithClientResource(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(6)
+ .setRpcServiceSharing(RpcServiceSharing.DEDICATED)
+ .withHaLeadershipControl()
+ .build());
+ }
+
+ @Override
+ public StreamExecutionEnvironment createExecutionEnvironment() {
+ return StreamExecutionEnvironment.getExecutionEnvironment();
+ }
+
+ @Override
+ public void triggerJobManagerFailover(JobClient jobClient, Runnable afterFailAction)
+ throws ExecutionException, InterruptedException {
+ final Optional<HaLeadershipControl> controlOptional =
+ miniCluster.getMiniCluster().getHaLeadershipControl();
+ if (!controlOptional.isPresent()) {
+ throw new UnsupportedOperationException(
+ "This MiniCluster does not support JobManager HA");
+ }
+ final HaLeadershipControl haLeadershipControl = controlOptional.get();
+ haLeadershipControl.revokeJobMasterLeadership(jobClient.getJobID()).get();
+ afterFailAction.run();
+ haLeadershipControl.grantJobMasterLeadership(jobClient.getJobID()).get();
+ }
+
+ @Override
+ public void triggerTaskManagerFailover(JobClient jobClient, Runnable afterFailAction)
+ throws Exception {
+ terminateTaskManager();
+ CommonTestUtils.waitForJobStatus(
+ jobClient,
+ Arrays.asList(JobStatus.FAILING, JobStatus.FAILED, JobStatus.RESTARTING),
+ Deadline.fromNow(Duration.ofSeconds(30)));
+ afterFailAction.run();
+ startTaskManager();
+ }
+
+ @Override
+ public void isolateNetwork(JobClient jobClient, Runnable afterFailAction) {
+ throw new UnsupportedOperationException("Cannot isolate network in a MiniCluster");
+ }
+
+ @Override
+ public void startUp() throws Exception {
+ if (isStarted) {
+ return;
+ }
+ this.miniCluster.before();
+ LOG.debug("MiniCluster is running");
+ isStarted = true;
+ }
+
+ @Override
+ public void tearDown() {
+ if (!isStarted) {
+ return;
+ }
+ isStarted = false;
+ this.miniCluster.after();
+ LOG.debug("MiniCluster has been tear down");
+ }
+
+ private void terminateTaskManager() throws Exception {
+ miniCluster.getMiniCluster().terminateTaskManager(latestTMIndex).get();
+ LOG.debug("TaskManager {} has been terminated.", latestTMIndex);
+ }
+
+ private void startTaskManager() throws Exception {
+ miniCluster.getMiniCluster().startTaskManager();
+ latestTMIndex++;
+ LOG.debug("New TaskManager {} has been launched.", latestTMIndex);
+ }
+
+ @Override
+ public String toString() {
+ return "MiniCluster";
+ }
+}
diff --git a/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/environment/RemoteClusterTestEnvironment.java b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/environment/RemoteClusterTestEnvironment.java
new file mode 100644
index 0000000..337ff2d
--- /dev/null
+++ b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/environment/RemoteClusterTestEnvironment.java
@@ -0,0 +1,71 @@
+/*
+ * 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.connectors.test.common.environment;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+/** Test environment for running test on a remote Flink cluster. */
+@Experimental
+public class RemoteClusterTestEnvironment implements TestEnvironment {
+
+ private final String host;
+ private final int port;
+ private final String[] jarPath;
+ private final Configuration config;
+
+ /**
+ * Construct a test environment for a remote Flink cluster.
+ *
+ * @param host Hostname of the remote JobManager
+ * @param port REST port of the remote JobManager
+ * @param jarPath Path of JARs to be shipped to Flink cluster
+ */
+ public RemoteClusterTestEnvironment(String host, int port, String... jarPath) {
+ this(host, port, new Configuration(), jarPath);
+ }
+
+ /**
+ * Construct a test environment for a remote Flink cluster with configurations.
+ *
+ * @param host Hostname of the remote JobManager
+ * @param port REST port of the remote JobManager
+ * @param config Configurations of the test environment
+ * @param jarPath Path of JARs to be shipped to Flink cluster
+ */
+ public RemoteClusterTestEnvironment(
+ String host, int port, Configuration config, String... jarPath) {
+ this.host = host;
+ this.port = port;
+ this.config = config;
+ this.jarPath = jarPath;
+ }
+
+ @Override
+ public StreamExecutionEnvironment createExecutionEnvironment() {
+ return StreamExecutionEnvironment.createRemoteEnvironment(host, port, jarPath);
+ }
+
+ @Override
+ public void startUp() {}
+
+ @Override
+ public void tearDown() {}
+}
[flink] 06/07: [FLINK-19554][connector/testing-framework] Base test
class for source with fundamental test cases
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit b6bf1aad5f2ebeb318b98162271eded355685ac8
Author: Qingsheng Ren <re...@gmail.com>
AuthorDate: Thu Aug 12 17:59:30 2021 +0800
[FLINK-19554][connector/testing-framework] Base test class for source with fundamental test cases
---
.../common/testsuites/SourceTestSuiteBase.java | 299 +++++++++++++++++++++
.../test/common/utils/TestDataMatchers.java | 246 +++++++++++++++++
.../test/common/utils/IteratorWithCurrentTest.java | 77 ++++++
.../test/common/utils/TestDataMatchersTest.java | 166 ++++++++++++
4 files changed, 788 insertions(+)
diff --git a/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/testsuites/SourceTestSuiteBase.java b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/testsuites/SourceTestSuiteBase.java
new file mode 100644
index 0000000..4b90272
--- /dev/null
+++ b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/testsuites/SourceTestSuiteBase.java
@@ -0,0 +1,299 @@
+/*
+ * 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.connectors.test.common.testsuites;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.time.Deadline;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.connectors.test.common.environment.ClusterControllable;
+import org.apache.flink.connectors.test.common.environment.TestEnvironment;
+import org.apache.flink.connectors.test.common.external.ExternalContext;
+import org.apache.flink.connectors.test.common.external.SourceSplitDataWriter;
+import org.apache.flink.connectors.test.common.junit.extensions.ConnectorTestingExtension;
+import org.apache.flink.connectors.test.common.junit.extensions.TestCaseInvocationContextProvider;
+import org.apache.flink.connectors.test.common.junit.extensions.TestLoggerExtension;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.runtime.testutils.CommonTestUtils;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.operators.collect.CollectResultIterator;
+import org.apache.flink.streaming.api.operators.collect.CollectSinkOperator;
+import org.apache.flink.streaming.api.operators.collect.CollectSinkOperatorFactory;
+import org.apache.flink.streaming.api.operators.collect.CollectStreamSink;
+import org.apache.flink.util.CloseableIterator;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
+
+import static org.apache.flink.connectors.test.common.utils.TestDataMatchers.matchesMultipleSplitTestData;
+import static org.apache.flink.connectors.test.common.utils.TestDataMatchers.matchesSplitTestData;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+/**
+ * Base class for all test suites.
+ *
+ * <p>All cases should have well-descriptive JavaDoc, including:
+ *
+ * <ul>
+ * <li>What's the purpose of this case
+ * <li>Simple description of how this case works
+ * <li>Condition to fulfill in order to pass this case
+ * <li>Requirement of running this case
+ * </ul>
+ */
+@ExtendWith({
+ ConnectorTestingExtension.class,
+ TestLoggerExtension.class,
+ TestCaseInvocationContextProvider.class
+})
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+@Experimental
+public abstract class SourceTestSuiteBase<T> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SourceTestSuiteBase.class);
+
+ // ----------------------------- Basic test cases ---------------------------------
+
+ /**
+ * Test connector source with only one split in the external system.
+ *
+ * <p>This test will create one split in the external system, write test data into it, and
+ * consume back via a Flink job with 1 parallelism.
+ *
+ * <p>The number and order of records consumed by Flink need to be identical to the test data
+ * written to the external system in order to pass this test.
+ *
+ * <p>A bounded source is required for this test.
+ */
+ @TestTemplate
+ @DisplayName("Test source with single split")
+ public void testSourceSingleSplit(TestEnvironment testEnv, ExternalContext<T> externalContext)
+ throws Exception {
+
+ // Write test data to external system
+ final Collection<T> testRecords = generateAndWriteTestData(externalContext);
+
+ // Build and execute Flink job
+ StreamExecutionEnvironment execEnv = testEnv.createExecutionEnvironment();
+
+ try (CloseableIterator<T> resultIterator =
+ execEnv.fromSource(
+ externalContext.createSource(Boundedness.BOUNDED),
+ WatermarkStrategy.noWatermarks(),
+ "Tested Source")
+ .setParallelism(1)
+ .executeAndCollect("Source Single Split Test")) {
+ // Check test result
+ assertThat(resultIterator, matchesSplitTestData(testRecords));
+ }
+ }
+
+ /**
+ * Test connector source with multiple splits in the external system
+ *
+ * <p>This test will create 4 splits in the external system, write test data to all splits, and
+ * consume back via a Flink job with 4 parallelism.
+ *
+ * <p>The number and order of records in each split consumed by Flink need to be identical to
+ * the test data written into the external system to pass this test. There's no requirement for
+ * record order across splits.
+ *
+ * <p>A bounded source is required for this test.
+ */
+ @TestTemplate
+ @DisplayName("Test source with multiple splits")
+ public void testMultipleSplits(TestEnvironment testEnv, ExternalContext<T> externalContext)
+ throws Exception {
+
+ final int splitNumber = 4;
+ final List<Collection<T>> testRecordCollections = new ArrayList<>();
+ for (int i = 0; i < splitNumber; i++) {
+ testRecordCollections.add(generateAndWriteTestData(externalContext));
+ }
+
+ LOG.debug("Build and execute Flink job");
+ StreamExecutionEnvironment execEnv = testEnv.createExecutionEnvironment();
+
+ try (final CloseableIterator<T> resultIterator =
+ execEnv.fromSource(
+ externalContext.createSource(Boundedness.BOUNDED),
+ WatermarkStrategy.noWatermarks(),
+ "Tested Source")
+ .setParallelism(splitNumber)
+ .executeAndCollect("Source Multiple Split Test")) {
+ // Check test result
+ assertThat(resultIterator, matchesMultipleSplitTestData(testRecordCollections));
+ }
+ }
+
+ /**
+ * Test connector source with an idle reader.
+ *
+ * <p>This test will create 4 split in the external system, write test data to all splits, and
+ * consume back via a Flink job with 5 parallelism, so at least one parallelism / source reader
+ * will be idle (assigned with no splits). If the split enumerator of the source doesn't signal
+ * NoMoreSplitsEvent to the idle source reader, the Flink job will never spin to FINISHED state.
+ *
+ * <p>The number and order of records in each split consumed by Flink need to be identical to
+ * the test data written into the external system to pass this test. There's no requirement for
+ * record order across splits.
+ *
+ * <p>A bounded source is required for this test.
+ */
+ @TestTemplate
+ @DisplayName("Test source with at least one idle parallelism")
+ public void testIdleReader(TestEnvironment testEnv, ExternalContext<T> externalContext)
+ throws Exception {
+
+ final int splitNumber = 4;
+ final List<Collection<T>> testRecordCollections = new ArrayList<>();
+ for (int i = 0; i < splitNumber; i++) {
+ testRecordCollections.add(generateAndWriteTestData(externalContext));
+ }
+
+ try (CloseableIterator<T> resultIterator =
+ testEnv.createExecutionEnvironment()
+ .fromSource(
+ externalContext.createSource(Boundedness.BOUNDED),
+ WatermarkStrategy.noWatermarks(),
+ "Tested Source")
+ .setParallelism(splitNumber + 1)
+ .executeAndCollect("Redundant Parallelism Test")) {
+ assertThat(resultIterator, matchesMultipleSplitTestData(testRecordCollections));
+ }
+ }
+
+ /**
+ * Test connector source with task manager failover.
+ *
+ * <p>This test will create 1 split in the external system, write test record set A into the
+ * split, restart task manager to trigger job failover, write test record set B into the split,
+ * and terminate the Flink job finally.
+ *
+ * <p>The number and order of records consumed by Flink should be identical to A before the
+ * failover and B after the failover in order to pass the test.
+ *
+ * <p>An unbounded source is required for this test, since TaskManager failover will be
+ * triggered in the middle of the test.
+ */
+ @TestTemplate
+ @DisplayName("Test TaskManager failure")
+ public void testTaskManagerFailure(
+ TestEnvironment testEnv,
+ ExternalContext<T> externalContext,
+ ClusterControllable controller)
+ throws Exception {
+
+ final Collection<T> testRecordsBeforeFailure =
+ externalContext.generateTestData(ThreadLocalRandom.current().nextLong());
+ final SourceSplitDataWriter<T> sourceSplitDataWriter =
+ externalContext.createSourceSplitDataWriter();
+ sourceSplitDataWriter.writeRecords(testRecordsBeforeFailure);
+
+ final StreamExecutionEnvironment env = testEnv.createExecutionEnvironment();
+
+ env.enableCheckpointing(50);
+ final DataStreamSource<T> dataStreamSource =
+ env.fromSource(
+ externalContext.createSource(Boundedness.CONTINUOUS_UNBOUNDED),
+ WatermarkStrategy.noWatermarks(),
+ "Tested Source")
+ .setParallelism(1);
+
+ // Since DataStream API doesn't expose job client for executeAndCollect(), we have
+ // to reuse these part of code to get both job client and result iterator :-(
+ // ------------------------------------ START ---------------------------------------------
+ TypeSerializer<T> serializer = dataStreamSource.getType().createSerializer(env.getConfig());
+ String accumulatorName = "dataStreamCollect_" + UUID.randomUUID();
+ CollectSinkOperatorFactory<T> factory =
+ new CollectSinkOperatorFactory<>(serializer, accumulatorName);
+ CollectSinkOperator<T> operator = (CollectSinkOperator<T>) factory.getOperator();
+ CollectResultIterator<T> iterator =
+ new CollectResultIterator<>(
+ operator.getOperatorIdFuture(),
+ serializer,
+ accumulatorName,
+ env.getCheckpointConfig());
+ CollectStreamSink<T> sink = new CollectStreamSink<>(dataStreamSource, factory);
+ sink.name("Data stream collect sink");
+ env.addOperator(sink.getTransformation());
+ final JobClient jobClient = env.executeAsync("TaskManager Failover Test");
+ iterator.setJobClient(jobClient);
+ // -------------------------------------- END ---------------------------------------------
+
+ assertThat(
+ iterator,
+ matchesSplitTestData(testRecordsBeforeFailure, testRecordsBeforeFailure.size()));
+
+ // -------------------------------- Trigger failover ---------------------------------------
+ controller.triggerTaskManagerFailover(jobClient, () -> {});
+
+ CommonTestUtils.waitForJobStatus(
+ jobClient,
+ Collections.singletonList(JobStatus.RUNNING),
+ Deadline.fromNow(Duration.ofSeconds(30)));
+
+ final Collection<T> testRecordsAfterFailure =
+ externalContext.generateTestData(ThreadLocalRandom.current().nextLong());
+ sourceSplitDataWriter.writeRecords(testRecordsAfterFailure);
+
+ assertThat(
+ iterator,
+ matchesSplitTestData(testRecordsAfterFailure, testRecordsAfterFailure.size()));
+
+ // Clean up
+ iterator.close();
+ CommonTestUtils.terminateJob(jobClient, Duration.ofSeconds(30));
+ CommonTestUtils.waitForJobStatus(
+ jobClient,
+ Collections.singletonList(JobStatus.CANCELED),
+ Deadline.fromNow(Duration.ofSeconds(30)));
+ }
+
+ // ----------------------------- Helper Functions ---------------------------------
+
+ /**
+ * Generate a set of test records and write it to the given split writer.
+ *
+ * @param externalContext External context
+ * @return Collection of generated test records
+ */
+ protected Collection<T> generateAndWriteTestData(ExternalContext<T> externalContext) {
+ final Collection<T> testRecordCollection =
+ externalContext.generateTestData(ThreadLocalRandom.current().nextLong());
+ LOG.debug("Writing {} records to external system", testRecordCollection.size());
+ externalContext.createSourceSplitDataWriter().writeRecords(testRecordCollection);
+ return testRecordCollection;
+ }
+}
diff --git a/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/utils/TestDataMatchers.java b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/utils/TestDataMatchers.java
new file mode 100644
index 0000000..87d4905
--- /dev/null
+++ b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/utils/TestDataMatchers.java
@@ -0,0 +1,246 @@
+/*
+ * 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.connectors.test.common.utils;
+
+import org.apache.flink.annotation.Internal;
+
+import org.hamcrest.Description;
+import org.hamcrest.TypeSafeDiagnosingMatcher;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.NoSuchElementException;
+
+/** Matchers for validating test data. */
+@Internal
+public class TestDataMatchers {
+
+ // ---------------------------- Matcher Builders ----------------------------------
+ public static <T> MultipleSplitDataMatcher<T> matchesMultipleSplitTestData(
+ Collection<Collection<T>> testDataCollections) {
+ return new MultipleSplitDataMatcher<>(testDataCollections);
+ }
+
+ public static <T> SplitDataMatcher<T> matchesSplitTestData(Collection<T> testData) {
+ return new SplitDataMatcher<>(testData);
+ }
+
+ public static <T> SplitDataMatcher<T> matchesSplitTestData(Collection<T> testData, int limit) {
+ return new SplitDataMatcher<>(testData, limit);
+ }
+
+ // ---------------------------- Matcher Definitions --------------------------------
+
+ /**
+ * Matcher for validating test data in a single split.
+ *
+ * @param <T> Type of validating record
+ */
+ public static class SplitDataMatcher<T> extends TypeSafeDiagnosingMatcher<Iterator<T>> {
+ private static final int UNSET = -1;
+
+ private final Collection<T> testData;
+ private final int limit;
+
+ private String mismatchDescription = null;
+
+ public SplitDataMatcher(Collection<T> testData) {
+ this.testData = testData;
+ this.limit = UNSET;
+ }
+
+ public SplitDataMatcher(Collection<T> testData, int limit) {
+ if (limit > testData.size()) {
+ throw new IllegalArgumentException(
+ "Limit validation size should be less than number of test records");
+ }
+ this.testData = testData;
+ this.limit = limit;
+ }
+
+ @Override
+ protected boolean matchesSafely(Iterator<T> resultIterator, Description description) {
+ if (mismatchDescription != null) {
+ description.appendText(mismatchDescription);
+ return false;
+ }
+
+ int recordCounter = 0;
+ for (T testRecord : testData) {
+ if (!resultIterator.hasNext()) {
+ mismatchDescription = "Result data is less than test data";
+ return false;
+ }
+ T resultRecord = resultIterator.next();
+ if (!testRecord.equals(resultRecord)) {
+ mismatchDescription =
+ String.format(
+ "Mismatched record at position %d: Expected '%s' but was '%s'",
+ recordCounter, testRecord, resultRecord);
+ return false;
+ }
+ recordCounter++;
+ if (limit != UNSET && recordCounter >= limit) {
+ break;
+ }
+ }
+ if (limit == UNSET && resultIterator.hasNext()) {
+ mismatchDescription = "Result data is more than test data";
+ return false;
+ } else {
+ return true;
+ }
+ }
+
+ @Override
+ public void describeTo(Description description) {
+ description.appendText(
+ "Records consumed by Flink should be identical to test data "
+ + "and preserve the order in split");
+ }
+ }
+
+ /**
+ * Matcher for validating test data from multiple splits.
+ *
+ * <p>Each collection has a pointer (iterator) pointing to current checking record. When a
+ * record is received in the stream, it will be compared to all current pointing records in
+ * collections, and the pointer to the identical record will move forward.
+ *
+ * <p>If the stream preserves the correctness and order of records in all splits, all pointers
+ * should reach the end of the collection finally.
+ *
+ * @param <T> Type of validating record
+ */
+ public static class MultipleSplitDataMatcher<T> extends TypeSafeDiagnosingMatcher<Iterator<T>> {
+
+ private final List<IteratorWithCurrent<T>> testDataIterators = new ArrayList<>();
+
+ private String mismatchDescription = null;
+
+ public MultipleSplitDataMatcher(Collection<Collection<T>> testDataCollections) {
+ for (Collection<T> testDataCollection : testDataCollections) {
+ testDataIterators.add(new IteratorWithCurrent<>(testDataCollection.iterator()));
+ }
+ }
+
+ @Override
+ protected boolean matchesSafely(Iterator<T> resultIterator, Description description) {
+ if (mismatchDescription != null) {
+ description.appendText(mismatchDescription);
+ return false;
+ }
+ while (resultIterator.hasNext()) {
+ final T record = resultIterator.next();
+ if (!matchThenNext(record)) {
+ mismatchDescription = String.format("Unexpected record '%s'", record);
+ return false;
+ }
+ }
+ if (!hasReachedEnd()) {
+ mismatchDescription = "Result data is less than test data";
+ return false;
+ } else {
+ return true;
+ }
+ }
+
+ @Override
+ public void describeTo(Description description) {
+ description.appendText(
+ "Records consumed by Flink should be identical to test data "
+ + "and preserve the order in multiple splits");
+ }
+
+ /**
+ * Check if any pointing data is identical to the record from the stream, and move the
+ * pointer to next record if matched. Otherwise throws an exception.
+ *
+ * @param record Record from stream
+ */
+ private boolean matchThenNext(T record) {
+ for (IteratorWithCurrent<T> testDataIterator : testDataIterators) {
+ if (record.equals(testDataIterator.current())) {
+ testDataIterator.next();
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Whether all pointers have reached the end of collections.
+ *
+ * @return True if all pointers have reached the end.
+ */
+ private boolean hasReachedEnd() {
+ for (IteratorWithCurrent<T> testDataIterator : testDataIterators) {
+ if (testDataIterator.hasNext()) {
+ return false;
+ }
+ }
+ return true;
+ }
+ }
+
+ /**
+ * An iterator wrapper which can access the element that the iterator is currently pointing to.
+ *
+ * @param <E> The type of elements returned by this iterator.
+ */
+ static class IteratorWithCurrent<E> implements Iterator<E> {
+
+ private final Iterator<E> originalIterator;
+ private E current;
+
+ public IteratorWithCurrent(Iterator<E> originalIterator) {
+ this.originalIterator = originalIterator;
+ try {
+ current = originalIterator.next();
+ } catch (NoSuchElementException e) {
+ current = null;
+ }
+ }
+
+ @Override
+ public boolean hasNext() {
+ return current != null;
+ }
+
+ @Override
+ public E next() {
+ if (current == null) {
+ throw new NoSuchElementException();
+ }
+ E previous = current;
+ if (originalIterator.hasNext()) {
+ current = originalIterator.next();
+ } else {
+ current = null;
+ }
+ return previous;
+ }
+
+ public E current() {
+ return current;
+ }
+ }
+}
diff --git a/flink-test-utils-parent/flink-connector-testing/src/test/java/org/apache/flink/connectors/test/common/utils/IteratorWithCurrentTest.java b/flink-test-utils-parent/flink-connector-testing/src/test/java/org/apache/flink/connectors/test/common/utils/IteratorWithCurrentTest.java
new file mode 100644
index 0000000..97a7b48
--- /dev/null
+++ b/flink-test-utils-parent/flink-connector-testing/src/test/java/org/apache/flink/connectors/test/common/utils/IteratorWithCurrentTest.java
@@ -0,0 +1,77 @@
+/*
+ * 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.connectors.test.common.utils;
+
+import org.apache.flink.connectors.test.common.utils.TestDataMatchers.IteratorWithCurrent;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.NoSuchElementException;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/** Unit test for {@link IteratorWithCurrent}. */
+public class IteratorWithCurrentTest {
+
+ @Test
+ public void testIterator() {
+ List<Integer> numberList = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ numberList.add(i);
+ }
+ IteratorWithCurrent<Integer> iterator = new IteratorWithCurrent<>(numberList.iterator());
+ Integer num = 0;
+ while (iterator.hasNext()) {
+ assertEquals(num, iterator.next());
+ num++;
+ }
+ assertEquals(10, num.intValue());
+ assertNull(iterator.current());
+ }
+
+ @Test
+ public void testEmptyList() {
+ IteratorWithCurrent<Integer> iterator =
+ new IteratorWithCurrent<>(Collections.emptyIterator());
+ assertNull(iterator.current());
+ assertThrows(NoSuchElementException.class, iterator::next);
+ }
+
+ @Test
+ public void testCurrentElement() {
+ List<Integer> numberList = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ numberList.add(i);
+ }
+ IteratorWithCurrent<Integer> iterator = new IteratorWithCurrent<>(numberList.iterator());
+ Integer num = 0;
+ while (iterator.hasNext()) {
+ assertEquals(num, iterator.current());
+ assertEquals(num, iterator.next());
+ num++;
+ }
+ assertEquals(10, num.intValue());
+ assertNull(iterator.current());
+ }
+}
diff --git a/flink-test-utils-parent/flink-connector-testing/src/test/java/org/apache/flink/connectors/test/common/utils/TestDataMatchersTest.java b/flink-test-utils-parent/flink-connector-testing/src/test/java/org/apache/flink/connectors/test/common/utils/TestDataMatchersTest.java
new file mode 100644
index 0000000..1833185
--- /dev/null
+++ b/flink-test-utils-parent/flink-connector-testing/src/test/java/org/apache/flink/connectors/test/common/utils/TestDataMatchersTest.java
@@ -0,0 +1,166 @@
+/*
+ * 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.connectors.test.common.utils;
+
+import org.hamcrest.Description;
+import org.hamcrest.StringDescription;
+import org.hamcrest.TypeSafeDiagnosingMatcher;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+/** Unit test for {@link TestDataMatchers}. */
+public class TestDataMatchersTest {
+ @Nested
+ class SplitDataMatcherTest {
+ private final List<String> testData = Arrays.asList("alpha", "beta", "gamma");
+
+ @Test
+ public void testPositiveCases() {
+ assertThat(testData.iterator(), TestDataMatchers.matchesSplitTestData(testData));
+ assertThat(
+ testData.iterator(),
+ TestDataMatchers.matchesSplitTestData(testData, testData.size()));
+ assertThat(
+ testData.iterator(),
+ TestDataMatchers.matchesSplitTestData(testData, testData.size() - 1));
+ }
+
+ @Test
+ public void testMismatch() throws Exception {
+ final List<String> resultData = new ArrayList<>(testData);
+ resultData.set(1, "delta");
+ final Iterator<String> resultIterator = resultData.iterator();
+
+ final TestDataMatchers.SplitDataMatcher<String> matcher =
+ TestDataMatchers.matchesSplitTestData(testData);
+
+ assertMatcherFailedWithDescription(
+ resultIterator,
+ matcher,
+ "Mismatched record at position 1: Expected 'beta' but was 'delta'");
+ }
+
+ @Test
+ public void testResultMoreThanExpected() throws Exception {
+ final List<String> resultData = new ArrayList<>(testData);
+ resultData.add("delta");
+ final Iterator<String> resultIterator = resultData.iterator();
+
+ final TestDataMatchers.SplitDataMatcher<String> matcher =
+ TestDataMatchers.matchesSplitTestData(testData);
+
+ assertMatcherFailedWithDescription(
+ resultIterator, matcher, "Result data is more than test data");
+ }
+
+ @Test
+ public void testResultLessThanExpected() throws Exception {
+ final List<String> resultData = new ArrayList<>(testData);
+ resultData.remove(testData.size() - 1);
+ final Iterator<String> resultIterator = resultData.iterator();
+
+ final TestDataMatchers.SplitDataMatcher<String> matcher =
+ TestDataMatchers.matchesSplitTestData(testData);
+
+ assertMatcherFailedWithDescription(
+ resultIterator, matcher, "Result data is less than test data");
+ }
+ }
+
+ @Nested
+ class MultipleSplitDataMatcherTest {
+ private final List<String> splitA = Arrays.asList("alpha", "beta", "gamma");
+ private final List<String> splitB = Arrays.asList("one", "two", "three");
+ private final List<String> splitC = Arrays.asList("1", "2", "3");
+ private final List<Collection<String>> testDataCollection =
+ Arrays.asList(splitA, splitB, splitC);
+
+ @Test
+ public void testPositiveCase() {
+ final List<String> result = unionLists(splitA, splitB, splitC);
+ assertThat(
+ result.iterator(),
+ TestDataMatchers.matchesMultipleSplitTestData(testDataCollection));
+ }
+
+ @Test
+ public void testResultLessThanExpected() throws Exception {
+ final ArrayList<String> splitATestDataWithoutLast = new ArrayList<>(splitA);
+ splitATestDataWithoutLast.remove(splitA.size() - 1);
+ final List<String> result = unionLists(splitATestDataWithoutLast, splitB, splitC);
+ final TestDataMatchers.MultipleSplitDataMatcher<String> matcher =
+ TestDataMatchers.matchesMultipleSplitTestData(testDataCollection);
+ assertMatcherFailedWithDescription(
+ result.iterator(), matcher, "Result data is less than test data");
+ }
+
+ @Test
+ public void testResultMoreThanExpected() throws Exception {
+ final List<String> result = unionLists(splitA, splitB, splitC);
+ result.add("delta");
+ final TestDataMatchers.MultipleSplitDataMatcher<String> matcher =
+ TestDataMatchers.matchesMultipleSplitTestData(testDataCollection);
+ assertMatcherFailedWithDescription(
+ result.iterator(), matcher, "Unexpected record 'delta'");
+ }
+
+ @Test
+ public void testOutOfOrder() throws Exception {
+ List<String> reverted = new ArrayList<>(splitC);
+ Collections.reverse(reverted);
+ final List<String> result = unionLists(splitA, splitB, reverted);
+ final TestDataMatchers.MultipleSplitDataMatcher<String> matcher =
+ TestDataMatchers.matchesMultipleSplitTestData(testDataCollection);
+ assertMatcherFailedWithDescription(result.iterator(), matcher, "Unexpected record '3'");
+ }
+ }
+
+ @SafeVarargs
+ private final <T> List<T> unionLists(List<T>... lists) {
+ return Stream.of(lists).flatMap(Collection::stream).collect(Collectors.toList());
+ }
+
+ private <T> void assertMatcherFailedWithDescription(
+ T object, TypeSafeDiagnosingMatcher<T> matcher, String expectedDescription)
+ throws Exception {
+ final Method method =
+ TypeSafeDiagnosingMatcher.class.getDeclaredMethod(
+ "matchesSafely", Object.class, Description.class);
+ method.setAccessible(true);
+ assertFalse((boolean) method.invoke(matcher, object, new Description.NullDescription()));
+
+ final StringDescription actualDescription = new StringDescription();
+ method.invoke(matcher, object, actualDescription);
+ Assertions.assertEquals(expectedDescription, actualDescription.toString());
+ }
+}
[flink] 01/07: [hotfix][test/testcontainers] Use Testcontainer BOM
to manage versions in the project
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 4fd33234407e07d99647e495819e009467828161
Author: Qingsheng Ren <re...@gmail.com>
AuthorDate: Mon Aug 9 11:51:52 2021 +0800
[hotfix][test/testcontainers] Use Testcontainer BOM to manage versions in the project
---
flink-connectors/flink-connector-elasticsearch6/pom.xml | 1 -
flink-connectors/flink-connector-elasticsearch7/pom.xml | 1 -
flink-connectors/flink-connector-jdbc/pom.xml | 7 +++++--
flink-connectors/flink-connector-kafka/pom.xml | 1 -
flink-connectors/flink-connector-kinesis/pom.xml | 1 -
flink-connectors/flink-connector-rabbitmq/pom.xml | 1 -
.../flink-end-to-end-tests-common-kafka/pom.xml | 1 -
flink-end-to-end-tests/flink-end-to-end-tests-common/pom.xml | 1 -
pom.xml | 8 ++++++++
9 files changed, 13 insertions(+), 9 deletions(-)
diff --git a/flink-connectors/flink-connector-elasticsearch6/pom.xml b/flink-connectors/flink-connector-elasticsearch6/pom.xml
index 6491887..e62bd26 100644
--- a/flink-connectors/flink-connector-elasticsearch6/pom.xml
+++ b/flink-connectors/flink-connector-elasticsearch6/pom.xml
@@ -91,7 +91,6 @@ under the License.
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>elasticsearch</artifactId>
- <version>1.15.1</version>
<scope>test</scope>
</dependency>
diff --git a/flink-connectors/flink-connector-elasticsearch7/pom.xml b/flink-connectors/flink-connector-elasticsearch7/pom.xml
index 0141d69..392c9a8 100644
--- a/flink-connectors/flink-connector-elasticsearch7/pom.xml
+++ b/flink-connectors/flink-connector-elasticsearch7/pom.xml
@@ -89,7 +89,6 @@ under the License.
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>elasticsearch</artifactId>
- <version>1.15.1</version>
<scope>test</scope>
</dependency>
diff --git a/flink-connectors/flink-connector-jdbc/pom.xml b/flink-connectors/flink-connector-jdbc/pom.xml
index 0110505..f7ad719 100644
--- a/flink-connectors/flink-connector-jdbc/pom.xml
+++ b/flink-connectors/flink-connector-jdbc/pom.xml
@@ -123,6 +123,11 @@ under the License.
<version>${otj-pg-embedded.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>postgresql</artifactId>
+ <scope>test</scope>
+ </dependency>
<!-- MySQL tests -->
<dependency>
@@ -166,14 +171,12 @@ under the License.
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>postgresql</artifactId>
- <version>1.15.3</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>mysql</artifactId>
- <version>1.15.3</version>
<scope>test</scope>
</dependency>
</dependencies>
diff --git a/flink-connectors/flink-connector-kafka/pom.xml b/flink-connectors/flink-connector-kafka/pom.xml
index 9a1fdef..2301a06 100644
--- a/flink-connectors/flink-connector-kafka/pom.xml
+++ b/flink-connectors/flink-connector-kafka/pom.xml
@@ -121,7 +121,6 @@ under the License.
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>kafka</artifactId>
- <version>1.15.1</version>
<scope>test</scope>
</dependency>
diff --git a/flink-connectors/flink-connector-kinesis/pom.xml b/flink-connectors/flink-connector-kinesis/pom.xml
index 8068b3f..fba1727 100644
--- a/flink-connectors/flink-connector-kinesis/pom.xml
+++ b/flink-connectors/flink-connector-kinesis/pom.xml
@@ -195,7 +195,6 @@ under the License.
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers</artifactId>
- <version>1.15.3</version>
<scope>test</scope>
</dependency>
diff --git a/flink-connectors/flink-connector-rabbitmq/pom.xml b/flink-connectors/flink-connector-rabbitmq/pom.xml
index 12b6cd2..f1904eb 100644
--- a/flink-connectors/flink-connector-rabbitmq/pom.xml
+++ b/flink-connectors/flink-connector-rabbitmq/pom.xml
@@ -87,7 +87,6 @@ under the License.
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>rabbitmq</artifactId>
- <version>1.15.3</version>
<scope>test</scope>
</dependency>
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/pom.xml b/flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/pom.xml
index ba5d81f..8c3b90d 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/pom.xml
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/pom.xml
@@ -104,7 +104,6 @@ under the License.
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>kafka</artifactId>
- <version>1.15.1</version>
</dependency>
<dependency>
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-common/pom.xml b/flink-end-to-end-tests/flink-end-to-end-tests-common/pom.xml
index aa8b301..702fe8d 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-common/pom.xml
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-common/pom.xml
@@ -63,7 +63,6 @@ under the License.
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers</artifactId>
- <version>1.15.1</version>
</dependency>
<dependency>
<!-- To ensure that flink-dist is built beforehand -->
diff --git a/pom.xml b/pom.xml
index e162904..9eace84 100644
--- a/pom.xml
+++ b/pom.xml
@@ -135,6 +135,7 @@ under the License.
<protoc.version>3.17.3</protoc.version>
<arrow.version>0.16.0</arrow.version>
<okhttp.version>3.14.9</okhttp.version>
+ <testcontainers.version>1.15.3</testcontainers.version>
<lz4.version>1.8.0</lz4.version>
<japicmp.skip>false</japicmp.skip>
<flink.convergence.phase>validate</flink.convergence.phase>
@@ -787,6 +788,13 @@ under the License.
<artifactId>snakeyaml</artifactId>
<version>1.27</version>
</dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>testcontainers-bom</artifactId>
+ <version>${testcontainers.version}</version>
+ <type>pom</type>
+ <scope>import</scope>
+ </dependency>
</dependencies>
</dependencyManagement>
[flink] 02/07: [FLINK-19554][connector/testing-framework] Basic
abstractions of connector testing framework
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 613a824c33d25925751bb875c8c54d9f50b7efe4
Author: Qingsheng Ren <re...@gmail.com>
AuthorDate: Thu Aug 12 17:37:00 2021 +0800
[FLINK-19554][connector/testing-framework] Basic abstractions of connector testing framework
---
.../flink-connector-testing/README.md | 44 ++++++++++++
.../flink-connector-testing/pom.xml | 79 ++++++++++++++++++++++
.../flink/connectors/test/common/TestResource.java | 53 +++++++++++++++
.../common/environment/ClusterControllable.java | 33 +++++++++
.../test/common/environment/TestEnvironment.java | 35 ++++++++++
.../test/common/external/ExternalContext.java | 75 ++++++++++++++++++++
.../common/external/SourceSplitDataWriter.java | 34 ++++++++++
flink-test-utils-parent/pom.xml | 1 +
8 files changed, 354 insertions(+)
diff --git a/flink-test-utils-parent/flink-connector-testing/README.md b/flink-test-utils-parent/flink-connector-testing/README.md
new file mode 100644
index 0000000..e6d2e8f
--- /dev/null
+++ b/flink-test-utils-parent/flink-connector-testing/README.md
@@ -0,0 +1,44 @@
+# Flink Connector Testing Framework
+
+This module is dedicated to providing a unified and easy-to-use testing framework for all
+Flink-managed and third-party connectors in Flink eco-system.
+
+## Overview
+
+The testing framework bases on JUnit 5 for test case running and resource lifecycle management. The
+design goal of the framework is: connector developer only needs to provide instance of their
+source/sink and it's related external system, then the framework will deal with tests with bundled
+IT and end-to-end test cases.
+
+## Interfaces
+
+### Test Environment
+
+Test environment provides an execution environment of running the Flink job for testing (mini
+cluster, local standalone cluster, Flink on Docker...), also a configuration for configuring the
+environment and passing other required information into the test scenario.
+
+The framework provides MiniCluster and Flink on Docker out-of-box. You can also implement your own
+test environment by implementing interface ```TestEnvironment```.
+
+### External System & Context
+
+You can define your own external system by implementing interface ```ExternalSystem```, and testing
+framework will handle the lifecycle of external system automatically.
+
+External context defines how to interact with the external system, providing instance of source and
+connecting to the external system, test data and a writer for producing data into external system.
+
+## Using Testing Framework in Your Connector
+
+Testing framework using JUnit 5 features for supporting test cases running. We provide a base class
+```SourceTestSuiteBase``` with some basic test cases. You can start simply by extending this base class,
+instantiate test environment, external system and context in your test class constructor, and
+annotate them with ```@WithTestEnvironment```, ```@WithExternalSystem```
+and ```@WithExternalContextFactory```.
+
+You can refer to ```KafkaSourceE2ECase``` as an example for using the testing framework.
+
+Also, you can develop your own test cases in your test class. Just simply annotate your test cases
+with ```@Case```, and testing framework will inject test environments and external systems into you
+cases.
diff --git a/flink-test-utils-parent/flink-connector-testing/pom.xml b/flink-test-utils-parent/flink-connector-testing/pom.xml
new file mode 100644
index 0000000..3fab36b
--- /dev/null
+++ b/flink-test-utils-parent/flink-connector-testing/pom.xml
@@ -0,0 +1,79 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-test-utils-parent</artifactId>
+ <version>1.14-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <artifactId>flink-connector-testing_${scala.binary.version}</artifactId>
+ <name>Flink : Test utils : Testing Framework</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter</artifactId>
+ <scope>compile</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-test-utils_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>testcontainers</artifactId>
+ </dependency>
+ </dependencies>
+
+ <dependencyManagement>
+ <dependencies>
+ <!-- We pick an arbitrary version of net.java.dev.jna:jna to satisfy dependency
+ convergence for org.testcontainers:testcontainers which transitively depends on
+ two different versions.-->
+ <dependency>
+ <groupId>net.java.dev.jna</groupId>
+ <artifactId>jna</artifactId>
+ <version>5.5.0</version>
+ </dependency>
+ </dependencies>
+ </dependencyManagement>
+</project>
diff --git a/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/TestResource.java b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/TestResource.java
new file mode 100644
index 0000000..05d0174
--- /dev/null
+++ b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/TestResource.java
@@ -0,0 +1,53 @@
+/*
+ * 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.connectors.test.common;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.connectors.test.common.environment.TestEnvironment;
+import org.apache.flink.connectors.test.common.junit.annotations.ExternalSystem;
+import org.apache.flink.connectors.test.common.junit.annotations.TestEnv;
+
+/**
+ * Basic abstractions for all resources used in connector testing framework, including {@link
+ * TestEnvironment} annotated by {@link TestEnv} and external system annotated by {@link
+ * ExternalSystem}.
+ *
+ * <p>Lifecycle of test resources will be managed by the framework.
+ */
+@Experimental
+public interface TestResource {
+
+ /**
+ * Start up the test resource.
+ *
+ * <p>The implementation of this method should be idempotent.
+ *
+ * @throws Exception if anything wrong when starting the resource
+ */
+ void startUp() throws Exception;
+
+ /**
+ * Tear down the test resource.
+ *
+ * <p>The test resource should be able to tear down even without a startup (could be a no-op).
+ *
+ * @throws Exception if anything wrong when tearing the resource down
+ */
+ void tearDown() throws Exception;
+}
diff --git a/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/environment/ClusterControllable.java b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/environment/ClusterControllable.java
new file mode 100644
index 0000000..df68b42
--- /dev/null
+++ b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/environment/ClusterControllable.java
@@ -0,0 +1,33 @@
+/*
+ * 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.connectors.test.common.environment;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.core.execution.JobClient;
+
+/** Interface for triggering failover in a Flink cluster. */
+@Experimental
+public interface ClusterControllable {
+
+ void triggerJobManagerFailover(JobClient jobClient, Runnable afterFailAction) throws Exception;
+
+ void triggerTaskManagerFailover(JobClient jobClient, Runnable afterFailAction) throws Exception;
+
+ void isolateNetwork(JobClient jobClient, Runnable afterFailAction) throws Exception;
+}
diff --git a/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/environment/TestEnvironment.java b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/environment/TestEnvironment.java
new file mode 100644
index 0000000..2c47da2
--- /dev/null
+++ b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/environment/TestEnvironment.java
@@ -0,0 +1,35 @@
+/*
+ * 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.connectors.test.common.environment;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.connectors.test.common.TestResource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+/** Test environment for running Flink jobs. */
+@Experimental
+public interface TestEnvironment extends TestResource {
+
+ /**
+ * Create a new {@link StreamExecutionEnvironment} for configuring and executing the Flink job.
+ *
+ * @return An instance of execution environment
+ */
+ StreamExecutionEnvironment createExecutionEnvironment();
+}
diff --git a/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/external/ExternalContext.java b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/external/ExternalContext.java
new file mode 100644
index 0000000..dacc3c1
--- /dev/null
+++ b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/external/ExternalContext.java
@@ -0,0 +1,75 @@
+/*
+ * 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.connectors.test.common.external;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.Source;
+
+import java.io.Serializable;
+import java.util.Collection;
+
+/**
+ * Context of the test interacting with external system.
+ *
+ * <p>This context is responsible for providing:
+ *
+ * <ul>
+ * <li>Instance of sources connecting to external system
+ * <li>{@link SourceSplitDataWriter} for creating splits and writing test data into the created
+ * split
+ * <li>Test data to write into the external system
+ * </ul>
+ *
+ * @param <T> Type of elements after deserialization by source
+ */
+@Experimental
+public interface ExternalContext<T> extends Serializable, AutoCloseable {
+
+ /**
+ * Create a new instance of connector source implemented in {@link Source}.
+ *
+ * @return A new instance of Source
+ */
+ Source<T, ?, ?> createSource(Boundedness boundedness);
+
+ /**
+ * Create a new split in the external system and a data writer corresponding to the new split.
+ *
+ * @return A data writer for the created split.
+ */
+ SourceSplitDataWriter<T> createSourceSplitDataWriter();
+
+ /**
+ * Generate test data.
+ *
+ * @param seed Seed for generating random test data set.
+ * @return Collection of generated test data.
+ */
+ Collection<T> generateTestData(long seed);
+
+ /**
+ * Factory for {@link ExternalContext}.
+ *
+ * @param <T> Type of elements after deserialization by source
+ */
+ interface Factory<T> {
+ ExternalContext<T> createExternalContext();
+ }
+}
diff --git a/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/external/SourceSplitDataWriter.java b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/external/SourceSplitDataWriter.java
new file mode 100644
index 0000000..247358b
--- /dev/null
+++ b/flink-test-utils-parent/flink-connector-testing/src/main/java/org/apache/flink/connectors/test/common/external/SourceSplitDataWriter.java
@@ -0,0 +1,34 @@
+/*
+ * 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.connectors.test.common.external;
+
+import org.apache.flink.annotation.Experimental;
+
+import java.util.Collection;
+
+/**
+ * A data writer for writing records into a {@link
+ * org.apache.flink.api.connector.source.SourceSplit} in the external system.
+ *
+ * @param <T> Type of writing records
+ */
+@Experimental
+public interface SourceSplitDataWriter<T> extends AutoCloseable {
+ void writeRecords(Collection<T> records);
+}
diff --git a/flink-test-utils-parent/pom.xml b/flink-test-utils-parent/pom.xml
index 801348e..2b4a4d7 100644
--- a/flink-test-utils-parent/pom.xml
+++ b/flink-test-utils-parent/pom.xml
@@ -39,6 +39,7 @@ under the License.
<module>flink-test-utils-junit</module>
<module>flink-test-utils</module>
<module>flink-connector-test-utils</module>
+ <module>flink-connector-testing</module>
</modules>
</project>
[flink] 07/07: [FLINK-19554][connector/testing-framework]
KafkaSource IT and E2E case based on connector testing framework
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit adee56117778f6962d7892a26b2b507b67ec707a
Author: Qingsheng Ren <re...@gmail.com>
AuthorDate: Thu Aug 12 18:01:11 2021 +0800
[FLINK-19554][connector/testing-framework] KafkaSource IT and E2E case based on connector testing framework
---
flink-connectors/flink-connector-kafka/pom.xml | 12 +-
.../connector/kafka/source/KafkaSourceITCase.java | 313 ++++++++++++---------
.../source/enumerator/KafkaEnumeratorTest.java | 2 +-
.../initializer/OffsetsInitializerTest.java | 2 +-
.../enumerator/subscriber/KafkaSubscriberTest.java | 2 +-
.../reader/KafkaPartitionSplitReaderTest.java | 2 +-
.../kafka/source/reader/KafkaSourceReaderTest.java | 2 +-
.../KafkaMultipleTopicExternalContext.java | 124 ++++++++
.../source/testutils/KafkaPartitionDataWriter.java | 60 ++++
.../testutils/KafkaSingleTopicExternalContext.java | 242 ++++++++++++++++
.../source/{ => testutils}/KafkaSourceTestEnv.java | 2 +-
.../flink-end-to-end-tests-common-kafka/pom.xml | 29 ++
.../flink/tests/util/kafka/KafkaSourceE2ECase.java | 69 +++++
.../modules-skipping-deployment.modulelist | 1 +
14 files changed, 721 insertions(+), 141 deletions(-)
diff --git a/flink-connectors/flink-connector-kafka/pom.xml b/flink-connectors/flink-connector-kafka/pom.xml
index 2301a06..0da6052 100644
--- a/flink-connectors/flink-connector-kafka/pom.xml
+++ b/flink-connectors/flink-connector-kafka/pom.xml
@@ -210,6 +210,12 @@ under the License.
<version>${project.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-testing_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
@@ -224,7 +230,8 @@ under the License.
</goals>
<configuration>
<includes>
- <include>**/KafkaTestEnvironmentImpl*</include>
+ <include>**/KafkaTestEnvironment*</include>
+ <include>**/testutils/*</include>
<include>META-INF/LICENSE</include>
<include>META-INF/NOTICE</include>
</includes>
@@ -247,7 +254,8 @@ under the License.
<addMavenDescriptor>false</addMavenDescriptor>
</archive>
<includes>
- <include>**/KafkaTestEnvironmentImpl*</include>
+ <include>**/KafkaTestEnvironment*</include>
+ <include>**/testutils/*</include>
<include>META-INF/LICENSE</include>
<include>META-INF/NOTICE</include>
</includes>
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceITCase.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceITCase.java
index 89ce39c..5c2107a 100644
--- a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceITCase.java
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceITCase.java
@@ -26,6 +26,15 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema;
+import org.apache.flink.connector.kafka.source.testutils.KafkaMultipleTopicExternalContext;
+import org.apache.flink.connector.kafka.source.testutils.KafkaSingleTopicExternalContext;
+import org.apache.flink.connector.kafka.source.testutils.KafkaSourceTestEnv;
+import org.apache.flink.connectors.test.common.environment.MiniClusterTestEnvironment;
+import org.apache.flink.connectors.test.common.external.DefaultContainerizedExternalSystem;
+import org.apache.flink.connectors.test.common.junit.annotations.ExternalContextFactory;
+import org.apache.flink.connectors.test.common.junit.annotations.ExternalSystem;
+import org.apache.flink.connectors.test.common.junit.annotations.TestEnv;
+import org.apache.flink.connectors.test.common.testsuites.SourceTestSuiteBase;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
@@ -40,9 +49,14 @@ import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.IntegerDeserializer;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.Test;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.TestInstance.Lifecycle;
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.utility.DockerImageName;
import java.io.IOException;
import java.io.Serializable;
@@ -54,147 +68,181 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
-import static org.junit.Assert.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
/** Unite test class for {@link KafkaSource}. */
public class KafkaSourceITCase {
private static final String TOPIC1 = "topic1";
private static final String TOPIC2 = "topic2";
- @BeforeClass
- public static void setup() throws Throwable {
- KafkaSourceTestEnv.setup();
- KafkaSourceTestEnv.setupTopic(TOPIC1, true, true);
- KafkaSourceTestEnv.setupTopic(TOPIC2, true, true);
- }
-
- @AfterClass
- public static void tearDown() throws Exception {
- KafkaSourceTestEnv.tearDown();
- }
-
- @Test
- public void testTimestamp() throws Throwable {
- final String topic = "testTimestamp";
- KafkaSourceTestEnv.createTestTopic(topic, 1, 1);
- KafkaSourceTestEnv.produceToKafka(
- Arrays.asList(
- new ProducerRecord<>(topic, 0, 1L, "key0", 0),
- new ProducerRecord<>(topic, 0, 2L, "key1", 1),
- new ProducerRecord<>(topic, 0, 3L, "key2", 2)));
-
- KafkaSource<PartitionAndValue> source =
- KafkaSource.<PartitionAndValue>builder()
- .setBootstrapServers(KafkaSourceTestEnv.brokerConnectionStrings)
- .setGroupId("testTimestampAndWatermark")
- .setTopics(topic)
- .setDeserializer(new TestingKafkaRecordDeserializationSchema())
- .setStartingOffsets(OffsetsInitializer.earliest())
- .setBounded(OffsetsInitializer.latest())
- .build();
-
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.setParallelism(1);
- DataStream<PartitionAndValue> stream =
- env.fromSource(source, WatermarkStrategy.noWatermarks(), "testTimestamp");
+ @Nested
+ @TestInstance(Lifecycle.PER_CLASS)
+ class KafkaSpecificTests {
+ @BeforeAll
+ public void setup() throws Throwable {
+ KafkaSourceTestEnv.setup();
+ KafkaSourceTestEnv.setupTopic(TOPIC1, true, true);
+ KafkaSourceTestEnv.setupTopic(TOPIC2, true, true);
+ }
- // Verify that the timestamp and watermark are working fine.
- stream.transform(
- "timestampVerifier",
- TypeInformation.of(PartitionAndValue.class),
- new WatermarkVerifyingOperator(v -> v));
- stream.addSink(new DiscardingSink<>());
- JobExecutionResult result = env.execute();
+ @AfterAll
+ public void tearDown() throws Exception {
+ KafkaSourceTestEnv.tearDown();
+ }
- assertEquals(Arrays.asList(1L, 2L, 3L), result.getAccumulatorResult("timestamp"));
- }
+ @Test
+ public void testTimestamp() throws Throwable {
+ final String topic = "testTimestamp";
+ KafkaSourceTestEnv.createTestTopic(topic, 1, 1);
+ KafkaSourceTestEnv.produceToKafka(
+ Arrays.asList(
+ new ProducerRecord<>(topic, 0, 1L, "key0", 0),
+ new ProducerRecord<>(topic, 0, 2L, "key1", 1),
+ new ProducerRecord<>(topic, 0, 3L, "key2", 2)));
+
+ KafkaSource<PartitionAndValue> source =
+ KafkaSource.<PartitionAndValue>builder()
+ .setBootstrapServers(KafkaSourceTestEnv.brokerConnectionStrings)
+ .setGroupId("testTimestampAndWatermark")
+ .setTopics(topic)
+ .setDeserializer(new TestingKafkaRecordDeserializationSchema())
+ .setStartingOffsets(OffsetsInitializer.earliest())
+ .setBounded(OffsetsInitializer.latest())
+ .build();
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+ DataStream<PartitionAndValue> stream =
+ env.fromSource(source, WatermarkStrategy.noWatermarks(), "testTimestamp");
+
+ // Verify that the timestamp and watermark are working fine.
+ stream.transform(
+ "timestampVerifier",
+ TypeInformation.of(PartitionAndValue.class),
+ new WatermarkVerifyingOperator(v -> v));
+ stream.addSink(new DiscardingSink<>());
+ JobExecutionResult result = env.execute();
+
+ assertEquals(Arrays.asList(1L, 2L, 3L), result.getAccumulatorResult("timestamp"));
+ }
- @Test
- public void testBasicRead() throws Exception {
- KafkaSource<PartitionAndValue> source =
- KafkaSource.<PartitionAndValue>builder()
- .setBootstrapServers(KafkaSourceTestEnv.brokerConnectionStrings)
- .setGroupId("testBasicRead")
- .setTopics(Arrays.asList(TOPIC1, TOPIC2))
- .setDeserializer(new TestingKafkaRecordDeserializationSchema())
- .setStartingOffsets(OffsetsInitializer.earliest())
- .setBounded(OffsetsInitializer.latest())
- .build();
+ @Test
+ public void testBasicRead() throws Exception {
+ KafkaSource<PartitionAndValue> source =
+ KafkaSource.<PartitionAndValue>builder()
+ .setBootstrapServers(KafkaSourceTestEnv.brokerConnectionStrings)
+ .setGroupId("testBasicRead")
+ .setTopics(Arrays.asList(TOPIC1, TOPIC2))
+ .setDeserializer(new TestingKafkaRecordDeserializationSchema())
+ .setStartingOffsets(OffsetsInitializer.earliest())
+ .setBounded(OffsetsInitializer.latest())
+ .build();
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+ DataStream<PartitionAndValue> stream =
+ env.fromSource(source, WatermarkStrategy.noWatermarks(), "testBasicRead");
+ executeAndVerify(env, stream);
+ }
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.setParallelism(1);
- DataStream<PartitionAndValue> stream =
- env.fromSource(source, WatermarkStrategy.noWatermarks(), "testBasicRead");
- executeAndVerify(env, stream);
- }
+ @Test
+ public void testValueOnlyDeserializer() throws Exception {
+ KafkaSource<Integer> source =
+ KafkaSource.<Integer>builder()
+ .setBootstrapServers(KafkaSourceTestEnv.brokerConnectionStrings)
+ .setGroupId("testValueOnlyDeserializer")
+ .setTopics(Arrays.asList(TOPIC1, TOPIC2))
+ .setDeserializer(
+ KafkaRecordDeserializationSchema.valueOnly(
+ IntegerDeserializer.class))
+ .setStartingOffsets(OffsetsInitializer.earliest())
+ .setBounded(OffsetsInitializer.latest())
+ .build();
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+ final CloseableIterator<Integer> resultIterator =
+ env.fromSource(
+ source,
+ WatermarkStrategy.noWatermarks(),
+ "testValueOnlyDeserializer")
+ .executeAndCollect();
+
+ AtomicInteger actualSum = new AtomicInteger();
+ resultIterator.forEachRemaining(actualSum::addAndGet);
+
+ // Calculate the actual sum of values
+ // Values in a partition should start from partition ID, and end with
+ // (NUM_RECORDS_PER_PARTITION - 1)
+ // e.g. Values in partition 5 should be {5, 6, 7, 8, 9}
+ int expectedSum = 0;
+ for (int partition = 0; partition < KafkaSourceTestEnv.NUM_PARTITIONS; partition++) {
+ for (int value = partition;
+ value < KafkaSourceTestEnv.NUM_RECORDS_PER_PARTITION;
+ value++) {
+ expectedSum += value;
+ }
+ }
- @Test
- public void testValueOnlyDeserializer() throws Exception {
- KafkaSource<Integer> source =
- KafkaSource.<Integer>builder()
- .setBootstrapServers(KafkaSourceTestEnv.brokerConnectionStrings)
- .setGroupId("testValueOnlyDeserializer")
- .setTopics(Arrays.asList(TOPIC1, TOPIC2))
- .setDeserializer(
- KafkaRecordDeserializationSchema.valueOnly(
- IntegerDeserializer.class))
- .setStartingOffsets(OffsetsInitializer.earliest())
- .setBounded(OffsetsInitializer.latest())
- .build();
+ // Since we have two topics, the expected sum value should be doubled
+ expectedSum *= 2;
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.setParallelism(1);
- final CloseableIterator<Integer> resultIterator =
- env.fromSource(
- source,
- WatermarkStrategy.noWatermarks(),
- "testValueOnlyDeserializer")
- .executeAndCollect();
-
- AtomicInteger actualSum = new AtomicInteger();
- resultIterator.forEachRemaining(actualSum::addAndGet);
-
- // Calculate the actual sum of values
- // Values in a partition should start from partition ID, and end with
- // (NUM_RECORDS_PER_PARTITION - 1)
- // e.g. Values in partition 5 should be {5, 6, 7, 8, 9}
- int expectedSum = 0;
- for (int partition = 0; partition < KafkaSourceTestEnv.NUM_PARTITIONS; partition++) {
- for (int value = partition;
- value < KafkaSourceTestEnv.NUM_RECORDS_PER_PARTITION;
- value++) {
- expectedSum += value;
- }
+ assertEquals(expectedSum, actualSum.get());
}
- // Since we have two topics, the expected sum value should be doubled
- expectedSum *= 2;
-
- assertEquals(expectedSum, actualSum.get());
+ @Test
+ public void testRedundantParallelism() throws Exception {
+ KafkaSource<PartitionAndValue> source =
+ KafkaSource.<PartitionAndValue>builder()
+ .setBootstrapServers(KafkaSourceTestEnv.brokerConnectionStrings)
+ .setGroupId("testRedundantParallelism")
+ .setTopics(Collections.singletonList(TOPIC1))
+ .setDeserializer(new TestingKafkaRecordDeserializationSchema())
+ .setStartingOffsets(OffsetsInitializer.earliest())
+ .setBounded(OffsetsInitializer.latest())
+ .build();
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ // Here we use (NUM_PARTITION + 1) as the parallelism, so one SourceReader will not be
+ // assigned with any splits. The redundant SourceReader should also be signaled with a
+ // NoMoreSplitsEvent and eventually spins to FINISHED state.
+ env.setParallelism(KafkaSourceTestEnv.NUM_PARTITIONS + 1);
+ DataStream<PartitionAndValue> stream =
+ env.fromSource(
+ source, WatermarkStrategy.noWatermarks(), "testRedundantParallelism");
+ executeAndVerify(env, stream);
+ }
}
- @Test(timeout = 30000L)
- public void testRedundantParallelism() throws Exception {
- KafkaSource<PartitionAndValue> source =
- KafkaSource.<PartitionAndValue>builder()
- .setBootstrapServers(KafkaSourceTestEnv.brokerConnectionStrings)
- .setGroupId("testRedundantParallelism")
- .setTopics(Collections.singletonList(TOPIC1))
- .setDeserializer(new TestingKafkaRecordDeserializationSchema())
- .setStartingOffsets(OffsetsInitializer.earliest())
- .setBounded(OffsetsInitializer.latest())
+ /** Integration test based on connector testing framework. */
+ @Nested
+ class IntegrationTests extends SourceTestSuiteBase<String> {
+ private static final String KAFKA_IMAGE_NAME = "confluentinc/cp-kafka:5.5.2";
+
+ // Defines test environment on Flink MiniCluster
+ @SuppressWarnings("unused")
+ @TestEnv
+ MiniClusterTestEnvironment flink = new MiniClusterTestEnvironment();
+
+ // Defines external system
+ @ExternalSystem
+ DefaultContainerizedExternalSystem<KafkaContainer> kafka =
+ DefaultContainerizedExternalSystem.builder()
+ .fromContainer(new KafkaContainer(DockerImageName.parse(KAFKA_IMAGE_NAME)))
.build();
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-
- // Here we use (NUM_PARTITION + 1) as the parallelism, so one SourceReader will not be
- // assigned with any splits. The redundant SourceReader should also be signaled with a
- // NoMoreSplitsEvent and eventually spins to FINISHED state.
- env.setParallelism(KafkaSourceTestEnv.NUM_PARTITIONS + 1);
- DataStream<PartitionAndValue> stream =
- env.fromSource(
- source, WatermarkStrategy.noWatermarks(), "testRedundantParallelism");
- executeAndVerify(env, stream);
+ // Defines 2 External context Factories, so test cases will be invoked twice using these two
+ // kinds of external contexts.
+ @SuppressWarnings("unused")
+ @ExternalContextFactory
+ KafkaSingleTopicExternalContext.Factory singleTopic =
+ new KafkaSingleTopicExternalContext.Factory(kafka.getContainer());
+
+ @SuppressWarnings("unused")
+ @ExternalContextFactory
+ KafkaMultipleTopicExternalContext.Factory multipleTopic =
+ new KafkaMultipleTopicExternalContext.Factory(kafka.getContainer());
}
// -----------------
@@ -250,24 +298,23 @@ public class KafkaSourceITCase {
}
@Override
- public void processElement(StreamRecord<PartitionAndValue> element) throws Exception {
+ public void processElement(StreamRecord<PartitionAndValue> element) {
getRuntimeContext().getAccumulator("timestamp").add(element.getTimestamp());
}
}
- @SuppressWarnings("serial")
private void executeAndVerify(
StreamExecutionEnvironment env, DataStream<PartitionAndValue> stream) throws Exception {
stream.addSink(
new RichSinkFunction<PartitionAndValue>() {
@Override
- public void open(Configuration parameters) throws Exception {
+ public void open(Configuration parameters) {
getRuntimeContext()
.addAccumulator("result", new ListAccumulator<PartitionAndValue>());
}
@Override
- public void invoke(PartitionAndValue value, Context context) throws Exception {
+ public void invoke(PartitionAndValue value, Context context) {
getRuntimeContext().getAccumulator("result").add(value);
}
});
@@ -283,10 +330,10 @@ public class KafkaSourceITCase {
int firstExpectedValue = Integer.parseInt(tp.substring(tp.indexOf('-') + 1));
for (int i = 0; i < values.size(); i++) {
assertEquals(
- String.format(
- "The %d-th value for partition %s should be %d", i, tp, i),
firstExpectedValue + i,
- (int) values.get(i));
+ (int) values.get(i),
+ String.format(
+ "The %d-th value for partition %s should be %d", i, tp, i));
}
});
}
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaEnumeratorTest.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaEnumeratorTest.java
index 54e5e47..423584f 100644
--- a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaEnumeratorTest.java
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaEnumeratorTest.java
@@ -21,11 +21,11 @@ package org.apache.flink.connector.kafka.source.enumerator;
import org.apache.flink.api.connector.source.ReaderInfo;
import org.apache.flink.api.connector.source.mocks.MockSplitEnumeratorContext;
import org.apache.flink.connector.kafka.source.KafkaSourceOptions;
-import org.apache.flink.connector.kafka.source.KafkaSourceTestEnv;
import org.apache.flink.connector.kafka.source.enumerator.initializer.NoStoppingOffsetsInitializer;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.connector.kafka.source.enumerator.subscriber.KafkaSubscriber;
import org.apache.flink.connector.kafka.source.split.KafkaPartitionSplit;
+import org.apache.flink.connector.kafka.source.testutils.KafkaSourceTestEnv;
import org.apache.flink.mock.Whitebox;
import org.apache.kafka.clients.admin.AdminClient;
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/initializer/OffsetsInitializerTest.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/initializer/OffsetsInitializerTest.java
index 0e84882..4bc9769 100644
--- a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/initializer/OffsetsInitializerTest.java
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/initializer/OffsetsInitializerTest.java
@@ -18,9 +18,9 @@
package org.apache.flink.connector.kafka.source.enumerator.initializer;
-import org.apache.flink.connector.kafka.source.KafkaSourceTestEnv;
import org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumerator;
import org.apache.flink.connector.kafka.source.split.KafkaPartitionSplit;
+import org.apache.flink.connector.kafka.source.testutils.KafkaSourceTestEnv;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/subscriber/KafkaSubscriberTest.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/subscriber/KafkaSubscriberTest.java
index 0fd9f1a..fc0f7de 100644
--- a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/subscriber/KafkaSubscriberTest.java
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/subscriber/KafkaSubscriberTest.java
@@ -18,7 +18,7 @@
package org.apache.flink.connector.kafka.source.enumerator.subscriber;
-import org.apache.flink.connector.kafka.source.KafkaSourceTestEnv;
+import org.apache.flink.connector.kafka.source.testutils.KafkaSourceTestEnv;
import org.apache.flink.util.ExceptionUtils;
import org.apache.kafka.clients.admin.AdminClient;
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReaderTest.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReaderTest.java
index 2a477dd..175c6ef 100644
--- a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReaderTest.java
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReaderTest.java
@@ -22,10 +22,10 @@ import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
import org.apache.flink.connector.base.source.reader.splitreader.SplitsAddition;
import org.apache.flink.connector.base.source.reader.splitreader.SplitsChange;
-import org.apache.flink.connector.kafka.source.KafkaSourceTestEnv;
import org.apache.flink.connector.kafka.source.metrics.KafkaSourceReaderMetrics;
import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema;
import org.apache.flink.connector.kafka.source.split.KafkaPartitionSplit;
+import org.apache.flink.connector.kafka.source.testutils.KafkaSourceTestEnv;
import org.apache.flink.connector.testutils.source.deserialization.TestingDeserializationContext;
import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java
index 18a024b..037c080 100644
--- a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java
@@ -25,10 +25,10 @@ import org.apache.flink.api.connector.source.SourceReaderContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.KafkaSourceBuilder;
-import org.apache.flink.connector.kafka.source.KafkaSourceTestEnv;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema;
import org.apache.flink.connector.kafka.source.split.KafkaPartitionSplit;
+import org.apache.flink.connector.kafka.source.testutils.KafkaSourceTestEnv;
import org.apache.flink.connector.testutils.source.reader.SourceReaderTestBase;
import org.apache.flink.connector.testutils.source.reader.TestingReaderContext;
import org.apache.flink.connector.testutils.source.reader.TestingReaderOutput;
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/testutils/KafkaMultipleTopicExternalContext.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/testutils/KafkaMultipleTopicExternalContext.java
new file mode 100644
index 0000000..1548a3e
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/testutils/KafkaMultipleTopicExternalContext.java
@@ -0,0 +1,124 @@
+/*
+ * 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.connector.kafka.source.testutils;
+
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.connector.kafka.source.KafkaSource;
+import org.apache.flink.connector.kafka.source.KafkaSourceBuilder;
+import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
+import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema;
+import org.apache.flink.connectors.test.common.external.ExternalContext;
+import org.apache.flink.connectors.test.common.external.SourceSplitDataWriter;
+
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.testcontainers.containers.KafkaContainer;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.regex.Pattern;
+
+/**
+ * Kafka external context that will create multiple topics with only one partitions as source
+ * splits.
+ */
+public class KafkaMultipleTopicExternalContext extends KafkaSingleTopicExternalContext {
+
+ private int numTopics = 0;
+
+ private final String topicPattern;
+
+ private final Map<String, SourceSplitDataWriter<String>> topicNameToSplitWriters =
+ new HashMap<>();
+
+ public KafkaMultipleTopicExternalContext(String bootstrapServers) {
+ super(bootstrapServers);
+ this.topicPattern =
+ "kafka-multiple-topic-[0-9]+-"
+ + ThreadLocalRandom.current().nextLong(Long.MAX_VALUE);
+ }
+
+ @Override
+ public SourceSplitDataWriter<String> createSourceSplitDataWriter() {
+ String topicName = getTopicName();
+ createTopic(topicName, 1, (short) 1);
+ final KafkaPartitionDataWriter splitWriter =
+ new KafkaPartitionDataWriter(
+ getKafkaProducerProperties(numTopics), new TopicPartition(topicName, 0));
+ topicNameToSplitWriters.put(topicName, splitWriter);
+ numTopics++;
+ return splitWriter;
+ }
+
+ @Override
+ public Source<String, ?, ?> createSource(Boundedness boundedness) {
+ KafkaSourceBuilder<String> builder = KafkaSource.builder();
+
+ if (boundedness == Boundedness.BOUNDED) {
+ builder = builder.setBounded(OffsetsInitializer.latest());
+ }
+
+ return builder.setGroupId("flink-kafka-multiple-topic-test")
+ .setBootstrapServers(bootstrapServers)
+ .setTopicPattern(Pattern.compile(topicPattern))
+ .setDeserializer(
+ KafkaRecordDeserializationSchema.valueOnly(StringDeserializer.class))
+ .build();
+ }
+
+ @Override
+ public void close() {
+ topicNameToSplitWriters.forEach(
+ (topicName, splitWriter) -> {
+ try {
+ splitWriter.close();
+ deleteTopic(topicName);
+ } catch (Exception e) {
+ kafkaAdminClient.close();
+ throw new RuntimeException("Cannot close split writer", e);
+ }
+ });
+ topicNameToSplitWriters.clear();
+ kafkaAdminClient.close();
+ }
+
+ private String getTopicName() {
+ return topicPattern.replace("[0-9]+", String.valueOf(numTopics));
+ }
+
+ @Override
+ public String toString() {
+ return "Multiple-topics Kafka";
+ }
+
+ /** Factory of {@link KafkaSingleTopicExternalContext}. */
+ public static class Factory extends KafkaSingleTopicExternalContext.Factory {
+
+ public Factory(KafkaContainer kafkaContainer) {
+ super(kafkaContainer);
+ }
+
+ @Override
+ public ExternalContext<String> createExternalContext() {
+ return new KafkaMultipleTopicExternalContext(getBootstrapServer());
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/testutils/KafkaPartitionDataWriter.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/testutils/KafkaPartitionDataWriter.java
new file mode 100644
index 0000000..5485137
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/testutils/KafkaPartitionDataWriter.java
@@ -0,0 +1,60 @@
+/*
+ * 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.connector.kafka.source.testutils;
+
+import org.apache.flink.connectors.test.common.external.SourceSplitDataWriter;
+
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.TopicPartition;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Collection;
+import java.util.Properties;
+
+/** Source split data writer for writing test data into Kafka topic partitions. */
+public class KafkaPartitionDataWriter implements SourceSplitDataWriter<String> {
+
+ private final KafkaProducer<byte[], byte[]> kafkaProducer;
+ private final TopicPartition topicPartition;
+
+ public KafkaPartitionDataWriter(Properties producerProperties, TopicPartition topicPartition) {
+ this.kafkaProducer = new KafkaProducer<>(producerProperties);
+ this.topicPartition = topicPartition;
+ }
+
+ @Override
+ public void writeRecords(Collection<String> records) {
+ for (String record : records) {
+ ProducerRecord<byte[], byte[]> producerRecord =
+ new ProducerRecord<>(
+ topicPartition.topic(),
+ topicPartition.partition(),
+ null,
+ record.getBytes(StandardCharsets.UTF_8));
+ kafkaProducer.send(producerRecord);
+ }
+ kafkaProducer.flush();
+ }
+
+ @Override
+ public void close() {
+ kafkaProducer.close();
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/testutils/KafkaSingleTopicExternalContext.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/testutils/KafkaSingleTopicExternalContext.java
new file mode 100644
index 0000000..81240cf
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/testutils/KafkaSingleTopicExternalContext.java
@@ -0,0 +1,242 @@
+/*
+ * 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.connector.kafka.source.testutils;
+
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.connector.kafka.source.KafkaSource;
+import org.apache.flink.connector.kafka.source.KafkaSourceBuilder;
+import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
+import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema;
+import org.apache.flink.connectors.test.common.external.ExternalContext;
+import org.apache.flink.connectors.test.common.external.SourceSplitDataWriter;
+
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.NewPartitions;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.KafkaContainer;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Random;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/**
+ * A Kafka external context that will create only one topic and use partitions in that topic as
+ * source splits.
+ */
+public class KafkaSingleTopicExternalContext implements ExternalContext<String> {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(KafkaSingleTopicExternalContext.class);
+
+ private static final String TOPIC_NAME_PREFIX = "kafka-single-topic";
+ private static final int DEFAULT_TIMEOUT = 30;
+ private static final int NUM_RECORDS_UPPER_BOUND = 500;
+ private static final int NUM_RECORDS_LOWER_BOUND = 100;
+
+ protected String bootstrapServers;
+ private final String topicName;
+
+ private final Map<Integer, SourceSplitDataWriter<String>> partitionToSplitWriter =
+ new HashMap<>();
+
+ private int numSplits = 0;
+
+ protected final AdminClient kafkaAdminClient;
+
+ public KafkaSingleTopicExternalContext(String bootstrapServers) {
+ this.bootstrapServers = bootstrapServers;
+ this.topicName =
+ TOPIC_NAME_PREFIX + "-" + ThreadLocalRandom.current().nextLong(Long.MAX_VALUE);
+ kafkaAdminClient = createAdminClient();
+ }
+
+ protected void createTopic(String topicName, int numPartitions, short replicationFactor) {
+ LOG.debug(
+ "Creating new Kafka topic {} with {} partitions and {} replicas",
+ topicName,
+ numPartitions,
+ replicationFactor);
+ NewTopic newTopic = new NewTopic(topicName, numPartitions, replicationFactor);
+ try {
+ kafkaAdminClient
+ .createTopics(Collections.singletonList(newTopic))
+ .all()
+ .get(DEFAULT_TIMEOUT, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ throw new RuntimeException(String.format("Cannot create topic '%s'", topicName), e);
+ }
+ }
+
+ protected void deleteTopic(String topicName) {
+ LOG.debug("Deleting Kafka topic {}", topicName);
+ try {
+ kafkaAdminClient
+ .deleteTopics(Collections.singletonList(topicName))
+ .all()
+ .get(DEFAULT_TIMEOUT, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ if (ExceptionUtils.getRootCause(e) instanceof UnknownTopicOrPartitionException) {
+ throw new RuntimeException(String.format("Cannot delete topic '%s'", topicName), e);
+ }
+ }
+ }
+
+ private AdminClient createAdminClient() {
+ Properties config = new Properties();
+ config.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+ return AdminClient.create(config);
+ }
+
+ @Override
+ public Source<String, ?, ?> createSource(Boundedness boundedness) {
+ KafkaSourceBuilder<String> builder = KafkaSource.builder();
+
+ if (boundedness == Boundedness.BOUNDED) {
+ builder = builder.setBounded(OffsetsInitializer.latest());
+ }
+ return builder.setGroupId("flink-kafka-test")
+ .setDeserializer(
+ KafkaRecordDeserializationSchema.valueOnly(StringDeserializer.class))
+ .setTopics(topicName)
+ .setBootstrapServers(bootstrapServers)
+ .build();
+ }
+
+ @Override
+ public SourceSplitDataWriter<String> createSourceSplitDataWriter() {
+ if (numSplits == 0) {
+ createTopic(topicName, 1, (short) 1);
+ numSplits++;
+ } else {
+ LOG.debug("Creating new partition for topic {}", topicName);
+ kafkaAdminClient.createPartitions(
+ Collections.singletonMap(topicName, NewPartitions.increaseTo(++numSplits)));
+ }
+ KafkaPartitionDataWriter splitWriter =
+ new KafkaPartitionDataWriter(
+ getKafkaProducerProperties(numSplits - 1),
+ new TopicPartition(topicName, numSplits - 1));
+ partitionToSplitWriter.put(numSplits - 1, splitWriter);
+ return splitWriter;
+ }
+
+ @Override
+ public Collection<String> generateTestData(long seed) {
+ Random random = new Random(seed);
+ List<String> randomStringRecords = new ArrayList<>();
+ int recordNum =
+ random.nextInt(NUM_RECORDS_UPPER_BOUND - NUM_RECORDS_LOWER_BOUND)
+ + NUM_RECORDS_LOWER_BOUND;
+ for (int i = 0; i < recordNum; i++) {
+ int stringLength = random.nextInt(50) + 1;
+ randomStringRecords.add(generateRandomString(stringLength, random));
+ }
+ return randomStringRecords;
+ }
+
+ private String generateRandomString(int length, Random random) {
+ String alphaNumericString =
+ "ABCDEFGHIJKLMNOPQRSTUVWXYZ" + "abcdefghijklmnopqrstuvwxyz" + "0123456789";
+ StringBuilder sb = new StringBuilder();
+ for (int i = 0; i < length; ++i) {
+ sb.append(alphaNumericString.charAt(random.nextInt(alphaNumericString.length())));
+ }
+ return sb.toString();
+ }
+
+ protected Properties getKafkaProducerProperties(int producerId) {
+ Properties kafkaProducerProperties = new Properties();
+ kafkaProducerProperties.setProperty(
+ ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+ kafkaProducerProperties.setProperty(
+ ProducerConfig.CLIENT_ID_CONFIG,
+ String.join(
+ "-",
+ "flink-kafka-split-writer",
+ Integer.toString(producerId),
+ Long.toString(ThreadLocalRandom.current().nextLong(Long.MAX_VALUE))));
+ kafkaProducerProperties.setProperty(
+ ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
+ kafkaProducerProperties.setProperty(
+ ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
+ return kafkaProducerProperties;
+ }
+
+ @Override
+ public void close() {
+ deleteTopic(topicName);
+ partitionToSplitWriter.forEach(
+ (partitionId, splitWriter) -> {
+ try {
+ splitWriter.close();
+ } catch (Exception e) {
+ kafkaAdminClient.close();
+ throw new RuntimeException("Cannot close split writer", e);
+ }
+ });
+ partitionToSplitWriter.clear();
+ kafkaAdminClient.close();
+ }
+
+ @Override
+ public String toString() {
+ return "Single-topic Kafka";
+ }
+
+ /** Factory of {@link KafkaSingleTopicExternalContext}. */
+ public static class Factory implements ExternalContext.Factory<String> {
+
+ private final KafkaContainer kafkaContainer;
+
+ public Factory(KafkaContainer kafkaContainer) {
+ this.kafkaContainer = kafkaContainer;
+ }
+
+ protected String getBootstrapServer() {
+ final String internalEndpoints =
+ kafkaContainer.getNetworkAliases().stream()
+ .map(host -> String.join(":", host, Integer.toString(9092)))
+ .collect(Collectors.joining(","));
+ return String.join(",", kafkaContainer.getBootstrapServers(), internalEndpoints);
+ }
+
+ @Override
+ public ExternalContext<String> createExternalContext() {
+ return new KafkaSingleTopicExternalContext(getBootstrapServer());
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceTestEnv.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/testutils/KafkaSourceTestEnv.java
similarity index 99%
rename from flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceTestEnv.java
rename to flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/testutils/KafkaSourceTestEnv.java
index 6262e34..df5d8ea 100644
--- a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceTestEnv.java
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/testutils/KafkaSourceTestEnv.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.flink.connector.kafka.source;
+package org.apache.flink.connector.kafka.source.testutils;
import org.apache.flink.connector.kafka.source.split.KafkaPartitionSplit;
import org.apache.flink.streaming.connectors.kafka.KafkaTestBase;
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/pom.xml b/flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/pom.xml
index 8c3b90d..a13ed11 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/pom.xml
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/pom.xml
@@ -133,6 +133,19 @@ under the License.
<version>1.10.0</version>
<scope>test</scope>
</dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-testing_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ </dependency>
</dependencies>
<build>
@@ -192,6 +205,22 @@ under the License.
<type>jar</type>
<outputDirectory>${project.build.directory}/dependencies</outputDirectory>
</artifactItem>
+ <artifactItem>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <destFileName>kafka-connector.jar</destFileName>
+ <type>jar</type>
+ <outputDirectory>${project.build.directory}/dependencies</outputDirectory>
+ </artifactItem>
+ <artifactItem>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>2.4.1</version>
+ <destFileName>kafka-clients.jar</destFileName>
+ <type>jar</type>
+ <outputDirectory>${project.build.directory}/dependencies</outputDirectory>
+ </artifactItem>
</artifactItems>
</configuration>
</plugin>
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
new file mode 100644
index 0000000..f220008
--- /dev/null
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/KafkaSourceE2ECase.java
@@ -0,0 +1,69 @@
+/*
+ * 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.kafka;
+
+import org.apache.flink.connector.kafka.source.testutils.KafkaMultipleTopicExternalContext;
+import org.apache.flink.connector.kafka.source.testutils.KafkaSingleTopicExternalContext;
+import org.apache.flink.connectors.test.common.external.DefaultContainerizedExternalSystem;
+import org.apache.flink.connectors.test.common.junit.annotations.ExternalContextFactory;
+import org.apache.flink.connectors.test.common.junit.annotations.ExternalSystem;
+import org.apache.flink.connectors.test.common.junit.annotations.TestEnv;
+import org.apache.flink.connectors.test.common.testsuites.SourceTestSuiteBase;
+import org.apache.flink.tests.util.TestUtils;
+import org.apache.flink.tests.util.flink.FlinkContainerTestEnvironment;
+
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.utility.DockerImageName;
+
+/** Kafka E2E test based on connector testing framework. */
+public class KafkaSourceE2ECase extends SourceTestSuiteBase<String> {
+ private static final String KAFKA_HOSTNAME = "kafka";
+ private static final String KAFKA_IMAGE_NAME = "confluentinc/cp-kafka:5.5.2";
+
+ // Defines TestEnvironment
+ @TestEnv
+ FlinkContainerTestEnvironment flink =
+ new FlinkContainerTestEnvironment(
+ 1,
+ 6,
+ TestUtils.getResource("kafka-connector.jar").toAbsolutePath().toString(),
+ TestUtils.getResource("kafka-clients.jar").toAbsolutePath().toString());
+
+ // Defines ConnectorExternalSystem
+ @ExternalSystem
+ DefaultContainerizedExternalSystem<KafkaContainer> kafka =
+ DefaultContainerizedExternalSystem.builder()
+ .fromContainer(
+ new KafkaContainer(DockerImageName.parse(KAFKA_IMAGE_NAME))
+ .withNetworkAliases(KAFKA_HOSTNAME))
+ .bindWithFlinkContainer(flink.getFlinkContainer())
+ .build();
+
+ // Defines 2 External context Factories, so test cases will be invoked twice using these two
+ // kinds of external contexts.
+ @SuppressWarnings("unused")
+ @ExternalContextFactory
+ KafkaSingleTopicExternalContext.Factory singleTopic =
+ new KafkaSingleTopicExternalContext.Factory(kafka.getContainer());
+
+ @SuppressWarnings("unused")
+ @ExternalContextFactory
+ KafkaMultipleTopicExternalContext.Factory multipleTopic =
+ new KafkaMultipleTopicExternalContext.Factory(kafka.getContainer());
+}
diff --git a/tools/ci/java-ci-tools/src/main/resources/modules-skipping-deployment.modulelist b/tools/ci/java-ci-tools/src/main/resources/modules-skipping-deployment.modulelist
index 89e5b58..521d055 100644
--- a/tools/ci/java-ci-tools/src/main/resources/modules-skipping-deployment.modulelist
+++ b/tools/ci/java-ci-tools/src/main/resources/modules-skipping-deployment.modulelist
@@ -38,3 +38,4 @@ flink-streaming-kafka-test-base
flink-heavy-deployment-stress-test
flink-elasticsearch5-test
flink-high-parallelism-iterations-test
+flink-end-to-end-tests-common-kafka
[flink] 03/07: [FLINK-19554][testutil/container] Support TM
restarting in FlinkContainer and network accessibility from other
containers
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 43066fe4afb6ddaf464c7051955cc2b857b13f93
Author: Qingsheng Ren <re...@gmail.com>
AuthorDate: Thu Aug 12 17:43:21 2021 +0800
[FLINK-19554][testutil/container] Support TM restarting in FlinkContainer and network accessibility from other containers
---
.../flink/tests/util/flink/FlinkContainer.java | 99 ++++++++++++++++++++--
1 file changed, 91 insertions(+), 8 deletions(-)
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/FlinkContainer.java b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/FlinkContainer.java
index bcb71b2..3034bf2 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/FlinkContainer.java
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/FlinkContainer.java
@@ -18,6 +18,8 @@
package org.apache.flink.tests.util.flink;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagersInfo;
import org.apache.flink.tests.util.parameters.ParameterProperty;
import org.apache.flink.tests.util.util.FileUtils;
@@ -35,6 +37,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.DockerClientFactory;
import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.Network;
import org.testcontainers.containers.wait.strategy.HttpWaitStrategy;
import org.testcontainers.images.builder.ImageFromDockerfile;
import org.testcontainers.lifecycle.TestDescription;
@@ -64,21 +67,27 @@ import java.util.stream.IntStream;
* container.
*/
public class FlinkContainer extends GenericContainer<FlinkContainer> implements TestLifecycleAware {
+
private static final Logger LOG = LoggerFactory.getLogger(FlinkContainer.class);
private static final ObjectMapper objectMapper = new ObjectMapper();
private static final String FLINK_BIN = "flink/bin";
+ public static final int JOB_MANAGER_REST_PORT = 8081;
private final TemporaryFolder temporaryFolder = new TemporaryFolder();
private final Path logBackupDir;
+ private final int numTaskManagers;
private FlinkContainer(
ImageFromDockerfile image, int numTaskManagers, @Nullable Path logBackupDir) {
super(image);
this.logBackupDir = logBackupDir;
- withExposedPorts(8081);
+ this.numTaskManagers = numTaskManagers;
+ withExposedPorts(JOB_MANAGER_REST_PORT);
+ // Create a network for connecting with other containers
+ withNetwork(Network.newNetwork());
waitingFor(
new HttpWaitStrategy()
- .forPort(8081)
+ .forPort(JOB_MANAGER_REST_PORT)
.forPath("/taskmanagers")
.forResponsePredicate(
response -> {
@@ -107,8 +116,7 @@ public class FlinkContainer extends GenericContainer<FlinkContainer> implements
public void afterTest(TestDescription description, Optional<Throwable> throwable) {
if (throwable.isPresent() && logBackupDir != null) {
try {
- final Path targetDirectory =
- logBackupDir.resolve("flink-" + UUID.randomUUID().toString());
+ final Path targetDirectory = logBackupDir.resolve("flink-" + UUID.randomUUID());
copyFileOrDirectoryFromContainer("flink/log/", targetDirectory);
LOG.info("Backed up logs to {}.", targetDirectory);
} catch (IOException e) {
@@ -136,7 +144,7 @@ public class FlinkContainer extends GenericContainer<FlinkContainer> implements
}
private static void unTar(TarArchiveInputStream tis, File destFolder) throws IOException {
- TarArchiveEntry entry = null;
+ TarArchiveEntry entry;
while ((entry = tis.getNextTarEntry()) != null) {
FileOutputStream fos = null;
try {
@@ -192,6 +200,38 @@ public class FlinkContainer extends GenericContainer<FlinkContainer> implements
}
}
+ /**
+ * Restart all task managers.
+ *
+ * @param afterFailAction Action to do between stopping and restarting task managers
+ * @throws Exception If anything wrong happens during the restart
+ */
+ public void restartTaskManager(Runnable afterFailAction) throws Exception {
+ final String[] stopCommand = {FLINK_BIN + "/taskmanager.sh", "stop-all"};
+ final ExecResult stopResult = execInContainer(stopCommand);
+ checkExitCode(stopResult, stopCommand);
+
+ afterFailAction.run();
+
+ for (int i = 0; i < numTaskManagers; i++) {
+ final String[] startCommand = {FLINK_BIN + "/taskmanager.sh", "start"};
+ final ExecResult startResult = execInContainer(startCommand);
+ checkExitCode(startResult, startCommand);
+ }
+ }
+
+ public void checkExitCode(ExecResult execResult, String... command) {
+ if (execResult.getExitCode() != 0) {
+ throw new IllegalStateException(
+ String.format(
+ "Command \"%s\" exited with code %d. \nSTDOUT: %s\nSTDERR: %s",
+ String.join(" ", command),
+ execResult.getExitCode(),
+ execResult.getStdout(),
+ execResult.getStderr()));
+ }
+ }
+
@Nonnull
private String copyAndGetContainerPath(String defaultEnvFile) {
Path path = Paths.get(defaultEnvFile);
@@ -211,6 +251,7 @@ public class FlinkContainer extends GenericContainer<FlinkContainer> implements
new ParameterProperty<>("logBackupDir", Paths::get);
private int numTaskManagers = 1;
+ private Configuration flinkConfiguration;
private String javaVersion;
private final TemporaryFolder temporaryFolder = new TemporaryFolder();
@@ -224,6 +265,17 @@ public class FlinkContainer extends GenericContainer<FlinkContainer> implements
}
/**
+ * Additional Flink configurations for the cluster.
+ *
+ * @param flinkConfiguration Additional Flink configurations
+ * @return Builder itself
+ */
+ public FlinkContainerBuilder withFlinkConfiguration(Configuration flinkConfiguration) {
+ this.flinkConfiguration = flinkConfiguration;
+ return this;
+ }
+
+ /**
* Specifies which OpenJDK version to use. If not provided explicitly, the image version
* will be derived based on the version of the java that runs the test.
*/
@@ -235,8 +287,23 @@ public class FlinkContainer extends GenericContainer<FlinkContainer> implements
public FlinkContainer build() {
try {
Path flinkDist = FileUtils.findFlinkDist();
+ Path flinkConfDirectory = flinkDist.resolve("conf");
+
+ // Load Flink configurations
+ final Configuration mergedFlinkConfiguration =
+ GlobalConfiguration.loadConfiguration(
+ flinkConfDirectory.toAbsolutePath().toString());
+
+ // Merge additional Flink configurations
+ if (flinkConfiguration != null) {
+ mergedFlinkConfiguration.addAll(flinkConfiguration);
+ }
+
+ // Create temporary folder for holding configuration files
temporaryFolder.create();
Path tmp = temporaryFolder.newFolder().toPath();
+
+ // Write worker file
Path workersFile = tmp.resolve("workers");
Files.write(
workersFile,
@@ -244,6 +311,14 @@ public class FlinkContainer extends GenericContainer<FlinkContainer> implements
.mapToObj(i -> "localhost")
.collect(Collectors.toList()));
+ // Write merged Flink configuration
+ Path configurationFile = tmp.resolve(GlobalConfiguration.FLINK_CONF_FILENAME);
+ final List<String> configurationLines =
+ mergedFlinkConfiguration.toMap().entrySet().stream()
+ .map(entry -> entry.getKey() + ": " + entry.getValue())
+ .collect(Collectors.toList());
+ Files.write(configurationFile, configurationLines);
+
// Building the docker image is split into two stages:
// 1. build a base image with an immutable flink-dist
// 2. based on the base image add any mutable files such as e.g. workers files
@@ -251,7 +326,8 @@ public class FlinkContainer extends GenericContainer<FlinkContainer> implements
// This lets us save some time for archiving and copying big, immutable files
// between tests runs.
String baseImage = buildBaseImage(flinkDist);
- ImageFromDockerfile configuredImage = buildConfiguredImage(workersFile, baseImage);
+ ImageFromDockerfile configuredImage =
+ buildConfiguredImage(workersFile, configurationFile, baseImage);
Optional<Path> logBackupDirectory = DISTRIBUTION_LOG_BACKUP_DIRECTORY.get();
if (!logBackupDirectory.isPresent()) {
@@ -267,17 +343,24 @@ public class FlinkContainer extends GenericContainer<FlinkContainer> implements
}
}
- private ImageFromDockerfile buildConfiguredImage(Path workersFile, String baseImage) {
+ private ImageFromDockerfile buildConfiguredImage(
+ Path workersFile, Path configurationFile, String baseImage) {
return new ImageFromDockerfile("flink-dist-configured")
.withDockerfileFromBuilder(
builder ->
builder.from(baseImage)
.copy("workers", "flink/conf/workers")
+ .copy(
+ GlobalConfiguration.FLINK_CONF_FILENAME,
+ "flink/conf/"
+ + GlobalConfiguration
+ .FLINK_CONF_FILENAME)
.cmd(
FLINK_BIN
+ "/start-cluster.sh && tail -f /dev/null")
.build())
- .withFileFromPath("workers", workersFile);
+ .withFileFromPath("workers", workersFile)
+ .withFileFromPath(GlobalConfiguration.FLINK_CONF_FILENAME, configurationFile);
}
@Nonnull