You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ma...@apache.org on 2022/12/12 18:33:05 UTC
[flink-connector-pulsar] branch main updated (3199922 -> ded956e)
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
from 3199922 [FLINK-30254][Connector/Pulsar] Resolve dependency convergence errors
new c613f4b [FLINK-23969][connector/pulsar] Create e2e tests for pulsar connector.
new db11068 [FLINK-24018][build] Remove Scala dependencies from Java APIs
new 52cbdb5 [hotfix][connector/testing-framework] Let ExternalContext#generateTestData returns List to preserve order
new 8de314d [FLINK-25038][testutils] Refactor FlinkContainer to split JM and TMs to individual containers and supports HA
new cacdffb [FLINK-25159][tests] Disable broken E2E tests
new 9046e81 [FLINK-25210][pulsar][e2e][tests] add resource file to test jar
new 082bd2f [FLINK-25287][connectors/testing-framework] Refactor connector testing framework interfaces for more scenarios
new e94e54b [FLINK-25287][connectors/pulsar] Use connector testing framework interface for Pulsar tests
new 4f85230 [hotfix][connectors] Improve the typo and code style
new 87b9f1b [FLINK-25840][tests] Add semantics test support for connector test framework
new 6971aa1 [FLINK-25288][tests][pulsar] Make pulsar e2e tests more robust
new 704ef1f [FLINK-24246][connector/pulsar] Bump PulsarClient version to latest 2.9.1
new 9c98b83 [FLINK-26025][connector/pulsar] Replace MockPulsar with new Pulsar test tools based on PulsarStandalone.
new 39169fa [hotfix][pulsar] Disable tests
new 44d53a7 [FLINK-26210][pulsar][tests] Add jaxb-api to e2e test
new bca31db Update version to 1.16-SNAPSHOT
new 356cf23 [FLINK-27854][tests] Add support for FlinkContainers to work with both existing Docker images and flink-dist builds
new 65ae7f7 [FLINK-27199][Connector/Pulsar] Bump the pulsar-client-all to latest 2.10.0.
new 9714ebb [FLINK-28409][tests] Move FlinkContainerTestEnvironment and dependendencies into flink-connector-test-utils
new f4a7fa0 Update version to 1.17-SNAPSHOT
new f333cf2 [FLINK-29285][tests] Move TestUtils#getResource
new 712f647 [FLINK-26182][Connector/pulsar] Extract common logic from Pulsar source testing tools.
new 0ce6280 [FLINK-26182][Connector/pulsar] Create e2e tests for the Pulsar source and sink based on the connector testing framework.
new 79e968d [FLINK-29532][Connector/Pulsar] Update Pulsar dependency to 2.10.1. This closes #20980
new 1b1aa9a [FLINK-29495][Connector/Pulsar] Refactor Pulsar tests from JUnit4 annotation to JUnit5 annotation for disabling tests on Java 11
new 9004447 [FLINK-29709][Connector/Pulsar] Bump the Pulsar to latest 2.10.2 (#21204)
new bf7af29 [FLINK-26027][Connector/Pulsar] Expose Pulsar producer metrics and add FLIP-33 sink metrics. (#21249)
new 858931e [FLINK-30336][Connector/Pulsar] Sync missing Pulsar e2e tests to external repo
new 7360e65 [hotfix] Increase timeout for compile and test runs
new e35b5e7 [hotfix] Set version to 4.0-SNAPSHOT
new ded956e [hotfix][architecture] Add ProductionCodeArchitectureTest
The 31 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:
.github/workflows/ci.yml | 4 +-
flink-connector-pulsar-e2e-tests/pom.xml | 258 +++++++++++++++++++++
.../flink/tests/util/pulsar/PulsarSinkE2ECase.java | 56 +++++
.../util/pulsar/PulsarSourceOrderedE2ECase.java | 45 ++--
.../util/pulsar/PulsarSourceUnorderedE2ECase.java | 35 +--
.../FlinkContainerWithPulsarEnvironment.java | 65 ++++++
.../common/PulsarContainerTestEnvironment.java | 16 +-
.../source/ExclusiveSubscriptionContext.java | 15 +-
.../pulsar/source/FailoverSubscriptionContext.java | 15 +-
.../src/test/resources/log4j2-test.properties | 24 +-
flink-connector-pulsar/pom.xml | 3 +-
...st.java => ProductionCodeArchitectureTest.java} | 9 +-
flink-sql-connector-pulsar/pom.xml | 2 +-
pom.xml | 4 +-
14 files changed, 467 insertions(+), 84 deletions(-)
create mode 100644 flink-connector-pulsar-e2e-tests/pom.xml
create mode 100644 flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSinkE2ECase.java
copy flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarUnorderedSourceITCase.java => flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java (52%)
copy flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarUnorderedSourceITCase.java => flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java (71%)
create mode 100644 flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
copy flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/stop/NeverStopCursor.java => flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/PulsarContainerTestEnvironment.java (59%)
copy flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/SharedSubscriptionContext.java => flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/source/ExclusiveSubscriptionContext.java (72%)
copy flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/SharedSubscriptionContext.java => flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/source/FailoverSubscriptionContext.java (72%)
copy .github/workflows/weekly.yml => flink-connector-pulsar-e2e-tests/src/test/resources/log4j2-test.properties (70%)
copy flink-connector-pulsar/src/test/java/org/apache/flink/architecture/{TestCodeArchitectureTest.java => ProductionCodeArchitectureTest.java} (81%)
[flink-connector-pulsar] 24/31: [FLINK-29532][Connector/Pulsar] Update Pulsar dependency to 2.10.1. This closes #20980
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit 79e968d7f62b53a5e72ad36a8e1ff8002adf286d
Author: MartijnVisser <ma...@apache.org>
AuthorDate: Mon Oct 10 10:01:27 2022 +0200
[FLINK-29532][Connector/Pulsar] Update Pulsar dependency to 2.10.1. This closes #20980
* [FLINK-29532][Connector/Pulsar] Update Pulsar dependency to 2.10.1
* [FLINK-29532][Connector/Pulsar] Fix NOTICE files
* Addressed review comments
---
flink-connector-pulsar-e2e-tests/pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/flink-connector-pulsar-e2e-tests/pom.xml b/flink-connector-pulsar-e2e-tests/pom.xml
index 6e6f7d3..5ba8508 100644
--- a/flink-connector-pulsar-e2e-tests/pom.xml
+++ b/flink-connector-pulsar-e2e-tests/pom.xml
@@ -31,7 +31,7 @@ under the License.
<name>Flink : E2E Tests : Pulsar</name>
<properties>
- <pulsar.version>2.10.0</pulsar.version>
+ <pulsar.version>2.10.1</pulsar.version>
<bouncycastle.version>1.69</bouncycastle.version>
<jaxb-api.version>2.3.1</jaxb-api.version>
</properties>
[flink-connector-pulsar] 14/31: [hotfix][pulsar] Disable tests
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit 39169fade9df3bcc2ce975410ac4ac17ed2e4a93
Author: Chesnay Schepler <ch...@apache.org>
AuthorDate: Fri Feb 18 08:54:32 2022 +0100
[hotfix][pulsar] Disable tests
---
flink-connector-pulsar-e2e-tests/pom.xml | 7 +++++++
1 file changed, 7 insertions(+)
diff --git a/flink-connector-pulsar-e2e-tests/pom.xml b/flink-connector-pulsar-e2e-tests/pom.xml
index 7c87ec7..ab47a08 100644
--- a/flink-connector-pulsar-e2e-tests/pom.xml
+++ b/flink-connector-pulsar-e2e-tests/pom.xml
@@ -60,6 +60,13 @@ under the License.
<build>
<plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <skip>true</skip>
+ </configuration>
+ </plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-dependency-plugin</artifactId>
[flink-connector-pulsar] 30/31: [hotfix] Set version to 4.0-SNAPSHOT
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit e35b5e746efd1c8601972c08c862e2b6daa18c0d
Author: Martijn Visser <ma...@apache.org>
AuthorDate: Fri Dec 9 12:55:30 2022 +0100
[hotfix] Set version to 4.0-SNAPSHOT
---
flink-connector-pulsar-e2e-tests/pom.xml | 2 +-
flink-connector-pulsar/pom.xml | 2 +-
flink-sql-connector-pulsar/pom.xml | 2 +-
pom.xml | 2 +-
4 files changed, 4 insertions(+), 4 deletions(-)
diff --git a/flink-connector-pulsar-e2e-tests/pom.xml b/flink-connector-pulsar-e2e-tests/pom.xml
index 7a6e6ce..5798d1e 100644
--- a/flink-connector-pulsar-e2e-tests/pom.xml
+++ b/flink-connector-pulsar-e2e-tests/pom.xml
@@ -23,7 +23,7 @@ under the License.
<parent>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-pulsar-parent</artifactId>
- <version>3.0-SNAPSHOT</version>
+ <version>4.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/flink-connector-pulsar/pom.xml b/flink-connector-pulsar/pom.xml
index 68eab5a..b23309b 100644
--- a/flink-connector-pulsar/pom.xml
+++ b/flink-connector-pulsar/pom.xml
@@ -26,7 +26,7 @@ under the License.
<parent>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-pulsar-parent</artifactId>
- <version>3.0-SNAPSHOT</version>
+ <version>4.0-SNAPSHOT</version>
</parent>
<artifactId>flink-connector-pulsar</artifactId>
diff --git a/flink-sql-connector-pulsar/pom.xml b/flink-sql-connector-pulsar/pom.xml
index 62b7bb7..87f46e4 100644
--- a/flink-sql-connector-pulsar/pom.xml
+++ b/flink-sql-connector-pulsar/pom.xml
@@ -26,7 +26,7 @@ under the License.
<parent>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-pulsar-parent</artifactId>
- <version>3.0-SNAPSHOT</version>
+ <version>4.0-SNAPSHOT</version>
</parent>
<artifactId>flink-sql-connector-pulsar</artifactId>
diff --git a/pom.xml b/pom.xml
index baf6aac..a2a7af5 100644
--- a/pom.xml
+++ b/pom.xml
@@ -29,7 +29,7 @@ under the License.
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-pulsar-parent</artifactId>
- <version>3.0-SNAPSHOT</version>
+ <version>4.0-SNAPSHOT</version>
<name>Flink : Connectors : Pulsar : Parent</name>
<packaging>pom</packaging>
<inceptionYear>2022</inceptionYear>
[flink-connector-pulsar] 09/31: [hotfix][connectors] Improve the typo and code style
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit 4f85230763e76fe9a399bfc3a9f6a35bec0a8996
Author: Leonard Xu <xb...@gmail.com>
AuthorDate: Sun Jan 23 09:07:51 2022 +0800
[hotfix][connectors] Improve the typo and code style
---
.../flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java | 2 ++
.../flink/tests/util/pulsar/cases/ExclusiveSubscriptionContext.java | 4 ++--
.../flink/tests/util/pulsar/cases/FailoverSubscriptionContext.java | 4 ++--
.../flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java | 6 ++----
.../flink/tests/util/pulsar/cases/SharedSubscriptionContext.java | 2 +-
5 files changed, 9 insertions(+), 9 deletions(-)
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
index e519618..4351147 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
@@ -46,10 +46,12 @@ public class PulsarSourceUnorderedE2ECase extends UnorderedSourceTestSuiteBase<S
new PulsarTestEnvironment(container(flink.getFlinkContainers().getJobManager()));
// Defines a set of external context Factories for different test cases.
+ @SuppressWarnings("unused")
@TestContext
PulsarTestContextFactory<String, SharedSubscriptionContext> shared =
new PulsarTestContextFactory<>(pulsar, SharedSubscriptionContext::new);
+ @SuppressWarnings("unused")
@TestContext
PulsarTestContextFactory<String, KeySharedSubscriptionContext> keyShared =
new PulsarTestContextFactory<>(pulsar, KeySharedSubscriptionContext::new);
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/ExclusiveSubscriptionContext.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/ExclusiveSubscriptionContext.java
index c3c4959..1245e14 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/ExclusiveSubscriptionContext.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/ExclusiveSubscriptionContext.java
@@ -30,9 +30,9 @@ import java.util.List;
import static org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerRuntime.PULSAR_ADMIN_URL;
import static org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerRuntime.PULSAR_SERVICE_URL;
-/** We would consuming from test splits by using {@link SubscriptionType#Exclusive} subscription. */
+/** We would consume from test splits by using {@link SubscriptionType#Exclusive} subscription. */
public class ExclusiveSubscriptionContext extends MultipleTopicTemplateContext {
- private static final long serialVersionUID = 6238209089442257487L;
+ private static final long serialVersionUID = 1L;
public ExclusiveSubscriptionContext(PulsarTestEnvironment environment) {
this(environment, Collections.emptyList());
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/FailoverSubscriptionContext.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/FailoverSubscriptionContext.java
index 9dbbec8..8ec1685 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/FailoverSubscriptionContext.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/FailoverSubscriptionContext.java
@@ -30,9 +30,9 @@ import java.util.List;
import static org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerRuntime.PULSAR_ADMIN_URL;
import static org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerRuntime.PULSAR_SERVICE_URL;
-/** We would consuming from test splits by using {@link SubscriptionType#Failover} subscription. */
+/** We would consume from test splits by using {@link SubscriptionType#Failover} subscription. */
public class FailoverSubscriptionContext extends MultipleTopicTemplateContext {
- private static final long serialVersionUID = 6238209089442257487L;
+ private static final long serialVersionUID = 1L;
public FailoverSubscriptionContext(PulsarTestEnvironment environment) {
this(environment, Collections.emptyList());
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java
index 0be3ac3..303783a 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java
@@ -50,11 +50,9 @@ import static org.apache.flink.connector.pulsar.testutils.runtime.container.Puls
import static org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerRuntime.PULSAR_SERVICE_URL;
import static org.apache.pulsar.client.api.Schema.STRING;
-/**
- * We would consuming from test splits by using {@link SubscriptionType#Key_Shared} subscription.
- */
+/** We would consume from test splits by using {@link SubscriptionType#Key_Shared} subscription. */
public class KeySharedSubscriptionContext extends PulsarTestContext<String> {
- private static final long serialVersionUID = 3246516520107893983L;
+ private static final long serialVersionUID = 1L;
private int index = 0;
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/SharedSubscriptionContext.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/SharedSubscriptionContext.java
index 5a4ce75..de53595 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/SharedSubscriptionContext.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/SharedSubscriptionContext.java
@@ -46,7 +46,7 @@ import static org.apache.pulsar.client.api.Schema.STRING;
/** We would consuming from test splits by using {@link SubscriptionType#Shared} subscription. */
public class SharedSubscriptionContext extends PulsarTestContext<String> {
- private static final long serialVersionUID = -2798707923661295245L;
+ private static final long serialVersionUID = 1L;
private int index = 0;
[flink-connector-pulsar] 15/31: [FLINK-26210][pulsar][tests] Add jaxb-api to e2e test
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit 44d53a7b790bcd450168649a35e4bf35411ef257
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Mon Feb 21 17:28:21 2022 +0800
[FLINK-26210][pulsar][tests] Add jaxb-api to e2e test
---
flink-connector-pulsar-e2e-tests/pom.xml | 15 +++++++++++++++
.../tests/util/pulsar/PulsarSourceOrderedE2ECase.java | 4 ++++
.../common/FlinkContainerWithPulsarEnvironment.java | 1 +
3 files changed, 20 insertions(+)
diff --git a/flink-connector-pulsar-e2e-tests/pom.xml b/flink-connector-pulsar-e2e-tests/pom.xml
index ab47a08..424b106 100644
--- a/flink-connector-pulsar-e2e-tests/pom.xml
+++ b/flink-connector-pulsar-e2e-tests/pom.xml
@@ -46,6 +46,13 @@ under the License.
<artifactId>flink-connector-pulsar</artifactId>
<version>${project.version}</version>
</dependency>
+ <!-- pulsar-client-all requires jaxb-api for javax.xml.bind.annotation.XmlElement -->
+ <!-- packaged in flink-dist but not provided in e2e environment. -->
+ <dependency>
+ <groupId>javax.xml.bind</groupId>
+ <artifactId>jaxb-api</artifactId>
+ <version>${jaxb.api.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-pulsar</artifactId>
@@ -153,6 +160,14 @@ under the License.
<type>jar</type>
<outputDirectory>${project.build.directory}/dependencies</outputDirectory>
</dependency>
+ <dependency>
+ <groupId>javax.xml.bind</groupId>
+ <artifactId>jaxb-api</artifactId>
+ <version>${jaxb-api.version}</version>
+ <destFileName>jaxb-api.jar</destFileName>
+ <type>jar</type>
+ <outputDirectory>${project.build.directory}/dependencies</outputDirectory>
+ </dependency>
<artifactItem>
<groupId>org.slf4j</groupId>
<artifactId>jul-to-slf4j</artifactId>
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
index 502b41d..234c1a0 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
@@ -29,11 +29,15 @@ import org.apache.flink.tests.util.pulsar.cases.ExclusiveSubscriptionContext;
import org.apache.flink.tests.util.pulsar.cases.FailoverSubscriptionContext;
import org.apache.flink.tests.util.pulsar.common.FlinkContainerWithPulsarEnvironment;
import org.apache.flink.tests.util.pulsar.common.PulsarContainerTestEnvironment;
+import org.apache.flink.testutils.junit.FailsOnJava11;
+
+import org.junit.experimental.categories.Category;
/**
* Pulsar E2E test based on connector testing framework. It's used for Failover & Exclusive
* subscription.
*/
+@Category(value = {FailsOnJava11.class})
public class PulsarSourceOrderedE2ECase extends SourceTestSuiteBase<String> {
// Defines the Semantic.
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
index ccfe277..9a1c6dd 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
@@ -42,6 +42,7 @@ public class FlinkContainerWithPulsarEnvironment extends FlinkContainerTestEnvir
resourcePath("bcprov-jdk15on.jar"),
resourcePath("bcutil-jdk15on.jar"),
resourcePath("bcprov-ext-jdk15on.jar"),
+ resourcePath("jaxb-api.jar"),
resourcePath("jul-to-slf4j.jar"));
}
[flink-connector-pulsar] 01/31: [FLINK-23969][connector/pulsar] Create e2e tests for pulsar connector.
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit c613f4b795e3579aea3706c0c4791ec7181e8478
Author: syhily <sy...@gmail.com>
AuthorDate: Mon Sep 13 21:52:31 2021 +0800
[FLINK-23969][connector/pulsar] Create e2e tests for pulsar connector.
---
flink-connector-pulsar-e2e-tests/pom.xml | 121 +++++++++++++++++++
.../util/pulsar/PulsarSourceOrderedE2ECase.java | 55 +++++++++
.../util/pulsar/PulsarSourceUnorderedE2ECase.java | 55 +++++++++
.../pulsar/cases/ExclusiveSubscriptionContext.java | 61 ++++++++++
.../pulsar/cases/FailoverSubscriptionContext.java | 61 ++++++++++
.../pulsar/cases/KeySharedSubscriptionContext.java | 134 +++++++++++++++++++++
.../pulsar/cases/SharedSubscriptionContext.java | 104 ++++++++++++++++
.../FlinkContainerWithPulsarEnvironment.java | 54 +++++++++
.../common/KeyedPulsarPartitionDataWriter.java | 62 ++++++++++
.../common/UnorderedSourceTestSuiteBase.java | 72 +++++++++++
.../src/test/resources/log4j2-test.properties | 28 +++++
11 files changed, 807 insertions(+)
diff --git a/flink-connector-pulsar-e2e-tests/pom.xml b/flink-connector-pulsar-e2e-tests/pom.xml
new file mode 100644
index 0000000..269f89c
--- /dev/null
+++ b/flink-connector-pulsar-e2e-tests/pom.xml
@@ -0,0 +1,121 @@
+<?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/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>flink-end-to-end-tests</artifactId>
+ <groupId>org.apache.flink</groupId>
+ <version>1.15-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>flink-end-to-end-tests-pulsar</artifactId>
+ <name>Flink : E2E Tests : Pulsar</name>
+
+ <properties>
+ <pulsar.version>2.8.0</pulsar.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-end-to-end-tests-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-pulsar_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-pulsar_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ </dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>pulsar</artifactId>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-dependency-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>copy</id>
+ <phase>pre-integration-test</phase>
+ <goals>
+ <goal>copy</goal>
+ </goals>
+ </execution>
+ </executions>
+ <configuration>
+ <artifactItems>
+ <artifactItem>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-pulsar_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <destFileName>pulsar-connector.jar</destFileName>
+ <type>jar</type>
+ <outputDirectory>${project.build.directory}/dependencies</outputDirectory>
+ </artifactItem>
+ <artifactItem>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar-client-all</artifactId>
+ <version>${pulsar.version}</version>
+ <destFileName>pulsar-client-all.jar</destFileName>
+ <type>jar</type>
+ <outputDirectory>${project.build.directory}/dependencies</outputDirectory>
+ </artifactItem>
+ <artifactItem>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar-client-api</artifactId>
+ <version>${pulsar.version}</version>
+ <destFileName>pulsar-client-api.jar</destFileName>
+ <type>jar</type>
+ <outputDirectory>${project.build.directory}/dependencies</outputDirectory>
+ </artifactItem>
+ <artifactItem>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar-client-admin-api</artifactId>
+ <version>${pulsar.version}</version>
+ <destFileName>pulsar-admin-api.jar</destFileName>
+ <type>jar</type>
+ <outputDirectory>${project.build.directory}/dependencies</outputDirectory>
+ </artifactItem>
+ <artifactItem>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jul-to-slf4j</artifactId>
+ <version>${slf4j.version}</version>
+ <destFileName>jul-to-slf4j.jar</destFileName>
+ <type>jar</type>
+ <outputDirectory>${project.build.directory}/dependencies</outputDirectory>
+ </artifactItem>
+ </artifactItems>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
new file mode 100644
index 0000000..1427c2b
--- /dev/null
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
@@ -0,0 +1,55 @@
+/*
+ * 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.pulsar;
+
+import org.apache.flink.connector.pulsar.testutils.PulsarTestContextFactory;
+import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
+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.pulsar.cases.ExclusiveSubscriptionContext;
+import org.apache.flink.tests.util.pulsar.cases.FailoverSubscriptionContext;
+import org.apache.flink.tests.util.pulsar.common.FlinkContainerWithPulsarEnvironment;
+
+import static org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntime.container;
+
+/**
+ * Pulsar E2E test based on connector testing framework. It's used for Failover & Exclusive
+ * subscription.
+ */
+public class PulsarSourceOrderedE2ECase extends SourceTestSuiteBase<String> {
+
+ // Defines TestEnvironment.
+ @TestEnv
+ FlinkContainerWithPulsarEnvironment flink = new FlinkContainerWithPulsarEnvironment(1, 6);
+
+ // Defines ConnectorExternalSystem.
+ @ExternalSystem
+ PulsarTestEnvironment pulsar = new PulsarTestEnvironment(container(flink.getFlinkContainer()));
+
+ // Defines a set of external context Factories for different test cases.
+ @ExternalContextFactory
+ PulsarTestContextFactory<String, ExclusiveSubscriptionContext> exclusive =
+ new PulsarTestContextFactory<>(pulsar, ExclusiveSubscriptionContext::new);
+
+ @ExternalContextFactory
+ PulsarTestContextFactory<String, FailoverSubscriptionContext> failover =
+ new PulsarTestContextFactory<>(pulsar, FailoverSubscriptionContext::new);
+}
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
new file mode 100644
index 0000000..25cab21
--- /dev/null
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
@@ -0,0 +1,55 @@
+/*
+ * 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.pulsar;
+
+import org.apache.flink.connector.pulsar.testutils.PulsarTestContextFactory;
+import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
+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.tests.util.pulsar.cases.KeySharedSubscriptionContext;
+import org.apache.flink.tests.util.pulsar.cases.SharedSubscriptionContext;
+import org.apache.flink.tests.util.pulsar.common.FlinkContainerWithPulsarEnvironment;
+import org.apache.flink.tests.util.pulsar.common.UnorderedSourceTestSuiteBase;
+
+import static org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntime.container;
+
+/**
+ * Pulsar E2E test based on connector testing framework. It's used for Shared & Key_Shared
+ * subscription.
+ */
+public class PulsarSourceUnorderedE2ECase extends UnorderedSourceTestSuiteBase<String> {
+
+ // Defines TestEnvironment.
+ @TestEnv
+ FlinkContainerWithPulsarEnvironment flink = new FlinkContainerWithPulsarEnvironment(1, 8);
+
+ // Defines ConnectorExternalSystem.
+ @ExternalSystem
+ PulsarTestEnvironment pulsar = new PulsarTestEnvironment(container(flink.getFlinkContainer()));
+
+ // Defines a set of external context Factories for different test cases.
+ @ExternalContextFactory
+ PulsarTestContextFactory<String, SharedSubscriptionContext> shared =
+ new PulsarTestContextFactory<>(pulsar, SharedSubscriptionContext::new);
+
+ @ExternalContextFactory
+ PulsarTestContextFactory<String, KeySharedSubscriptionContext> keyShared =
+ new PulsarTestContextFactory<>(pulsar, KeySharedSubscriptionContext::new);
+}
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/ExclusiveSubscriptionContext.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/ExclusiveSubscriptionContext.java
new file mode 100644
index 0000000..18b2ffc
--- /dev/null
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/ExclusiveSubscriptionContext.java
@@ -0,0 +1,61 @@
+/*
+ * 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.pulsar.cases;
+
+import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
+import org.apache.flink.connector.pulsar.testutils.cases.MultipleTopicTemplateContext;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import static org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerRuntime.PULSAR_ADMIN_URL;
+import static org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerRuntime.PULSAR_SERVICE_URL;
+
+/** We would consuming from test splits by using {@link SubscriptionType#Exclusive} subscription. */
+public class ExclusiveSubscriptionContext extends MultipleTopicTemplateContext {
+ private static final long serialVersionUID = 6238209089442257487L;
+
+ public ExclusiveSubscriptionContext(PulsarTestEnvironment environment) {
+ super(environment);
+ }
+
+ @Override
+ protected String displayName() {
+ return "consuming message by Exclusive";
+ }
+
+ @Override
+ protected String subscriptionName() {
+ return "pulsar-exclusive-subscription";
+ }
+
+ @Override
+ protected SubscriptionType subscriptionType() {
+ return SubscriptionType.Exclusive;
+ }
+
+ @Override
+ protected String serviceUrl() {
+ return PULSAR_SERVICE_URL;
+ }
+
+ @Override
+ protected String adminUrl() {
+ return PULSAR_ADMIN_URL;
+ }
+}
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/FailoverSubscriptionContext.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/FailoverSubscriptionContext.java
new file mode 100644
index 0000000..c322efa
--- /dev/null
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/FailoverSubscriptionContext.java
@@ -0,0 +1,61 @@
+/*
+ * 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.pulsar.cases;
+
+import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
+import org.apache.flink.connector.pulsar.testutils.cases.MultipleTopicTemplateContext;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import static org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerRuntime.PULSAR_ADMIN_URL;
+import static org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerRuntime.PULSAR_SERVICE_URL;
+
+/** We would consuming from test splits by using {@link SubscriptionType#Failover} subscription. */
+public class FailoverSubscriptionContext extends MultipleTopicTemplateContext {
+ private static final long serialVersionUID = 6238209089442257487L;
+
+ public FailoverSubscriptionContext(PulsarTestEnvironment environment) {
+ super(environment);
+ }
+
+ @Override
+ protected String displayName() {
+ return "consuming message by Failover";
+ }
+
+ @Override
+ protected String subscriptionName() {
+ return "pulsar-failover-subscription";
+ }
+
+ @Override
+ protected SubscriptionType subscriptionType() {
+ return SubscriptionType.Failover;
+ }
+
+ @Override
+ protected String serviceUrl() {
+ return PULSAR_SERVICE_URL;
+ }
+
+ @Override
+ protected String adminUrl() {
+ return PULSAR_ADMIN_URL;
+ }
+}
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java
new file mode 100644
index 0000000..e442418
--- /dev/null
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java
@@ -0,0 +1,134 @@
+/*
+ * 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.pulsar.cases;
+
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.connector.pulsar.source.PulsarSource;
+import org.apache.flink.connector.pulsar.source.PulsarSourceBuilder;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.FixedRangeGenerator;
+import org.apache.flink.connector.pulsar.testutils.PulsarTestContext;
+import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
+import org.apache.flink.connectors.test.common.external.SourceSplitDataWriter;
+import org.apache.flink.tests.util.pulsar.common.KeyedPulsarPartitionDataWriter;
+
+import org.apache.pulsar.client.api.RegexSubscriptionMode;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.common.util.Murmur3_32Hash;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+import static java.util.Collections.singletonList;
+import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
+import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange.RANGE_SIZE;
+import static org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema.pulsarSchema;
+import static org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerRuntime.PULSAR_ADMIN_URL;
+import static org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerRuntime.PULSAR_SERVICE_URL;
+import static org.apache.pulsar.client.api.Schema.STRING;
+
+/**
+ * We would consuming from test splits by using {@link SubscriptionType#Key_Shared} subscription.
+ */
+public class KeySharedSubscriptionContext extends PulsarTestContext<String> {
+ private static final long serialVersionUID = 3246516520107893983L;
+
+ private int index = 0;
+
+ private final List<KeyedPulsarPartitionDataWriter> writers = new ArrayList<>();
+
+ // Message keys.
+ private final String key1;
+ private final String key2;
+
+ public KeySharedSubscriptionContext(PulsarTestEnvironment environment) {
+ super(environment);
+
+ // Init message keys.
+ this.key1 = randomAlphabetic(8);
+ String newKey2;
+ do {
+ newKey2 = randomAlphabetic(8);
+ } while (keyHash(key1) == keyHash(newKey2));
+ this.key2 = newKey2;
+ }
+
+ @Override
+ protected String displayName() {
+ return "consuming message by Key_Shared";
+ }
+
+ @Override
+ public Source<String, ?, ?> createSource(Boundedness boundedness) {
+ int keyHash = keyHash(key1);
+ TopicRange range = new TopicRange(keyHash, keyHash);
+
+ PulsarSourceBuilder<String> builder =
+ PulsarSource.builder()
+ .setDeserializationSchema(pulsarSchema(STRING))
+ .setServiceUrl(PULSAR_SERVICE_URL)
+ .setAdminUrl(PULSAR_ADMIN_URL)
+ .setTopicPattern(
+ "pulsar-[0-9]+-key-shared", RegexSubscriptionMode.AllTopics)
+ .setSubscriptionType(SubscriptionType.Key_Shared)
+ .setSubscriptionName("pulsar-key-shared")
+ .setRangeGenerator(new FixedRangeGenerator(singletonList(range)));
+ if (boundedness == Boundedness.BOUNDED) {
+ // Using latest stop cursor for making sure the source could be stopped.
+ builder.setBoundedStopCursor(StopCursor.latest());
+ }
+
+ return builder.build();
+ }
+
+ @Override
+ public SourceSplitDataWriter<String> createSourceSplitDataWriter() {
+ String topicName = "pulsar-" + index + "-key-shared";
+ operator.createTopic(topicName, 1);
+ index++;
+
+ String partitionName = TopicNameUtils.topicNameWithPartition(topicName, 0);
+ KeyedPulsarPartitionDataWriter writer =
+ new KeyedPulsarPartitionDataWriter(operator, partitionName, key1, key2);
+ writers.add(writer);
+
+ return writer;
+ }
+
+ @Override
+ public Collection<String> generateTestData(int splitIndex, long seed) {
+ return generateStringTestData(splitIndex, seed);
+ }
+
+ @Override
+ public void close() {
+ for (KeyedPulsarPartitionDataWriter writer : writers) {
+ writer.close();
+ }
+ writers.clear();
+ }
+
+ private int keyHash(String key) {
+ return Murmur3_32Hash.getInstance().makeHash(key.getBytes()) % RANGE_SIZE;
+ }
+}
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/SharedSubscriptionContext.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/SharedSubscriptionContext.java
new file mode 100644
index 0000000..f936b6f
--- /dev/null
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/SharedSubscriptionContext.java
@@ -0,0 +1,104 @@
+/*
+ * 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.pulsar.cases;
+
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.connector.pulsar.source.PulsarSource;
+import org.apache.flink.connector.pulsar.source.PulsarSourceBuilder;
+import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils;
+import org.apache.flink.connector.pulsar.testutils.PulsarPartitionDataWriter;
+import org.apache.flink.connector.pulsar.testutils.PulsarTestContext;
+import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
+import org.apache.flink.connectors.test.common.external.SourceSplitDataWriter;
+
+import org.apache.pulsar.client.api.RegexSubscriptionMode;
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+import static org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema.pulsarSchema;
+import static org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerRuntime.PULSAR_ADMIN_URL;
+import static org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerRuntime.PULSAR_SERVICE_URL;
+import static org.apache.pulsar.client.api.Schema.STRING;
+
+/** We would consuming from test splits by using {@link SubscriptionType#Shared} subscription. */
+public class SharedSubscriptionContext extends PulsarTestContext<String> {
+ private static final long serialVersionUID = -2798707923661295245L;
+
+ private int index = 0;
+
+ private final List<PulsarPartitionDataWriter> writers = new ArrayList<>();
+
+ public SharedSubscriptionContext(PulsarTestEnvironment environment) {
+ super(environment);
+ }
+
+ @Override
+ protected String displayName() {
+ return "consuming message by Shared";
+ }
+
+ @Override
+ public Source<String, ?, ?> createSource(Boundedness boundedness) {
+ PulsarSourceBuilder<String> builder =
+ PulsarSource.builder()
+ .setDeserializationSchema(pulsarSchema(STRING))
+ .setServiceUrl(PULSAR_SERVICE_URL)
+ .setAdminUrl(PULSAR_ADMIN_URL)
+ .setTopicPattern("pulsar-[0-9]+-shared", RegexSubscriptionMode.AllTopics)
+ .setSubscriptionType(SubscriptionType.Shared)
+ .setSubscriptionName("pulsar-shared");
+ if (boundedness == Boundedness.BOUNDED) {
+ // Using latest stop cursor for making sure the source could be stopped.
+ builder.setBoundedStopCursor(StopCursor.latest());
+ }
+
+ return builder.build();
+ }
+
+ @Override
+ public SourceSplitDataWriter<String> createSourceSplitDataWriter() {
+ String topicName = "pulsar-" + index + "-shared";
+ operator.createTopic(topicName, 1);
+ index++;
+
+ String partitionName = TopicNameUtils.topicNameWithPartition(topicName, 0);
+ PulsarPartitionDataWriter writer = new PulsarPartitionDataWriter(operator, partitionName);
+ writers.add(writer);
+
+ return writer;
+ }
+
+ @Override
+ public Collection<String> generateTestData(int splitIndex, long seed) {
+ return generateStringTestData(splitIndex, seed);
+ }
+
+ @Override
+ public void close() {
+ for (PulsarPartitionDataWriter writer : writers) {
+ writer.close();
+ }
+ writers.clear();
+ }
+}
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
new file mode 100644
index 0000000..890d09e
--- /dev/null
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
@@ -0,0 +1,54 @@
+/*
+ * 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.pulsar.common;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.MemorySize;
+import org.apache.flink.tests.util.TestUtils;
+import org.apache.flink.tests.util.flink.FlinkContainerTestEnvironment;
+
+import static org.apache.flink.configuration.TaskManagerOptions.TASK_OFF_HEAP_MEMORY;
+
+/** A Flink Container which would bundles pulsar connector in its classpath. */
+public class FlinkContainerWithPulsarEnvironment extends FlinkContainerTestEnvironment {
+
+ public FlinkContainerWithPulsarEnvironment(int numTaskManagers, int numSlotsPerTaskManager) {
+ super(
+ numTaskManagers,
+ numSlotsPerTaskManager,
+ resourcePath("pulsar-connector.jar"),
+ resourcePath("pulsar-client-all.jar"),
+ resourcePath("pulsar-client-api.jar"),
+ resourcePath("pulsar-admin-api.jar"),
+ resourcePath("jul-to-slf4j.jar"));
+ }
+
+ private static String resourcePath(String jarName) {
+ return TestUtils.getResource(jarName).toAbsolutePath().toString();
+ }
+
+ @Override
+ protected Configuration flinkConfiguration() {
+ Configuration configuration = super.flinkConfiguration();
+ // Increase the off heap memory for avoiding direct buffer memory error on Pulsar e2e tests.
+ configuration.set(TASK_OFF_HEAP_MEMORY, MemorySize.ofMebiBytes(100));
+
+ return configuration;
+ }
+}
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/KeyedPulsarPartitionDataWriter.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/KeyedPulsarPartitionDataWriter.java
new file mode 100644
index 0000000..eea97e6
--- /dev/null
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/KeyedPulsarPartitionDataWriter.java
@@ -0,0 +1,62 @@
+/*
+ * 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.pulsar.common;
+
+import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntimeOperator;
+import org.apache.flink.connectors.test.common.external.SourceSplitDataWriter;
+
+import org.apache.pulsar.client.api.Schema;
+
+import java.util.Collection;
+import java.util.List;
+
+import static java.util.stream.Collectors.toList;
+
+/**
+ * Source split data writer for writing test data into a Pulsar topic partition. It will write the
+ * message with two keys.
+ */
+public class KeyedPulsarPartitionDataWriter implements SourceSplitDataWriter<String> {
+
+ private final PulsarRuntimeOperator operator;
+ private final String fullTopicName;
+ private final String key1;
+ private final String key2;
+
+ public KeyedPulsarPartitionDataWriter(
+ PulsarRuntimeOperator operator, String fullTopicName, String key1, String key2) {
+ this.operator = operator;
+ this.fullTopicName = fullTopicName;
+ this.key1 = key1;
+ this.key2 = key2;
+ }
+
+ @Override
+ public void writeRecords(Collection<String> records) {
+ operator.sendMessages(fullTopicName, Schema.STRING, key1, records);
+
+ List<String> newRecords = records.stream().map(a -> a + key1).collect(toList());
+ operator.sendMessages(fullTopicName, Schema.STRING, key2, newRecords);
+ }
+
+ @Override
+ public void close() {
+ // Nothing to do.
+ }
+}
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/UnorderedSourceTestSuiteBase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/UnorderedSourceTestSuiteBase.java
new file mode 100644
index 0000000..c452fe6
--- /dev/null
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/UnorderedSourceTestSuiteBase.java
@@ -0,0 +1,72 @@
+/*
+ * 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.pulsar.common;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.Source;
+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.streaming.api.environment.StreamExecutionEnvironment;
+
+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 java.util.Collection;
+import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+
+/** A source test template for testing the messages which could be consumed in a unordered way. */
+@ExtendWith({
+ ConnectorTestingExtension.class,
+ TestLoggerExtension.class,
+ TestCaseInvocationContextProvider.class
+})
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public abstract class UnorderedSourceTestSuiteBase<T> {
+
+ @TestTemplate
+ @DisplayName("Test source with one split and four consumers")
+ public void testOneSplitWithMultipleConsumers(
+ TestEnvironment testEnv, ExternalContext<T> externalContext) throws Exception {
+ Collection<T> testData =
+ externalContext.generateTestData(0, ThreadLocalRandom.current().nextLong());
+ SourceSplitDataWriter<T> writer = externalContext.createSourceSplitDataWriter();
+ writer.writeRecords(testData);
+
+ Source<T, ?, ?> source = externalContext.createSource(Boundedness.BOUNDED);
+ StreamExecutionEnvironment execEnv = testEnv.createExecutionEnvironment();
+ List<T> results =
+ execEnv.fromSource(source, WatermarkStrategy.noWatermarks(), "Pulsar source")
+ .setParallelism(4)
+ .executeAndCollect(
+ "Source single split with four readers.", testData.size());
+
+ assertThat(results, containsInAnyOrder(testData.toArray()));
+ }
+}
diff --git a/flink-connector-pulsar-e2e-tests/src/test/resources/log4j2-test.properties b/flink-connector-pulsar-e2e-tests/src/test/resources/log4j2-test.properties
new file mode 100644
index 0000000..835c2ec
--- /dev/null
+++ b/flink-connector-pulsar-e2e-tests/src/test/resources/log4j2-test.properties
@@ -0,0 +1,28 @@
+################################################################################
+# 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.
+################################################################################
+
+# Set root logger level to OFF to not flood build logs
+# set manually to INFO for debugging purposes
+rootLogger.level = OFF
+rootLogger.appenderRef.test.ref = TestLogger
+
+appender.testlogger.name = TestLogger
+appender.testlogger.type = CONSOLE
+appender.testlogger.target = SYSTEM_ERR
+appender.testlogger.layout.type = PatternLayout
+appender.testlogger.layout.pattern = %-4r [%t] %-5p %c %x - %m%n
[flink-connector-pulsar] 13/31: [FLINK-26025][connector/pulsar] Replace MockPulsar with new Pulsar test tools based on PulsarStandalone.
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit 9c98b83104a15a58d16cb0c753e201f33044a903
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Wed Feb 9 15:09:50 2022 +0800
[FLINK-26025][connector/pulsar] Replace MockPulsar with new Pulsar test tools based on PulsarStandalone.
1. Drop some unused fields in test classes.
2. Fix the checkstyle issues for source test.
3. Fix violations for Pulsar connector according to the flink-architecture-tests.
4. Create a standalone Pulsar for test.
5. Add new methods to PulsarRuntimeOperator.
6. Fix the bug in PulsarContainerRuntime, support running tests in E2E environment.
7. Create PulsarContainerTestEnvironment for supporting E2E tests.
8. Add a lot of comments for Pulsar testing tools.
9. Drop mocked Pulsar service, use standalone Pulsar instead.
---
.../util/pulsar/PulsarSourceOrderedE2ECase.java | 7 ++---
.../util/pulsar/PulsarSourceUnorderedE2ECase.java | 7 ++---
.../pulsar/cases/ExclusiveSubscriptionContext.java | 14 ----------
.../pulsar/cases/FailoverSubscriptionContext.java | 14 ----------
.../pulsar/cases/KeySharedSubscriptionContext.java | 7 ++---
.../pulsar/cases/SharedSubscriptionContext.java | 7 ++---
.../common/PulsarContainerTestEnvironment.java | 31 ++++++++++++++++++++++
7 files changed, 39 insertions(+), 48 deletions(-)
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
index 7d22e80..502b41d 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
@@ -19,7 +19,6 @@
package org.apache.flink.tests.util.pulsar;
import org.apache.flink.connector.pulsar.testutils.PulsarTestContextFactory;
-import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
import org.apache.flink.connector.testframe.junit.annotations.TestContext;
import org.apache.flink.connector.testframe.junit.annotations.TestEnv;
import org.apache.flink.connector.testframe.junit.annotations.TestExternalSystem;
@@ -29,8 +28,7 @@ import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.tests.util.pulsar.cases.ExclusiveSubscriptionContext;
import org.apache.flink.tests.util.pulsar.cases.FailoverSubscriptionContext;
import org.apache.flink.tests.util.pulsar.common.FlinkContainerWithPulsarEnvironment;
-
-import static org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntime.container;
+import org.apache.flink.tests.util.pulsar.common.PulsarContainerTestEnvironment;
/**
* Pulsar E2E test based on connector testing framework. It's used for Failover & Exclusive
@@ -48,8 +46,7 @@ public class PulsarSourceOrderedE2ECase extends SourceTestSuiteBase<String> {
// Defines ConnectorExternalSystem.
@TestExternalSystem
- PulsarTestEnvironment pulsar =
- new PulsarTestEnvironment(container(flink.getFlinkContainers().getJobManager()));
+ PulsarContainerTestEnvironment pulsar = new PulsarContainerTestEnvironment(flink);
// Defines a set of external context Factories for different test cases.
@TestContext
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
index d14d8f9..5039048 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
@@ -19,7 +19,6 @@
package org.apache.flink.tests.util.pulsar;
import org.apache.flink.connector.pulsar.testutils.PulsarTestContextFactory;
-import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
import org.apache.flink.connector.testframe.junit.annotations.TestContext;
import org.apache.flink.connector.testframe.junit.annotations.TestEnv;
import org.apache.flink.connector.testframe.junit.annotations.TestExternalSystem;
@@ -28,10 +27,9 @@ import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.tests.util.pulsar.cases.KeySharedSubscriptionContext;
import org.apache.flink.tests.util.pulsar.cases.SharedSubscriptionContext;
import org.apache.flink.tests.util.pulsar.common.FlinkContainerWithPulsarEnvironment;
+import org.apache.flink.tests.util.pulsar.common.PulsarContainerTestEnvironment;
import org.apache.flink.tests.util.pulsar.common.UnorderedSourceTestSuiteBase;
-import static org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntime.container;
-
/**
* Pulsar E2E test based on connector testing framework. It's used for Shared & Key_Shared
* subscription.
@@ -48,8 +46,7 @@ public class PulsarSourceUnorderedE2ECase extends UnorderedSourceTestSuiteBase<S
// Defines ConnectorExternalSystem.
@TestExternalSystem
- PulsarTestEnvironment pulsar =
- new PulsarTestEnvironment(container(flink.getFlinkContainers().getJobManager()));
+ PulsarContainerTestEnvironment pulsar = new PulsarContainerTestEnvironment(flink);
// Defines a set of external context Factories for different test cases.
@SuppressWarnings("unused")
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/ExclusiveSubscriptionContext.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/ExclusiveSubscriptionContext.java
index 1245e14..6fea0c9 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/ExclusiveSubscriptionContext.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/ExclusiveSubscriptionContext.java
@@ -27,12 +27,8 @@ import java.net.URL;
import java.util.Collections;
import java.util.List;
-import static org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerRuntime.PULSAR_ADMIN_URL;
-import static org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerRuntime.PULSAR_SERVICE_URL;
-
/** We would consume from test splits by using {@link SubscriptionType#Exclusive} subscription. */
public class ExclusiveSubscriptionContext extends MultipleTopicTemplateContext {
- private static final long serialVersionUID = 1L;
public ExclusiveSubscriptionContext(PulsarTestEnvironment environment) {
this(environment, Collections.emptyList());
@@ -57,14 +53,4 @@ public class ExclusiveSubscriptionContext extends MultipleTopicTemplateContext {
protected SubscriptionType subscriptionType() {
return SubscriptionType.Exclusive;
}
-
- @Override
- protected String serviceUrl() {
- return PULSAR_SERVICE_URL;
- }
-
- @Override
- protected String adminUrl() {
- return PULSAR_ADMIN_URL;
- }
}
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/FailoverSubscriptionContext.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/FailoverSubscriptionContext.java
index 8ec1685..c473488 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/FailoverSubscriptionContext.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/FailoverSubscriptionContext.java
@@ -27,12 +27,8 @@ import java.net.URL;
import java.util.Collections;
import java.util.List;
-import static org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerRuntime.PULSAR_ADMIN_URL;
-import static org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerRuntime.PULSAR_SERVICE_URL;
-
/** We would consume from test splits by using {@link SubscriptionType#Failover} subscription. */
public class FailoverSubscriptionContext extends MultipleTopicTemplateContext {
- private static final long serialVersionUID = 1L;
public FailoverSubscriptionContext(PulsarTestEnvironment environment) {
this(environment, Collections.emptyList());
@@ -57,14 +53,4 @@ public class FailoverSubscriptionContext extends MultipleTopicTemplateContext {
protected SubscriptionType subscriptionType() {
return SubscriptionType.Failover;
}
-
- @Override
- protected String serviceUrl() {
- return PULSAR_SERVICE_URL;
- }
-
- @Override
- protected String adminUrl() {
- return PULSAR_ADMIN_URL;
- }
}
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java
index 303783a..5ad369b 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java
@@ -46,13 +46,10 @@ import static java.util.Collections.singletonList;
import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange.RANGE_SIZE;
import static org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema.pulsarSchema;
-import static org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerRuntime.PULSAR_ADMIN_URL;
-import static org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerRuntime.PULSAR_SERVICE_URL;
import static org.apache.pulsar.client.api.Schema.STRING;
/** We would consume from test splits by using {@link SubscriptionType#Key_Shared} subscription. */
public class KeySharedSubscriptionContext extends PulsarTestContext<String> {
- private static final long serialVersionUID = 1L;
private int index = 0;
@@ -92,8 +89,8 @@ public class KeySharedSubscriptionContext extends PulsarTestContext<String> {
PulsarSourceBuilder<String> builder =
PulsarSource.builder()
.setDeserializationSchema(pulsarSchema(STRING))
- .setServiceUrl(PULSAR_SERVICE_URL)
- .setAdminUrl(PULSAR_ADMIN_URL)
+ .setServiceUrl(operator.serviceUrl())
+ .setAdminUrl(operator.adminUrl())
.setTopicPattern(
"pulsar-[0-9]+-key-shared", RegexSubscriptionMode.AllTopics)
.setSubscriptionType(SubscriptionType.Key_Shared)
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/SharedSubscriptionContext.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/SharedSubscriptionContext.java
index de53595..1a2db66 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/SharedSubscriptionContext.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/SharedSubscriptionContext.java
@@ -40,13 +40,10 @@ import java.util.Collections;
import java.util.List;
import static org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema.pulsarSchema;
-import static org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerRuntime.PULSAR_ADMIN_URL;
-import static org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerRuntime.PULSAR_SERVICE_URL;
import static org.apache.pulsar.client.api.Schema.STRING;
/** We would consuming from test splits by using {@link SubscriptionType#Shared} subscription. */
public class SharedSubscriptionContext extends PulsarTestContext<String> {
- private static final long serialVersionUID = 1L;
private int index = 0;
@@ -71,8 +68,8 @@ public class SharedSubscriptionContext extends PulsarTestContext<String> {
PulsarSourceBuilder<String> builder =
PulsarSource.builder()
.setDeserializationSchema(pulsarSchema(STRING))
- .setServiceUrl(PULSAR_SERVICE_URL)
- .setAdminUrl(PULSAR_ADMIN_URL)
+ .setServiceUrl(operator.serviceUrl())
+ .setAdminUrl(operator.adminUrl())
.setTopicPattern("pulsar-[0-9]+-shared", RegexSubscriptionMode.AllTopics)
.setSubscriptionType(SubscriptionType.Shared)
.setSubscriptionName("pulsar-shared");
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/PulsarContainerTestEnvironment.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/PulsarContainerTestEnvironment.java
new file mode 100644
index 0000000..654347b
--- /dev/null
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/PulsarContainerTestEnvironment.java
@@ -0,0 +1,31 @@
+/*
+ * 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.pulsar.common;
+
+import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
+
+import static org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntime.container;
+
+/** This test environment is used for create a Pulsar standalone instance for e2e tests. */
+public class PulsarContainerTestEnvironment extends PulsarTestEnvironment {
+
+ public PulsarContainerTestEnvironment(FlinkContainerWithPulsarEnvironment flinkEnvironment) {
+ super(container(flinkEnvironment.getFlinkContainers().getJobManager()));
+ }
+}
[flink-connector-pulsar] 25/31: [FLINK-29495][Connector/Pulsar] Refactor Pulsar tests from JUnit4 annotation to JUnit5 annotation for disabling tests on Java 11
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit 1b1aa9aeed26345f792e1da551d11aaf36382654
Author: Martijn Visser <ma...@apache.org>
AuthorDate: Mon Oct 10 14:31:01 2022 +0200
[FLINK-29495][Connector/Pulsar] Refactor Pulsar tests from JUnit4 annotation to JUnit5 annotation for disabling tests on Java 11
---
.../java/org/apache/flink/tests/util/pulsar/PulsarSinkE2ECase.java | 5 ++---
.../apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java | 5 ++---
.../apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java | 5 ++---
3 files changed, 6 insertions(+), 9 deletions(-)
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSinkE2ECase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSinkE2ECase.java
index a19f593..46c8cfb 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSinkE2ECase.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSinkE2ECase.java
@@ -28,13 +28,12 @@ import org.apache.flink.connector.testframe.junit.annotations.TestSemantics;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.tests.util.pulsar.common.FlinkContainerWithPulsarEnvironment;
import org.apache.flink.tests.util.pulsar.common.PulsarContainerTestEnvironment;
-import org.apache.flink.testutils.junit.FailsOnJava11;
-import org.junit.experimental.categories.Category;
+import org.junit.jupiter.api.Tag;
/** Pulsar sink E2E test based on connector testing framework. */
@SuppressWarnings("unused")
-@Category(value = {FailsOnJava11.class})
+@Tag("org.apache.flink.testutils.junit.FailsOnJava11")
public class PulsarSinkE2ECase extends PulsarSinkTestSuiteBase {
@TestSemantics
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
index ea6a982..c223fb6 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
@@ -29,16 +29,15 @@ import org.apache.flink.tests.util.pulsar.common.FlinkContainerWithPulsarEnviron
import org.apache.flink.tests.util.pulsar.common.PulsarContainerTestEnvironment;
import org.apache.flink.tests.util.pulsar.source.ExclusiveSubscriptionContext;
import org.apache.flink.tests.util.pulsar.source.FailoverSubscriptionContext;
-import org.apache.flink.testutils.junit.FailsOnJava11;
-import org.junit.experimental.categories.Category;
+import org.junit.jupiter.api.Tag;
/**
* Pulsar E2E test based on connector testing framework. It's used for Failover & Exclusive
* subscription.
*/
@SuppressWarnings("unused")
-@Category(value = {FailsOnJava11.class})
+@Tag("org.apache.flink.testutils.junit.FailsOnJava11")
public class PulsarSourceOrderedE2ECase extends SourceTestSuiteBase<String> {
// Defines the Semantic.
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
index 89692d1..770d118 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
@@ -29,16 +29,15 @@ import org.apache.flink.connector.testframe.junit.annotations.TestSemantics;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.tests.util.pulsar.common.FlinkContainerWithPulsarEnvironment;
import org.apache.flink.tests.util.pulsar.common.PulsarContainerTestEnvironment;
-import org.apache.flink.testutils.junit.FailsOnJava11;
-import org.junit.experimental.categories.Category;
+import org.junit.jupiter.api.Tag;
/**
* Pulsar E2E test based on connector testing framework. It's used for Shared & Key_Shared
* subscription.
*/
@SuppressWarnings("unused")
-@Category(value = {FailsOnJava11.class})
+@Tag("org.apache.flink.testutils.junit.FailsOnJava11")
public class PulsarSourceUnorderedE2ECase extends UnorderedSourceTestSuiteBase<String> {
// Defines the Semantic.
[flink-connector-pulsar] 04/31: [FLINK-25038][testutils] Refactor FlinkContainer to split JM and TMs to individual containers and supports HA
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit 8de314deb821ef6ab763d32642c5896ef9a2db23
Author: Qingsheng Ren <re...@gmail.com>
AuthorDate: Thu Oct 14 15:48:21 2021 +0800
[FLINK-25038][testutils] Refactor FlinkContainer to split JM and TMs to individual containers and supports HA
---
.../org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java | 3 ++-
.../apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java | 3 ++-
2 files changed, 4 insertions(+), 2 deletions(-)
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
index 1427c2b..9a499e6 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
@@ -42,7 +42,8 @@ public class PulsarSourceOrderedE2ECase extends SourceTestSuiteBase<String> {
// Defines ConnectorExternalSystem.
@ExternalSystem
- PulsarTestEnvironment pulsar = new PulsarTestEnvironment(container(flink.getFlinkContainer()));
+ PulsarTestEnvironment pulsar =
+ new PulsarTestEnvironment(container(flink.getFlinkContainers().getJobManager()));
// Defines a set of external context Factories for different test cases.
@ExternalContextFactory
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
index 25cab21..797c7b1 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
@@ -42,7 +42,8 @@ public class PulsarSourceUnorderedE2ECase extends UnorderedSourceTestSuiteBase<S
// Defines ConnectorExternalSystem.
@ExternalSystem
- PulsarTestEnvironment pulsar = new PulsarTestEnvironment(container(flink.getFlinkContainer()));
+ PulsarTestEnvironment pulsar =
+ new PulsarTestEnvironment(container(flink.getFlinkContainers().getJobManager()));
// Defines a set of external context Factories for different test cases.
@ExternalContextFactory
[flink-connector-pulsar] 03/31: [hotfix][connector/testing-framework] Let ExternalContext#generateTestData returns List to preserve order
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit 52cbdb59d5103c319ff70f789a6837c17beb0dce
Author: Qingsheng Ren <re...@gmail.com>
AuthorDate: Mon Aug 23 17:18:13 2021 +0800
[hotfix][connector/testing-framework] Let ExternalContext#generateTestData returns List to preserve order
---
.../flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java | 3 +--
.../flink/tests/util/pulsar/cases/SharedSubscriptionContext.java | 3 +--
2 files changed, 2 insertions(+), 4 deletions(-)
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java
index e442418..5d937ba 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java
@@ -36,7 +36,6 @@ import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.common.util.Murmur3_32Hash;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.List;
import static java.util.Collections.singletonList;
@@ -116,7 +115,7 @@ public class KeySharedSubscriptionContext extends PulsarTestContext<String> {
}
@Override
- public Collection<String> generateTestData(int splitIndex, long seed) {
+ public List<String> generateTestData(int splitIndex, long seed) {
return generateStringTestData(splitIndex, seed);
}
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/SharedSubscriptionContext.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/SharedSubscriptionContext.java
index f936b6f..52e30b3 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/SharedSubscriptionContext.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/SharedSubscriptionContext.java
@@ -33,7 +33,6 @@ import org.apache.pulsar.client.api.RegexSubscriptionMode;
import org.apache.pulsar.client.api.SubscriptionType;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.List;
import static org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema.pulsarSchema;
@@ -90,7 +89,7 @@ public class SharedSubscriptionContext extends PulsarTestContext<String> {
}
@Override
- public Collection<String> generateTestData(int splitIndex, long seed) {
+ public List<String> generateTestData(int splitIndex, long seed) {
return generateStringTestData(splitIndex, seed);
}
[flink-connector-pulsar] 06/31: [FLINK-25210][pulsar][e2e][tests] add resource file to test jar
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit 9046e81e5a93600d86a135826fd57ebd6f6172fb
Author: Yufei Zhang <af...@gmail.com>
AuthorDate: Wed Dec 15 20:48:01 2021 +0800
[FLINK-25210][pulsar][e2e][tests] add resource file to test jar
---
.../org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java | 3 ---
.../apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java | 3 ---
2 files changed, 6 deletions(-)
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
index 95c1486..9a499e6 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
@@ -28,15 +28,12 @@ import org.apache.flink.tests.util.pulsar.cases.ExclusiveSubscriptionContext;
import org.apache.flink.tests.util.pulsar.cases.FailoverSubscriptionContext;
import org.apache.flink.tests.util.pulsar.common.FlinkContainerWithPulsarEnvironment;
-import org.junit.jupiter.api.Disabled;
-
import static org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntime.container;
/**
* Pulsar E2E test based on connector testing framework. It's used for Failover & Exclusive
* subscription.
*/
-@Disabled("FLINK-25210")
public class PulsarSourceOrderedE2ECase extends SourceTestSuiteBase<String> {
// Defines TestEnvironment.
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
index c59831f..797c7b1 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
@@ -28,15 +28,12 @@ import org.apache.flink.tests.util.pulsar.cases.SharedSubscriptionContext;
import org.apache.flink.tests.util.pulsar.common.FlinkContainerWithPulsarEnvironment;
import org.apache.flink.tests.util.pulsar.common.UnorderedSourceTestSuiteBase;
-import org.junit.jupiter.api.Disabled;
-
import static org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntime.container;
/**
* Pulsar E2E test based on connector testing framework. It's used for Shared & Key_Shared
* subscription.
*/
-@Disabled("FLINK-25210")
public class PulsarSourceUnorderedE2ECase extends UnorderedSourceTestSuiteBase<String> {
// Defines TestEnvironment.
[flink-connector-pulsar] 07/31: [FLINK-25287][connectors/testing-framework] Refactor connector testing framework interfaces for more scenarios
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit 082bd2fda175a635bde2168f2d18b37be0032b4a
Author: Qingsheng Ren <re...@gmail.com>
AuthorDate: Mon Jan 17 16:57:42 2022 +0800
[FLINK-25287][connectors/testing-framework] Refactor connector testing framework interfaces for more scenarios
---
.../util/pulsar/common/FlinkContainerWithPulsarEnvironment.java | 7 +++----
1 file changed, 3 insertions(+), 4 deletions(-)
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
index 890d09e..14feeed 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
@@ -30,6 +30,7 @@ public class FlinkContainerWithPulsarEnvironment extends FlinkContainerTestEnvir
public FlinkContainerWithPulsarEnvironment(int numTaskManagers, int numSlotsPerTaskManager) {
super(
+ flinkConfiguration(),
numTaskManagers,
numSlotsPerTaskManager,
resourcePath("pulsar-connector.jar"),
@@ -43,12 +44,10 @@ public class FlinkContainerWithPulsarEnvironment extends FlinkContainerTestEnvir
return TestUtils.getResource(jarName).toAbsolutePath().toString();
}
- @Override
- protected Configuration flinkConfiguration() {
- Configuration configuration = super.flinkConfiguration();
+ protected static Configuration flinkConfiguration() {
+ Configuration configuration = new Configuration();
// Increase the off heap memory for avoiding direct buffer memory error on Pulsar e2e tests.
configuration.set(TASK_OFF_HEAP_MEMORY, MemorySize.ofMebiBytes(100));
-
return configuration;
}
}
[flink-connector-pulsar] 21/31: [FLINK-29285][tests] Move TestUtils#getResource
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit f333cf2d5a0bf0a5dfe0421df4c391dd84a22afa
Author: Chesnay Schepler <ch...@apache.org>
AuthorDate: Wed Sep 14 17:12:10 2022 +0200
[FLINK-29285][tests] Move TestUtils#getResource
---
.../tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
index a4141b0..9f34554 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
@@ -23,7 +23,7 @@ import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.connector.testframe.container.FlinkContainerTestEnvironment;
-import org.apache.flink.tests.util.TestUtils;
+import org.apache.flink.test.resources.ResourceTestUtils;
/** A Flink Container which would bundles pulsar connector in its classpath. */
public class FlinkContainerWithPulsarEnvironment extends FlinkContainerTestEnvironment {
@@ -47,7 +47,7 @@ public class FlinkContainerWithPulsarEnvironment extends FlinkContainerTestEnvir
}
private static String resourcePath(String jarName) {
- return TestUtils.getResource(jarName).toAbsolutePath().toString();
+ return ResourceTestUtils.getResource(jarName).toAbsolutePath().toString();
}
protected static Configuration flinkConfiguration() {
[flink-connector-pulsar] 29/31: [hotfix] Increase timeout for compile and test runs
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit 7360e65fafc199def54d44051c3db74aee90f63e
Author: Martijn Visser <ma...@apache.org>
AuthorDate: Fri Dec 9 11:38:45 2022 +0100
[hotfix] Increase timeout for compile and test runs
---
.github/workflows/ci.yml | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 09d8182..ea8a6c1 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -38,7 +38,7 @@ jobs:
strategy:
matrix:
jdk: [8, 11]
- timeout-minutes: 40
+ timeout-minutes: 80
env:
MVN_COMMON_OPTIONS: -U -B --no-transfer-progress -Dflink.version=${{ inputs.flink_version }}
MVN_CONNECTION_OPTIONS: -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.httpconnectionManager.ttlSeconds=120
@@ -80,7 +80,7 @@ jobs:
run: wget -q -c ${{ inputs.flink_url }} -O - | tar -xz
- name: Compile and test flink-connector-pulsar
- timeout-minutes: 30
+ timeout-minutes: 60
run: |
set -o pipefail
[flink-connector-pulsar] 27/31: [FLINK-26027][Connector/Pulsar] Expose Pulsar producer metrics and add FLIP-33 sink metrics. (#21249)
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit bf7af298a0a3dcdff1ec6153a388cd82e29f575f
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Fri Nov 11 21:10:11 2022 +0800
[FLINK-26027][Connector/Pulsar] Expose Pulsar producer metrics and add FLIP-33 sink metrics. (#21249)
---
.../java/org/apache/flink/tests/util/pulsar/PulsarSinkE2ECase.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSinkE2ECase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSinkE2ECase.java
index 46c8cfb..b945258 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSinkE2ECase.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSinkE2ECase.java
@@ -20,11 +20,11 @@ package org.apache.flink.tests.util.pulsar;
import org.apache.flink.connector.pulsar.testutils.PulsarTestContextFactory;
import org.apache.flink.connector.pulsar.testutils.sink.PulsarSinkTestContext;
-import org.apache.flink.connector.pulsar.testutils.sink.PulsarSinkTestSuiteBase;
import org.apache.flink.connector.testframe.junit.annotations.TestContext;
import org.apache.flink.connector.testframe.junit.annotations.TestEnv;
import org.apache.flink.connector.testframe.junit.annotations.TestExternalSystem;
import org.apache.flink.connector.testframe.junit.annotations.TestSemantics;
+import org.apache.flink.connector.testframe.testsuites.SinkTestSuiteBase;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.tests.util.pulsar.common.FlinkContainerWithPulsarEnvironment;
import org.apache.flink.tests.util.pulsar.common.PulsarContainerTestEnvironment;
@@ -34,7 +34,7 @@ import org.junit.jupiter.api.Tag;
/** Pulsar sink E2E test based on connector testing framework. */
@SuppressWarnings("unused")
@Tag("org.apache.flink.testutils.junit.FailsOnJava11")
-public class PulsarSinkE2ECase extends PulsarSinkTestSuiteBase {
+public class PulsarSinkE2ECase extends SinkTestSuiteBase<String> {
@TestSemantics
CheckpointingMode[] semantics =
[flink-connector-pulsar] 19/31: [FLINK-28409][tests] Move FlinkContainerTestEnvironment and dependendencies into flink-connector-test-utils
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit 9714ebb997b84d4f8e6f20a5078ad573f33bc2bf
Author: Alexander Fedulov <14...@users.noreply.github.com>
AuthorDate: Wed Jul 6 20:38:26 2022 +0200
[FLINK-28409][tests] Move FlinkContainerTestEnvironment and dependendencies into flink-connector-test-utils
---
.../tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
index d2f9ab4..a4141b0 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
@@ -22,8 +22,8 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.TaskManagerOptions;
+import org.apache.flink.connector.testframe.container.FlinkContainerTestEnvironment;
import org.apache.flink.tests.util.TestUtils;
-import org.apache.flink.tests.util.flink.container.FlinkContainerTestEnvironment;
/** A Flink Container which would bundles pulsar connector in its classpath. */
public class FlinkContainerWithPulsarEnvironment extends FlinkContainerTestEnvironment {
[flink-connector-pulsar] 16/31: Update version to 1.16-SNAPSHOT
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit bca31dbba3038842c2a355eb5a17b2fb4e4ffb81
Author: Yun Gao <ga...@gmail.com>
AuthorDate: Tue Mar 15 01:27:55 2022 +0800
Update version to 1.16-SNAPSHOT
---
flink-connector-pulsar-e2e-tests/pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/flink-connector-pulsar-e2e-tests/pom.xml b/flink-connector-pulsar-e2e-tests/pom.xml
index 424b106..13bf409 100644
--- a/flink-connector-pulsar-e2e-tests/pom.xml
+++ b/flink-connector-pulsar-e2e-tests/pom.xml
@@ -23,7 +23,7 @@ under the License.
<parent>
<artifactId>flink-end-to-end-tests</artifactId>
<groupId>org.apache.flink</groupId>
- <version>1.15-SNAPSHOT</version>
+ <version>1.16-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
[flink-connector-pulsar] 08/31: [FLINK-25287][connectors/pulsar] Use connector testing framework interface for Pulsar tests
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit e94e54b5201b08ba353133571ac2bd76392708c4
Author: Qingsheng Ren <re...@gmail.com>
AuthorDate: Tue Jan 18 20:44:00 2022 +0800
[FLINK-25287][connectors/pulsar] Use connector testing framework interface for Pulsar tests
---
.../util/pulsar/PulsarSourceOrderedE2ECase.java | 14 ++++----
.../util/pulsar/PulsarSourceUnorderedE2ECase.java | 12 +++----
.../pulsar/cases/ExclusiveSubscriptionContext.java | 11 +++++-
.../pulsar/cases/FailoverSubscriptionContext.java | 11 +++++-
.../pulsar/cases/KeySharedSubscriptionContext.java | 28 +++++++++++----
.../pulsar/cases/SharedSubscriptionContext.java | 28 +++++++++++----
.../common/KeyedPulsarPartitionDataWriter.java | 7 ++--
.../common/UnorderedSourceTestSuiteBase.java | 40 +++++++++++++++-------
8 files changed, 107 insertions(+), 44 deletions(-)
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
index 9a499e6..8641f50 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
@@ -20,10 +20,10 @@ package org.apache.flink.tests.util.pulsar;
import org.apache.flink.connector.pulsar.testutils.PulsarTestContextFactory;
import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
-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.connector.testframe.junit.annotations.TestContext;
+import org.apache.flink.connector.testframe.junit.annotations.TestEnv;
+import org.apache.flink.connector.testframe.junit.annotations.TestExternalSystem;
+import org.apache.flink.connector.testframe.testsuites.SourceTestSuiteBase;
import org.apache.flink.tests.util.pulsar.cases.ExclusiveSubscriptionContext;
import org.apache.flink.tests.util.pulsar.cases.FailoverSubscriptionContext;
import org.apache.flink.tests.util.pulsar.common.FlinkContainerWithPulsarEnvironment;
@@ -41,16 +41,16 @@ public class PulsarSourceOrderedE2ECase extends SourceTestSuiteBase<String> {
FlinkContainerWithPulsarEnvironment flink = new FlinkContainerWithPulsarEnvironment(1, 6);
// Defines ConnectorExternalSystem.
- @ExternalSystem
+ @TestExternalSystem
PulsarTestEnvironment pulsar =
new PulsarTestEnvironment(container(flink.getFlinkContainers().getJobManager()));
// Defines a set of external context Factories for different test cases.
- @ExternalContextFactory
+ @TestContext
PulsarTestContextFactory<String, ExclusiveSubscriptionContext> exclusive =
new PulsarTestContextFactory<>(pulsar, ExclusiveSubscriptionContext::new);
- @ExternalContextFactory
+ @TestContext
PulsarTestContextFactory<String, FailoverSubscriptionContext> failover =
new PulsarTestContextFactory<>(pulsar, FailoverSubscriptionContext::new);
}
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
index 797c7b1..e519618 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
@@ -20,9 +20,9 @@ package org.apache.flink.tests.util.pulsar;
import org.apache.flink.connector.pulsar.testutils.PulsarTestContextFactory;
import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
-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.connector.testframe.junit.annotations.TestContext;
+import org.apache.flink.connector.testframe.junit.annotations.TestEnv;
+import org.apache.flink.connector.testframe.junit.annotations.TestExternalSystem;
import org.apache.flink.tests.util.pulsar.cases.KeySharedSubscriptionContext;
import org.apache.flink.tests.util.pulsar.cases.SharedSubscriptionContext;
import org.apache.flink.tests.util.pulsar.common.FlinkContainerWithPulsarEnvironment;
@@ -41,16 +41,16 @@ public class PulsarSourceUnorderedE2ECase extends UnorderedSourceTestSuiteBase<S
FlinkContainerWithPulsarEnvironment flink = new FlinkContainerWithPulsarEnvironment(1, 8);
// Defines ConnectorExternalSystem.
- @ExternalSystem
+ @TestExternalSystem
PulsarTestEnvironment pulsar =
new PulsarTestEnvironment(container(flink.getFlinkContainers().getJobManager()));
// Defines a set of external context Factories for different test cases.
- @ExternalContextFactory
+ @TestContext
PulsarTestContextFactory<String, SharedSubscriptionContext> shared =
new PulsarTestContextFactory<>(pulsar, SharedSubscriptionContext::new);
- @ExternalContextFactory
+ @TestContext
PulsarTestContextFactory<String, KeySharedSubscriptionContext> keyShared =
new PulsarTestContextFactory<>(pulsar, KeySharedSubscriptionContext::new);
}
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/ExclusiveSubscriptionContext.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/ExclusiveSubscriptionContext.java
index 18b2ffc..c3c4959 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/ExclusiveSubscriptionContext.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/ExclusiveSubscriptionContext.java
@@ -23,6 +23,10 @@ import org.apache.flink.connector.pulsar.testutils.cases.MultipleTopicTemplateCo
import org.apache.pulsar.client.api.SubscriptionType;
+import java.net.URL;
+import java.util.Collections;
+import java.util.List;
+
import static org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerRuntime.PULSAR_ADMIN_URL;
import static org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerRuntime.PULSAR_SERVICE_URL;
@@ -31,7 +35,12 @@ public class ExclusiveSubscriptionContext extends MultipleTopicTemplateContext {
private static final long serialVersionUID = 6238209089442257487L;
public ExclusiveSubscriptionContext(PulsarTestEnvironment environment) {
- super(environment);
+ this(environment, Collections.emptyList());
+ }
+
+ public ExclusiveSubscriptionContext(
+ PulsarTestEnvironment environment, List<URL> connectorJarPaths) {
+ super(environment, connectorJarPaths);
}
@Override
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/FailoverSubscriptionContext.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/FailoverSubscriptionContext.java
index c322efa..9dbbec8 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/FailoverSubscriptionContext.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/FailoverSubscriptionContext.java
@@ -23,6 +23,10 @@ import org.apache.flink.connector.pulsar.testutils.cases.MultipleTopicTemplateCo
import org.apache.pulsar.client.api.SubscriptionType;
+import java.net.URL;
+import java.util.Collections;
+import java.util.List;
+
import static org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerRuntime.PULSAR_ADMIN_URL;
import static org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerRuntime.PULSAR_SERVICE_URL;
@@ -31,7 +35,12 @@ public class FailoverSubscriptionContext extends MultipleTopicTemplateContext {
private static final long serialVersionUID = 6238209089442257487L;
public FailoverSubscriptionContext(PulsarTestEnvironment environment) {
- super(environment);
+ this(environment, Collections.emptyList());
+ }
+
+ public FailoverSubscriptionContext(
+ PulsarTestEnvironment environment, List<URL> connectorJarPaths) {
+ super(environment, connectorJarPaths);
}
@Override
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java
index 5d937ba..0be3ac3 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java
@@ -18,6 +18,7 @@
package org.apache.flink.tests.util.pulsar.cases;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.api.connector.source.Source;
import org.apache.flink.connector.pulsar.source.PulsarSource;
@@ -28,14 +29,17 @@ import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange;
import org.apache.flink.connector.pulsar.source.enumerator.topic.range.FixedRangeGenerator;
import org.apache.flink.connector.pulsar.testutils.PulsarTestContext;
import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
-import org.apache.flink.connectors.test.common.external.SourceSplitDataWriter;
+import org.apache.flink.connector.testframe.external.ExternalSystemSplitDataWriter;
+import org.apache.flink.connector.testframe.external.source.TestingSourceSettings;
import org.apache.flink.tests.util.pulsar.common.KeyedPulsarPartitionDataWriter;
import org.apache.pulsar.client.api.RegexSubscriptionMode;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.common.util.Murmur3_32Hash;
+import java.net.URL;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import static java.util.Collections.singletonList;
@@ -61,7 +65,12 @@ public class KeySharedSubscriptionContext extends PulsarTestContext<String> {
private final String key2;
public KeySharedSubscriptionContext(PulsarTestEnvironment environment) {
- super(environment);
+ this(environment, Collections.emptyList());
+ }
+
+ public KeySharedSubscriptionContext(
+ PulsarTestEnvironment environment, List<URL> connectorJarPaths) {
+ super(environment, connectorJarPaths);
// Init message keys.
this.key1 = randomAlphabetic(8);
@@ -78,7 +87,7 @@ public class KeySharedSubscriptionContext extends PulsarTestContext<String> {
}
@Override
- public Source<String, ?, ?> createSource(Boundedness boundedness) {
+ public Source<String, ?, ?> createSource(TestingSourceSettings sourceSettings) {
int keyHash = keyHash(key1);
TopicRange range = new TopicRange(keyHash, keyHash);
@@ -92,7 +101,7 @@ public class KeySharedSubscriptionContext extends PulsarTestContext<String> {
.setSubscriptionType(SubscriptionType.Key_Shared)
.setSubscriptionName("pulsar-key-shared")
.setRangeGenerator(new FixedRangeGenerator(singletonList(range)));
- if (boundedness == Boundedness.BOUNDED) {
+ if (sourceSettings.getBoundedness() == Boundedness.BOUNDED) {
// Using latest stop cursor for making sure the source could be stopped.
builder.setBoundedStopCursor(StopCursor.latest());
}
@@ -101,7 +110,8 @@ public class KeySharedSubscriptionContext extends PulsarTestContext<String> {
}
@Override
- public SourceSplitDataWriter<String> createSourceSplitDataWriter() {
+ public ExternalSystemSplitDataWriter<String> createSourceSplitDataWriter(
+ TestingSourceSettings sourceSettings) {
String topicName = "pulsar-" + index + "-key-shared";
operator.createTopic(topicName, 1);
index++;
@@ -115,10 +125,16 @@ public class KeySharedSubscriptionContext extends PulsarTestContext<String> {
}
@Override
- public List<String> generateTestData(int splitIndex, long seed) {
+ public List<String> generateTestData(
+ TestingSourceSettings sourceSettings, int splitIndex, long seed) {
return generateStringTestData(splitIndex, seed);
}
+ @Override
+ public TypeInformation<String> getProducedType() {
+ return TypeInformation.of(String.class);
+ }
+
@Override
public void close() {
for (KeyedPulsarPartitionDataWriter writer : writers) {
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/SharedSubscriptionContext.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/SharedSubscriptionContext.java
index 52e30b3..5a4ce75 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/SharedSubscriptionContext.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/SharedSubscriptionContext.java
@@ -18,6 +18,7 @@
package org.apache.flink.tests.util.pulsar.cases;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.api.connector.source.Source;
import org.apache.flink.connector.pulsar.source.PulsarSource;
@@ -27,12 +28,15 @@ import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils;
import org.apache.flink.connector.pulsar.testutils.PulsarPartitionDataWriter;
import org.apache.flink.connector.pulsar.testutils.PulsarTestContext;
import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
-import org.apache.flink.connectors.test.common.external.SourceSplitDataWriter;
+import org.apache.flink.connector.testframe.external.ExternalSystemSplitDataWriter;
+import org.apache.flink.connector.testframe.external.source.TestingSourceSettings;
import org.apache.pulsar.client.api.RegexSubscriptionMode;
import org.apache.pulsar.client.api.SubscriptionType;
+import java.net.URL;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import static org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema.pulsarSchema;
@@ -49,7 +53,12 @@ public class SharedSubscriptionContext extends PulsarTestContext<String> {
private final List<PulsarPartitionDataWriter> writers = new ArrayList<>();
public SharedSubscriptionContext(PulsarTestEnvironment environment) {
- super(environment);
+ this(environment, Collections.emptyList());
+ }
+
+ public SharedSubscriptionContext(
+ PulsarTestEnvironment environment, List<URL> connectorJarPaths) {
+ super(environment, connectorJarPaths);
}
@Override
@@ -58,7 +67,7 @@ public class SharedSubscriptionContext extends PulsarTestContext<String> {
}
@Override
- public Source<String, ?, ?> createSource(Boundedness boundedness) {
+ public Source<String, ?, ?> createSource(TestingSourceSettings sourceSettings) {
PulsarSourceBuilder<String> builder =
PulsarSource.builder()
.setDeserializationSchema(pulsarSchema(STRING))
@@ -67,7 +76,7 @@ public class SharedSubscriptionContext extends PulsarTestContext<String> {
.setTopicPattern("pulsar-[0-9]+-shared", RegexSubscriptionMode.AllTopics)
.setSubscriptionType(SubscriptionType.Shared)
.setSubscriptionName("pulsar-shared");
- if (boundedness == Boundedness.BOUNDED) {
+ if (sourceSettings.getBoundedness() == Boundedness.BOUNDED) {
// Using latest stop cursor for making sure the source could be stopped.
builder.setBoundedStopCursor(StopCursor.latest());
}
@@ -76,7 +85,8 @@ public class SharedSubscriptionContext extends PulsarTestContext<String> {
}
@Override
- public SourceSplitDataWriter<String> createSourceSplitDataWriter() {
+ public ExternalSystemSplitDataWriter<String> createSourceSplitDataWriter(
+ TestingSourceSettings sourceSettings) {
String topicName = "pulsar-" + index + "-shared";
operator.createTopic(topicName, 1);
index++;
@@ -89,10 +99,16 @@ public class SharedSubscriptionContext extends PulsarTestContext<String> {
}
@Override
- public List<String> generateTestData(int splitIndex, long seed) {
+ public List<String> generateTestData(
+ TestingSourceSettings sourceSettings, int splitIndex, long seed) {
return generateStringTestData(splitIndex, seed);
}
+ @Override
+ public TypeInformation<String> getProducedType() {
+ return TypeInformation.of(String.class);
+ }
+
@Override
public void close() {
for (PulsarPartitionDataWriter writer : writers) {
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/KeyedPulsarPartitionDataWriter.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/KeyedPulsarPartitionDataWriter.java
index eea97e6..e431e4c 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/KeyedPulsarPartitionDataWriter.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/KeyedPulsarPartitionDataWriter.java
@@ -19,11 +19,10 @@
package org.apache.flink.tests.util.pulsar.common;
import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntimeOperator;
-import org.apache.flink.connectors.test.common.external.SourceSplitDataWriter;
+import org.apache.flink.connector.testframe.external.ExternalSystemSplitDataWriter;
import org.apache.pulsar.client.api.Schema;
-import java.util.Collection;
import java.util.List;
import static java.util.stream.Collectors.toList;
@@ -32,7 +31,7 @@ import static java.util.stream.Collectors.toList;
* Source split data writer for writing test data into a Pulsar topic partition. It will write the
* message with two keys.
*/
-public class KeyedPulsarPartitionDataWriter implements SourceSplitDataWriter<String> {
+public class KeyedPulsarPartitionDataWriter implements ExternalSystemSplitDataWriter<String> {
private final PulsarRuntimeOperator operator;
private final String fullTopicName;
@@ -48,7 +47,7 @@ public class KeyedPulsarPartitionDataWriter implements SourceSplitDataWriter<Str
}
@Override
- public void writeRecords(Collection<String> records) {
+ public void writeRecords(List<String> records) {
operator.sendMessages(fullTopicName, Schema.STRING, key1, records);
List<String> newRecords = records.stream().map(a -> a + key1).collect(toList());
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/UnorderedSourceTestSuiteBase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/UnorderedSourceTestSuiteBase.java
index c452fe6..01527ea 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/UnorderedSourceTestSuiteBase.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/UnorderedSourceTestSuiteBase.java
@@ -21,20 +21,22 @@ package org.apache.flink.tests.util.pulsar.common;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.api.connector.source.Source;
-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.connector.testframe.environment.TestEnvironment;
+import org.apache.flink.connector.testframe.environment.TestEnvironmentSettings;
+import org.apache.flink.connector.testframe.external.ExternalSystemSplitDataWriter;
+import org.apache.flink.connector.testframe.external.source.DataStreamSourceExternalContext;
+import org.apache.flink.connector.testframe.external.source.TestingSourceSettings;
+import org.apache.flink.connector.testframe.junit.extensions.ConnectorTestingExtension;
+import org.apache.flink.connector.testframe.junit.extensions.TestCaseInvocationContextProvider;
+import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.util.TestLoggerExtension;
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 java.util.Collection;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
@@ -53,14 +55,26 @@ public abstract class UnorderedSourceTestSuiteBase<T> {
@TestTemplate
@DisplayName("Test source with one split and four consumers")
public void testOneSplitWithMultipleConsumers(
- TestEnvironment testEnv, ExternalContext<T> externalContext) throws Exception {
- Collection<T> testData =
- externalContext.generateTestData(0, ThreadLocalRandom.current().nextLong());
- SourceSplitDataWriter<T> writer = externalContext.createSourceSplitDataWriter();
+ TestEnvironment testEnv, DataStreamSourceExternalContext<T> externalContext)
+ throws Exception {
+ TestingSourceSettings sourceSettings =
+ TestingSourceSettings.builder()
+ .setBoundedness(Boundedness.BOUNDED)
+ .setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
+ .build();
+ TestEnvironmentSettings envOptions =
+ TestEnvironmentSettings.builder()
+ .setConnectorJarPaths(externalContext.getConnectorJarPaths())
+ .build();
+ List<T> testData =
+ externalContext.generateTestData(
+ sourceSettings, 0, ThreadLocalRandom.current().nextLong());
+ ExternalSystemSplitDataWriter<T> writer =
+ externalContext.createSourceSplitDataWriter(sourceSettings);
writer.writeRecords(testData);
- Source<T, ?, ?> source = externalContext.createSource(Boundedness.BOUNDED);
- StreamExecutionEnvironment execEnv = testEnv.createExecutionEnvironment();
+ Source<T, ?, ?> source = externalContext.createSource(sourceSettings);
+ StreamExecutionEnvironment execEnv = testEnv.createExecutionEnvironment(envOptions);
List<T> results =
execEnv.fromSource(source, WatermarkStrategy.noWatermarks(), "Pulsar source")
.setParallelism(4)
[flink-connector-pulsar] 12/31: [FLINK-24246][connector/pulsar] Bump PulsarClient version to latest 2.9.1
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit 704ef1f5a93c19107fb8a8cb51f843ce07256fb5
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Wed Feb 9 13:00:53 2022 +0800
[FLINK-24246][connector/pulsar] Bump PulsarClient version to latest 2.9.1
1. Bump the pulsar-client-all version in pom file.
2. Exclude useless dependencies for pulsar-client-all.
3. Bump the Pulsar docker version.
4. Change the dependencies to pass the tests.
5. Drop PulsarTransactionUtils and fix compile issues in tests.
6. Add bouncycastle to Pulsar e2e tests.
---
flink-connector-pulsar-e2e-tests/pom.xml | 43 +++++++++++++++++++++-
.../FlinkContainerWithPulsarEnvironment.java | 5 +++
2 files changed, 47 insertions(+), 1 deletion(-)
diff --git a/flink-connector-pulsar-e2e-tests/pom.xml b/flink-connector-pulsar-e2e-tests/pom.xml
index e7caf8b..7c87ec7 100644
--- a/flink-connector-pulsar-e2e-tests/pom.xml
+++ b/flink-connector-pulsar-e2e-tests/pom.xml
@@ -31,7 +31,8 @@ under the License.
<name>Flink : E2E Tests : Pulsar</name>
<properties>
- <pulsar.version>2.8.0</pulsar.version>
+ <pulsar.version>2.9.1</pulsar.version>
+ <bouncycastle.version>1.69</bouncycastle.version>
</properties>
<dependencies>
@@ -105,6 +106,46 @@ under the License.
<type>jar</type>
<outputDirectory>${project.build.directory}/dependencies</outputDirectory>
</artifactItem>
+ <dependency>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>bouncy-castle-bc</artifactId>
+ <version>${pulsar.version}</version>
+ <destFileName>bouncy-castle-bc.jar</destFileName>
+ <type>jar</type>
+ <outputDirectory>${project.build.directory}/dependencies</outputDirectory>
+ </dependency>
+ <dependency>
+ <groupId>org.bouncycastle</groupId>
+ <artifactId>bcpkix-jdk15on</artifactId>
+ <version>${bouncycastle.version}</version>
+ <destFileName>bcpkix-jdk15on.jar</destFileName>
+ <type>jar</type>
+ <outputDirectory>${project.build.directory}/dependencies</outputDirectory>
+ </dependency>
+ <dependency>
+ <groupId>org.bouncycastle</groupId>
+ <artifactId>bcprov-jdk15on</artifactId>
+ <version>${bouncycastle.version}</version>
+ <destFileName>bcprov-jdk15on.jar</destFileName>
+ <type>jar</type>
+ <outputDirectory>${project.build.directory}/dependencies</outputDirectory>
+ </dependency>
+ <dependency>
+ <groupId>org.bouncycastle</groupId>
+ <artifactId>bcutil-jdk15on</artifactId>
+ <version>${bouncycastle.version}</version>
+ <destFileName>bcutil-jdk15on.jar</destFileName>
+ <type>jar</type>
+ <outputDirectory>${project.build.directory}/dependencies</outputDirectory>
+ </dependency>
+ <dependency>
+ <groupId>org.bouncycastle</groupId>
+ <artifactId>bcprov-ext-jdk15on</artifactId>
+ <version>${bouncycastle.version}</version>
+ <destFileName>bcprov-ext-jdk15on.jar</destFileName>
+ <type>jar</type>
+ <outputDirectory>${project.build.directory}/dependencies</outputDirectory>
+ </dependency>
<artifactItem>
<groupId>org.slf4j</groupId>
<artifactId>jul-to-slf4j</artifactId>
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
index 52957fc..ccfe277 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
@@ -37,6 +37,11 @@ public class FlinkContainerWithPulsarEnvironment extends FlinkContainerTestEnvir
resourcePath("pulsar-client-all.jar"),
resourcePath("pulsar-client-api.jar"),
resourcePath("pulsar-admin-api.jar"),
+ resourcePath("bouncy-castle-bc.jar"),
+ resourcePath("bcpkix-jdk15on.jar"),
+ resourcePath("bcprov-jdk15on.jar"),
+ resourcePath("bcutil-jdk15on.jar"),
+ resourcePath("bcprov-ext-jdk15on.jar"),
resourcePath("jul-to-slf4j.jar"));
}
[flink-connector-pulsar] 17/31: [FLINK-27854][tests] Add support for FlinkContainers to work with both existing Docker images and flink-dist builds
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit 356cf239aab2b91391359961d71ab139276e06a2
Author: Alexander Fedulov <14...@users.noreply.github.com>
AuthorDate: Tue May 10 13:50:00 2022 +0200
[FLINK-27854][tests] Add support for FlinkContainers to work with both existing Docker images and flink-dist builds
- Refactors FlinkContainers initialization prpocess by introducing centralized configuration holders (FlinkContainersSettings and TestContainersSettings).
- Previously FlinkContainers workflow was based on always recreating images from local 'flink-dist'. This contribution adds support for external Docker images (required for externalized connectors repositories, where 'flink-dist' is not present).
---
.../tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
index 9a1c6dd..d2f9ab4 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
@@ -23,7 +23,7 @@ import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.tests.util.TestUtils;
-import org.apache.flink.tests.util.flink.FlinkContainerTestEnvironment;
+import org.apache.flink.tests.util.flink.container.FlinkContainerTestEnvironment;
/** A Flink Container which would bundles pulsar connector in its classpath. */
public class FlinkContainerWithPulsarEnvironment extends FlinkContainerTestEnvironment {
[flink-connector-pulsar] 22/31: [FLINK-26182][Connector/pulsar] Extract common logic from Pulsar source testing tools.
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit 712f6475bb564399202c9b6c3f5d54e5a6a4c755
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Tue Sep 27 18:04:24 2022 +0800
[FLINK-26182][Connector/pulsar] Extract common logic from Pulsar source testing tools.
---
flink-connector-pulsar-e2e-tests/pom.xml | 24 +++-
.../util/pulsar/PulsarSourceOrderedE2ECase.java | 5 +-
.../util/pulsar/PulsarSourceUnorderedE2ECase.java | 13 +-
.../pulsar/cases/KeySharedSubscriptionContext.java | 144 ---------------------
.../pulsar/cases/SharedSubscriptionContext.java | 116 -----------------
.../FlinkContainerWithPulsarEnvironment.java | 3 +-
.../common/KeyedPulsarPartitionDataWriter.java | 19 +--
.../common/UnorderedSourceTestSuiteBase.java | 86 ------------
.../ExclusiveSubscriptionContext.java | 19 +--
.../FailoverSubscriptionContext.java | 19 +--
.../source/KeySharedSubscriptionContext.java | 87 +++++++++++++
.../SharedSubscriptionContext.java} | 30 ++---
12 files changed, 151 insertions(+), 414 deletions(-)
diff --git a/flink-connector-pulsar-e2e-tests/pom.xml b/flink-connector-pulsar-e2e-tests/pom.xml
index c105a2b..6e6f7d3 100644
--- a/flink-connector-pulsar-e2e-tests/pom.xml
+++ b/flink-connector-pulsar-e2e-tests/pom.xml
@@ -33,6 +33,7 @@ under the License.
<properties>
<pulsar.version>2.10.0</pulsar.version>
<bouncycastle.version>1.69</bouncycastle.version>
+ <jaxb-api.version>2.3.1</jaxb-api.version>
</properties>
<dependencies>
@@ -41,6 +42,17 @@ under the License.
<artifactId>flink-end-to-end-tests-common</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-test-utils</artifactId>
+ <version>${project.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-pulsar</artifactId>
@@ -70,9 +82,6 @@ under the License.
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
- <configuration>
- <skip>true</skip>
- </configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
@@ -88,6 +97,15 @@ under the License.
</executions>
<configuration>
<artifactItems>
+ <artifactItem>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-test-utils</artifactId>
+ <version>${project.version}</version>
+ <destFileName>flink-connector-testing.jar</destFileName>
+ <type>jar</type>
+ <outputDirectory>${project.build.directory}/dependencies
+ </outputDirectory>
+ </artifactItem>
<artifactItem>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-pulsar</artifactId>
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
index 234c1a0..ea6a982 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
@@ -25,10 +25,10 @@ import org.apache.flink.connector.testframe.junit.annotations.TestExternalSystem
import org.apache.flink.connector.testframe.junit.annotations.TestSemantics;
import org.apache.flink.connector.testframe.testsuites.SourceTestSuiteBase;
import org.apache.flink.streaming.api.CheckpointingMode;
-import org.apache.flink.tests.util.pulsar.cases.ExclusiveSubscriptionContext;
-import org.apache.flink.tests.util.pulsar.cases.FailoverSubscriptionContext;
import org.apache.flink.tests.util.pulsar.common.FlinkContainerWithPulsarEnvironment;
import org.apache.flink.tests.util.pulsar.common.PulsarContainerTestEnvironment;
+import org.apache.flink.tests.util.pulsar.source.ExclusiveSubscriptionContext;
+import org.apache.flink.tests.util.pulsar.source.FailoverSubscriptionContext;
import org.apache.flink.testutils.junit.FailsOnJava11;
import org.junit.experimental.categories.Category;
@@ -37,6 +37,7 @@ import org.junit.experimental.categories.Category;
* Pulsar E2E test based on connector testing framework. It's used for Failover & Exclusive
* subscription.
*/
+@SuppressWarnings("unused")
@Category(value = {FailsOnJava11.class})
public class PulsarSourceOrderedE2ECase extends SourceTestSuiteBase<String> {
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
index 5039048..15333a1 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
@@ -19,21 +19,26 @@
package org.apache.flink.tests.util.pulsar;
import org.apache.flink.connector.pulsar.testutils.PulsarTestContextFactory;
+import org.apache.flink.connector.pulsar.testutils.source.UnorderedSourceTestSuiteBase;
import org.apache.flink.connector.testframe.junit.annotations.TestContext;
import org.apache.flink.connector.testframe.junit.annotations.TestEnv;
import org.apache.flink.connector.testframe.junit.annotations.TestExternalSystem;
import org.apache.flink.connector.testframe.junit.annotations.TestSemantics;
import org.apache.flink.streaming.api.CheckpointingMode;
-import org.apache.flink.tests.util.pulsar.cases.KeySharedSubscriptionContext;
-import org.apache.flink.tests.util.pulsar.cases.SharedSubscriptionContext;
import org.apache.flink.tests.util.pulsar.common.FlinkContainerWithPulsarEnvironment;
import org.apache.flink.tests.util.pulsar.common.PulsarContainerTestEnvironment;
-import org.apache.flink.tests.util.pulsar.common.UnorderedSourceTestSuiteBase;
+import org.apache.flink.tests.util.pulsar.source.KeySharedSubscriptionContext;
+import org.apache.flink.tests.util.pulsar.source.SharedSubscriptionContext;
+import org.apache.flink.testutils.junit.FailsOnJava11;
+
+import org.junit.experimental.categories.Category;
/**
* Pulsar E2E test based on connector testing framework. It's used for Shared & Key_Shared
* subscription.
*/
+@SuppressWarnings("unused")
+@Category(value = {FailsOnJava11.class})
public class PulsarSourceUnorderedE2ECase extends UnorderedSourceTestSuiteBase<String> {
// Defines the Semantic.
@@ -49,12 +54,10 @@ public class PulsarSourceUnorderedE2ECase extends UnorderedSourceTestSuiteBase<S
PulsarContainerTestEnvironment pulsar = new PulsarContainerTestEnvironment(flink);
// Defines a set of external context Factories for different test cases.
- @SuppressWarnings("unused")
@TestContext
PulsarTestContextFactory<String, SharedSubscriptionContext> shared =
new PulsarTestContextFactory<>(pulsar, SharedSubscriptionContext::new);
- @SuppressWarnings("unused")
@TestContext
PulsarTestContextFactory<String, KeySharedSubscriptionContext> keyShared =
new PulsarTestContextFactory<>(pulsar, KeySharedSubscriptionContext::new);
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java
deleted file mode 100644
index 5ad369b..0000000
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/KeySharedSubscriptionContext.java
+++ /dev/null
@@ -1,144 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.tests.util.pulsar.cases;
-
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.connector.source.Boundedness;
-import org.apache.flink.api.connector.source.Source;
-import org.apache.flink.connector.pulsar.source.PulsarSource;
-import org.apache.flink.connector.pulsar.source.PulsarSourceBuilder;
-import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
-import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils;
-import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange;
-import org.apache.flink.connector.pulsar.source.enumerator.topic.range.FixedRangeGenerator;
-import org.apache.flink.connector.pulsar.testutils.PulsarTestContext;
-import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
-import org.apache.flink.connector.testframe.external.ExternalSystemSplitDataWriter;
-import org.apache.flink.connector.testframe.external.source.TestingSourceSettings;
-import org.apache.flink.tests.util.pulsar.common.KeyedPulsarPartitionDataWriter;
-
-import org.apache.pulsar.client.api.RegexSubscriptionMode;
-import org.apache.pulsar.client.api.SubscriptionType;
-import org.apache.pulsar.common.util.Murmur3_32Hash;
-
-import java.net.URL;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-
-import static java.util.Collections.singletonList;
-import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
-import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange.RANGE_SIZE;
-import static org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema.pulsarSchema;
-import static org.apache.pulsar.client.api.Schema.STRING;
-
-/** We would consume from test splits by using {@link SubscriptionType#Key_Shared} subscription. */
-public class KeySharedSubscriptionContext extends PulsarTestContext<String> {
-
- private int index = 0;
-
- private final List<KeyedPulsarPartitionDataWriter> writers = new ArrayList<>();
-
- // Message keys.
- private final String key1;
- private final String key2;
-
- public KeySharedSubscriptionContext(PulsarTestEnvironment environment) {
- this(environment, Collections.emptyList());
- }
-
- public KeySharedSubscriptionContext(
- PulsarTestEnvironment environment, List<URL> connectorJarPaths) {
- super(environment, connectorJarPaths);
-
- // Init message keys.
- this.key1 = randomAlphabetic(8);
- String newKey2;
- do {
- newKey2 = randomAlphabetic(8);
- } while (keyHash(key1) == keyHash(newKey2));
- this.key2 = newKey2;
- }
-
- @Override
- protected String displayName() {
- return "consuming message by Key_Shared";
- }
-
- @Override
- public Source<String, ?, ?> createSource(TestingSourceSettings sourceSettings) {
- int keyHash = keyHash(key1);
- TopicRange range = new TopicRange(keyHash, keyHash);
-
- PulsarSourceBuilder<String> builder =
- PulsarSource.builder()
- .setDeserializationSchema(pulsarSchema(STRING))
- .setServiceUrl(operator.serviceUrl())
- .setAdminUrl(operator.adminUrl())
- .setTopicPattern(
- "pulsar-[0-9]+-key-shared", RegexSubscriptionMode.AllTopics)
- .setSubscriptionType(SubscriptionType.Key_Shared)
- .setSubscriptionName("pulsar-key-shared")
- .setRangeGenerator(new FixedRangeGenerator(singletonList(range)));
- if (sourceSettings.getBoundedness() == Boundedness.BOUNDED) {
- // Using latest stop cursor for making sure the source could be stopped.
- builder.setBoundedStopCursor(StopCursor.latest());
- }
-
- return builder.build();
- }
-
- @Override
- public ExternalSystemSplitDataWriter<String> createSourceSplitDataWriter(
- TestingSourceSettings sourceSettings) {
- String topicName = "pulsar-" + index + "-key-shared";
- operator.createTopic(topicName, 1);
- index++;
-
- String partitionName = TopicNameUtils.topicNameWithPartition(topicName, 0);
- KeyedPulsarPartitionDataWriter writer =
- new KeyedPulsarPartitionDataWriter(operator, partitionName, key1, key2);
- writers.add(writer);
-
- return writer;
- }
-
- @Override
- public List<String> generateTestData(
- TestingSourceSettings sourceSettings, int splitIndex, long seed) {
- return generateStringTestData(splitIndex, seed);
- }
-
- @Override
- public TypeInformation<String> getProducedType() {
- return TypeInformation.of(String.class);
- }
-
- @Override
- public void close() {
- for (KeyedPulsarPartitionDataWriter writer : writers) {
- writer.close();
- }
- writers.clear();
- }
-
- private int keyHash(String key) {
- return Murmur3_32Hash.getInstance().makeHash(key.getBytes()) % RANGE_SIZE;
- }
-}
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/SharedSubscriptionContext.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/SharedSubscriptionContext.java
deleted file mode 100644
index 1a2db66..0000000
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/SharedSubscriptionContext.java
+++ /dev/null
@@ -1,116 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.tests.util.pulsar.cases;
-
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.connector.source.Boundedness;
-import org.apache.flink.api.connector.source.Source;
-import org.apache.flink.connector.pulsar.source.PulsarSource;
-import org.apache.flink.connector.pulsar.source.PulsarSourceBuilder;
-import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
-import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils;
-import org.apache.flink.connector.pulsar.testutils.PulsarPartitionDataWriter;
-import org.apache.flink.connector.pulsar.testutils.PulsarTestContext;
-import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
-import org.apache.flink.connector.testframe.external.ExternalSystemSplitDataWriter;
-import org.apache.flink.connector.testframe.external.source.TestingSourceSettings;
-
-import org.apache.pulsar.client.api.RegexSubscriptionMode;
-import org.apache.pulsar.client.api.SubscriptionType;
-
-import java.net.URL;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-
-import static org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema.pulsarSchema;
-import static org.apache.pulsar.client.api.Schema.STRING;
-
-/** We would consuming from test splits by using {@link SubscriptionType#Shared} subscription. */
-public class SharedSubscriptionContext extends PulsarTestContext<String> {
-
- private int index = 0;
-
- private final List<PulsarPartitionDataWriter> writers = new ArrayList<>();
-
- public SharedSubscriptionContext(PulsarTestEnvironment environment) {
- this(environment, Collections.emptyList());
- }
-
- public SharedSubscriptionContext(
- PulsarTestEnvironment environment, List<URL> connectorJarPaths) {
- super(environment, connectorJarPaths);
- }
-
- @Override
- protected String displayName() {
- return "consuming message by Shared";
- }
-
- @Override
- public Source<String, ?, ?> createSource(TestingSourceSettings sourceSettings) {
- PulsarSourceBuilder<String> builder =
- PulsarSource.builder()
- .setDeserializationSchema(pulsarSchema(STRING))
- .setServiceUrl(operator.serviceUrl())
- .setAdminUrl(operator.adminUrl())
- .setTopicPattern("pulsar-[0-9]+-shared", RegexSubscriptionMode.AllTopics)
- .setSubscriptionType(SubscriptionType.Shared)
- .setSubscriptionName("pulsar-shared");
- if (sourceSettings.getBoundedness() == Boundedness.BOUNDED) {
- // Using latest stop cursor for making sure the source could be stopped.
- builder.setBoundedStopCursor(StopCursor.latest());
- }
-
- return builder.build();
- }
-
- @Override
- public ExternalSystemSplitDataWriter<String> createSourceSplitDataWriter(
- TestingSourceSettings sourceSettings) {
- String topicName = "pulsar-" + index + "-shared";
- operator.createTopic(topicName, 1);
- index++;
-
- String partitionName = TopicNameUtils.topicNameWithPartition(topicName, 0);
- PulsarPartitionDataWriter writer = new PulsarPartitionDataWriter(operator, partitionName);
- writers.add(writer);
-
- return writer;
- }
-
- @Override
- public List<String> generateTestData(
- TestingSourceSettings sourceSettings, int splitIndex, long seed) {
- return generateStringTestData(splitIndex, seed);
- }
-
- @Override
- public TypeInformation<String> getProducedType() {
- return TypeInformation.of(String.class);
- }
-
- @Override
- public void close() {
- for (PulsarPartitionDataWriter writer : writers) {
- writer.close();
- }
- writers.clear();
- }
-}
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
index 9f34554..f5e862f 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
@@ -43,7 +43,8 @@ public class FlinkContainerWithPulsarEnvironment extends FlinkContainerTestEnvir
resourcePath("bcutil-jdk15on.jar"),
resourcePath("bcprov-ext-jdk15on.jar"),
resourcePath("jaxb-api.jar"),
- resourcePath("jul-to-slf4j.jar"));
+ resourcePath("jul-to-slf4j.jar"),
+ resourcePath("flink-connector-testing.jar"));
}
private static String resourcePath(String jarName) {
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/KeyedPulsarPartitionDataWriter.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/KeyedPulsarPartitionDataWriter.java
index e431e4c..d5f6e11 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/KeyedPulsarPartitionDataWriter.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/KeyedPulsarPartitionDataWriter.java
@@ -35,23 +35,26 @@ public class KeyedPulsarPartitionDataWriter implements ExternalSystemSplitDataWr
private final PulsarRuntimeOperator operator;
private final String fullTopicName;
- private final String key1;
- private final String key2;
+ private final String keyToRead;
+ private final String keyToExclude;
public KeyedPulsarPartitionDataWriter(
- PulsarRuntimeOperator operator, String fullTopicName, String key1, String key2) {
+ PulsarRuntimeOperator operator,
+ String fullTopicName,
+ String keyToRead,
+ String keyToExclude) {
this.operator = operator;
this.fullTopicName = fullTopicName;
- this.key1 = key1;
- this.key2 = key2;
+ this.keyToRead = keyToRead;
+ this.keyToExclude = keyToExclude;
}
@Override
public void writeRecords(List<String> records) {
- operator.sendMessages(fullTopicName, Schema.STRING, key1, records);
+ List<String> newRecords = records.stream().map(a -> a + keyToRead).collect(toList());
- List<String> newRecords = records.stream().map(a -> a + key1).collect(toList());
- operator.sendMessages(fullTopicName, Schema.STRING, key2, newRecords);
+ operator.sendMessages(fullTopicName, Schema.STRING, keyToExclude, newRecords);
+ operator.sendMessages(fullTopicName, Schema.STRING, keyToRead, records);
}
@Override
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/UnorderedSourceTestSuiteBase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/UnorderedSourceTestSuiteBase.java
deleted file mode 100644
index 01527ea..0000000
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/UnorderedSourceTestSuiteBase.java
+++ /dev/null
@@ -1,86 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.tests.util.pulsar.common;
-
-import org.apache.flink.api.common.eventtime.WatermarkStrategy;
-import org.apache.flink.api.connector.source.Boundedness;
-import org.apache.flink.api.connector.source.Source;
-import org.apache.flink.connector.testframe.environment.TestEnvironment;
-import org.apache.flink.connector.testframe.environment.TestEnvironmentSettings;
-import org.apache.flink.connector.testframe.external.ExternalSystemSplitDataWriter;
-import org.apache.flink.connector.testframe.external.source.DataStreamSourceExternalContext;
-import org.apache.flink.connector.testframe.external.source.TestingSourceSettings;
-import org.apache.flink.connector.testframe.junit.extensions.ConnectorTestingExtension;
-import org.apache.flink.connector.testframe.junit.extensions.TestCaseInvocationContextProvider;
-import org.apache.flink.streaming.api.CheckpointingMode;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.util.TestLoggerExtension;
-
-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 java.util.List;
-import java.util.concurrent.ThreadLocalRandom;
-
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.containsInAnyOrder;
-
-/** A source test template for testing the messages which could be consumed in a unordered way. */
-@ExtendWith({
- ConnectorTestingExtension.class,
- TestLoggerExtension.class,
- TestCaseInvocationContextProvider.class
-})
-@TestInstance(TestInstance.Lifecycle.PER_CLASS)
-public abstract class UnorderedSourceTestSuiteBase<T> {
-
- @TestTemplate
- @DisplayName("Test source with one split and four consumers")
- public void testOneSplitWithMultipleConsumers(
- TestEnvironment testEnv, DataStreamSourceExternalContext<T> externalContext)
- throws Exception {
- TestingSourceSettings sourceSettings =
- TestingSourceSettings.builder()
- .setBoundedness(Boundedness.BOUNDED)
- .setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
- .build();
- TestEnvironmentSettings envOptions =
- TestEnvironmentSettings.builder()
- .setConnectorJarPaths(externalContext.getConnectorJarPaths())
- .build();
- List<T> testData =
- externalContext.generateTestData(
- sourceSettings, 0, ThreadLocalRandom.current().nextLong());
- ExternalSystemSplitDataWriter<T> writer =
- externalContext.createSourceSplitDataWriter(sourceSettings);
- writer.writeRecords(testData);
-
- Source<T, ?, ?> source = externalContext.createSource(sourceSettings);
- StreamExecutionEnvironment execEnv = testEnv.createExecutionEnvironment(envOptions);
- List<T> results =
- execEnv.fromSource(source, WatermarkStrategy.noWatermarks(), "Pulsar source")
- .setParallelism(4)
- .executeAndCollect(
- "Source single split with four readers.", testData.size());
-
- assertThat(results, containsInAnyOrder(testData.toArray()));
- }
-}
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/ExclusiveSubscriptionContext.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/source/ExclusiveSubscriptionContext.java
similarity index 71%
rename from flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/ExclusiveSubscriptionContext.java
rename to flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/source/ExclusiveSubscriptionContext.java
index 6fea0c9..4906ad6 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/ExclusiveSubscriptionContext.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/source/ExclusiveSubscriptionContext.java
@@ -16,32 +16,23 @@
* limitations under the License.
*/
-package org.apache.flink.tests.util.pulsar.cases;
+package org.apache.flink.tests.util.pulsar.source;
import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
-import org.apache.flink.connector.pulsar.testutils.cases.MultipleTopicTemplateContext;
+import org.apache.flink.connector.pulsar.testutils.source.cases.MultipleTopicConsumingContext;
import org.apache.pulsar.client.api.SubscriptionType;
-import java.net.URL;
-import java.util.Collections;
-import java.util.List;
-
/** We would consume from test splits by using {@link SubscriptionType#Exclusive} subscription. */
-public class ExclusiveSubscriptionContext extends MultipleTopicTemplateContext {
+public class ExclusiveSubscriptionContext extends MultipleTopicConsumingContext {
public ExclusiveSubscriptionContext(PulsarTestEnvironment environment) {
- this(environment, Collections.emptyList());
- }
-
- public ExclusiveSubscriptionContext(
- PulsarTestEnvironment environment, List<URL> connectorJarPaths) {
- super(environment, connectorJarPaths);
+ super(environment);
}
@Override
protected String displayName() {
- return "consuming message by Exclusive";
+ return "consume message by Exclusive";
}
@Override
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/FailoverSubscriptionContext.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/source/FailoverSubscriptionContext.java
similarity index 71%
copy from flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/FailoverSubscriptionContext.java
copy to flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/source/FailoverSubscriptionContext.java
index c473488..3134db4 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/FailoverSubscriptionContext.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/source/FailoverSubscriptionContext.java
@@ -16,32 +16,23 @@
* limitations under the License.
*/
-package org.apache.flink.tests.util.pulsar.cases;
+package org.apache.flink.tests.util.pulsar.source;
import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
-import org.apache.flink.connector.pulsar.testutils.cases.MultipleTopicTemplateContext;
+import org.apache.flink.connector.pulsar.testutils.source.cases.MultipleTopicConsumingContext;
import org.apache.pulsar.client.api.SubscriptionType;
-import java.net.URL;
-import java.util.Collections;
-import java.util.List;
-
/** We would consume from test splits by using {@link SubscriptionType#Failover} subscription. */
-public class FailoverSubscriptionContext extends MultipleTopicTemplateContext {
+public class FailoverSubscriptionContext extends MultipleTopicConsumingContext {
public FailoverSubscriptionContext(PulsarTestEnvironment environment) {
- this(environment, Collections.emptyList());
- }
-
- public FailoverSubscriptionContext(
- PulsarTestEnvironment environment, List<URL> connectorJarPaths) {
- super(environment, connectorJarPaths);
+ super(environment);
}
@Override
protected String displayName() {
- return "consuming message by Failover";
+ return "consume message by Failover";
}
@Override
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/source/KeySharedSubscriptionContext.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/source/KeySharedSubscriptionContext.java
new file mode 100644
index 0000000..0cae6e5
--- /dev/null
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/source/KeySharedSubscriptionContext.java
@@ -0,0 +1,87 @@
+/*
+ * 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.pulsar.source;
+
+import org.apache.flink.connector.pulsar.source.PulsarSourceBuilder;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.range.FixedRangeGenerator;
+import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
+import org.apache.flink.connector.pulsar.testutils.source.cases.MultipleTopicConsumingContext;
+import org.apache.flink.connector.testframe.external.ExternalSystemSplitDataWriter;
+import org.apache.flink.connector.testframe.external.source.TestingSourceSettings;
+import org.apache.flink.tests.util.pulsar.common.KeyedPulsarPartitionDataWriter;
+
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.common.util.Murmur3_32Hash;
+
+import static java.util.Collections.singletonList;
+import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
+import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange.RANGE_SIZE;
+import static org.apache.pulsar.client.api.SubscriptionType.Key_Shared;
+
+/** We would consume from test splits by using {@link SubscriptionType#Key_Shared} subscription. */
+public class KeySharedSubscriptionContext extends MultipleTopicConsumingContext {
+
+ private final String keyToRead;
+ private final String keyToExclude;
+
+ public KeySharedSubscriptionContext(PulsarTestEnvironment environment) {
+ super(environment, Key_Shared);
+
+ this.keyToRead = randomAlphabetic(8);
+
+ // Make sure they have different hash code.
+ int readHash = keyHash(keyToRead);
+ String randomKey;
+ do {
+ randomKey = randomAlphabetic(8);
+ } while (keyHash(randomKey) == readHash);
+ this.keyToExclude = randomKey;
+ }
+
+ @Override
+ public ExternalSystemSplitDataWriter<String> createSourceSplitDataWriter(
+ TestingSourceSettings sourceSettings) {
+ String partitionName = generatePartitionName();
+ return new KeyedPulsarPartitionDataWriter(operator, partitionName, keyToRead, keyToExclude);
+ }
+
+ @Override
+ protected String displayName() {
+ return "consume message by Key_Shared";
+ }
+
+ @Override
+ protected void setSourceBuilder(PulsarSourceBuilder<String> builder) {
+ int keyHash = keyHash(keyToRead);
+ TopicRange range = new TopicRange(keyHash, keyHash);
+
+ builder.setRangeGenerator(new FixedRangeGenerator(singletonList(range)));
+ }
+
+ @Override
+ protected String subscriptionName() {
+ return "pulsar-key-shared-subscription";
+ }
+
+ // This method is copied from Pulsar for calculating message key hash.
+ private int keyHash(String key) {
+ return Murmur3_32Hash.getInstance().makeHash(key.getBytes()) % RANGE_SIZE;
+ }
+}
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/FailoverSubscriptionContext.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/source/SharedSubscriptionContext.java
similarity index 57%
rename from flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/FailoverSubscriptionContext.java
rename to flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/source/SharedSubscriptionContext.java
index c473488..fe9f078 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/cases/FailoverSubscriptionContext.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/source/SharedSubscriptionContext.java
@@ -16,41 +16,29 @@
* limitations under the License.
*/
-package org.apache.flink.tests.util.pulsar.cases;
+package org.apache.flink.tests.util.pulsar.source;
import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
-import org.apache.flink.connector.pulsar.testutils.cases.MultipleTopicTemplateContext;
+import org.apache.flink.connector.pulsar.testutils.source.cases.MultipleTopicConsumingContext;
import org.apache.pulsar.client.api.SubscriptionType;
-import java.net.URL;
-import java.util.Collections;
-import java.util.List;
+import static org.apache.pulsar.client.api.SubscriptionType.Shared;
-/** We would consume from test splits by using {@link SubscriptionType#Failover} subscription. */
-public class FailoverSubscriptionContext extends MultipleTopicTemplateContext {
+/** We would consume from test splits by using {@link SubscriptionType#Shared} subscription. */
+public class SharedSubscriptionContext extends MultipleTopicConsumingContext {
- public FailoverSubscriptionContext(PulsarTestEnvironment environment) {
- this(environment, Collections.emptyList());
- }
-
- public FailoverSubscriptionContext(
- PulsarTestEnvironment environment, List<URL> connectorJarPaths) {
- super(environment, connectorJarPaths);
+ public SharedSubscriptionContext(PulsarTestEnvironment environment) {
+ super(environment, Shared);
}
@Override
protected String displayName() {
- return "consuming message by Failover";
+ return "consume message by Shared";
}
@Override
protected String subscriptionName() {
- return "pulsar-failover-subscription";
- }
-
- @Override
- protected SubscriptionType subscriptionType() {
- return SubscriptionType.Failover;
+ return "pulsar-shared-subscription";
}
}
[flink-connector-pulsar] 11/31: [FLINK-25288][tests][pulsar] Make pulsar e2e tests more robust
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit 6971aa180a580afc4dbb0f4ec7e7ae64f0572cd8
Author: Leonard Xu <xb...@gmail.com>
AuthorDate: Tue Feb 15 01:24:03 2022 +0800
[FLINK-25288][tests][pulsar] Make pulsar e2e tests more robust
This closes #18516.
---
.../common/FlinkContainerWithPulsarEnvironment.java | 16 ++++++++++++----
1 file changed, 12 insertions(+), 4 deletions(-)
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
index 14feeed..52957fc 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
@@ -19,12 +19,12 @@
package org.apache.flink.tests.util.pulsar.common;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.MemorySize;
+import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.tests.util.TestUtils;
import org.apache.flink.tests.util.flink.FlinkContainerTestEnvironment;
-import static org.apache.flink.configuration.TaskManagerOptions.TASK_OFF_HEAP_MEMORY;
-
/** A Flink Container which would bundles pulsar connector in its classpath. */
public class FlinkContainerWithPulsarEnvironment extends FlinkContainerTestEnvironment {
@@ -46,8 +46,16 @@ public class FlinkContainerWithPulsarEnvironment extends FlinkContainerTestEnvir
protected static Configuration flinkConfiguration() {
Configuration configuration = new Configuration();
- // Increase the off heap memory for avoiding direct buffer memory error on Pulsar e2e tests.
- configuration.set(TASK_OFF_HEAP_MEMORY, MemorySize.ofMebiBytes(100));
+ // Increase the off heap memory of TaskManager to avoid direct buffer memory error in Pulsar
+ // e2e tests.
+ configuration.set(TaskManagerOptions.TASK_OFF_HEAP_MEMORY, MemorySize.ofMebiBytes(100));
+
+ // Increase the jvm metaspace memory to avoid java.lang.OutOfMemoryError: Metaspace
+ configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(2048));
+ configuration.set(TaskManagerOptions.JVM_METASPACE, MemorySize.ofMebiBytes(512));
+ configuration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(2048));
+ configuration.set(JobManagerOptions.JVM_METASPACE, MemorySize.ofMebiBytes(512));
+
return configuration;
}
}
[flink-connector-pulsar] 02/31: [FLINK-24018][build] Remove Scala dependencies from Java APIs
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit db11068e1900c0486a675a3059e7f31ecfed981d
Author: Chesnay Schepler <ch...@apache.org>
AuthorDate: Tue Oct 26 00:48:06 2021 +0200
[FLINK-24018][build] Remove Scala dependencies from Java APIs
---
flink-connector-pulsar-e2e-tests/pom.xml | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
diff --git a/flink-connector-pulsar-e2e-tests/pom.xml b/flink-connector-pulsar-e2e-tests/pom.xml
index 269f89c..e7caf8b 100644
--- a/flink-connector-pulsar-e2e-tests/pom.xml
+++ b/flink-connector-pulsar-e2e-tests/pom.xml
@@ -42,12 +42,12 @@ under the License.
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-pulsar_${scala.binary.version}</artifactId>
+ <artifactId>flink-connector-pulsar</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-pulsar_${scala.binary.version}</artifactId>
+ <artifactId>flink-connector-pulsar</artifactId>
<version>${project.version}</version>
<type>test-jar</type>
</dependency>
@@ -75,7 +75,7 @@ under the License.
<artifactItems>
<artifactItem>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-pulsar_${scala.binary.version}</artifactId>
+ <artifactId>flink-connector-pulsar</artifactId>
<version>${project.version}</version>
<destFileName>pulsar-connector.jar</destFileName>
<type>jar</type>
[flink-connector-pulsar] 31/31: [hotfix][architecture] Add ProductionCodeArchitectureTest
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit ded956e8535d39b8ec604a896baa66508b54240d
Author: Martijn Visser <ma...@apache.org>
AuthorDate: Fri Dec 9 16:01:54 2022 +0100
[hotfix][architecture] Add ProductionCodeArchitectureTest
---
.../ProductionCodeArchitectureTest.java | 41 ++++++++++++++++++++++
1 file changed, 41 insertions(+)
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/architecture/ProductionCodeArchitectureTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/architecture/ProductionCodeArchitectureTest.java
new file mode 100644
index 0000000..b88b338
--- /dev/null
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/architecture/ProductionCodeArchitectureTest.java
@@ -0,0 +1,41 @@
+/*
+ * 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.architecture;
+
+import org.apache.flink.architecture.common.ImportOptions;
+
+import com.tngtech.archunit.core.importer.ImportOption;
+import com.tngtech.archunit.junit.AnalyzeClasses;
+import com.tngtech.archunit.junit.ArchTest;
+import com.tngtech.archunit.junit.ArchTests;
+
+/** production code for Architecture tests. */
+@AnalyzeClasses(
+ packages = "org.apache.flink.connector.pulsar",
+ importOptions = {
+ ImportOption.DoNotIncludeTests.class,
+ ImportOption.DoNotIncludeArchives.class,
+ ImportOptions.ExcludeScalaImportOption.class,
+ ImportOptions.ExcludeShadedImportOption.class
+ })
+public class ProductionCodeArchitectureTest {
+
+ @ArchTest
+ public static final ArchTests COMMON_TESTS = ArchTests.in(ProductionCodeArchitectureBase.class);
+}
[flink-connector-pulsar] 05/31: [FLINK-25159][tests] Disable broken E2E tests
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit cacdffb54b1462f6b5c9a7de61b82de6f1893888
Author: Chesnay Schepler <ch...@apache.org>
AuthorDate: Mon Dec 6 14:59:53 2021 +0100
[FLINK-25159][tests] Disable broken E2E tests
---
.../org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java | 3 +++
.../apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java | 3 +++
2 files changed, 6 insertions(+)
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
index 9a499e6..95c1486 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
@@ -28,12 +28,15 @@ import org.apache.flink.tests.util.pulsar.cases.ExclusiveSubscriptionContext;
import org.apache.flink.tests.util.pulsar.cases.FailoverSubscriptionContext;
import org.apache.flink.tests.util.pulsar.common.FlinkContainerWithPulsarEnvironment;
+import org.junit.jupiter.api.Disabled;
+
import static org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntime.container;
/**
* Pulsar E2E test based on connector testing framework. It's used for Failover & Exclusive
* subscription.
*/
+@Disabled("FLINK-25210")
public class PulsarSourceOrderedE2ECase extends SourceTestSuiteBase<String> {
// Defines TestEnvironment.
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
index 797c7b1..c59831f 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
@@ -28,12 +28,15 @@ import org.apache.flink.tests.util.pulsar.cases.SharedSubscriptionContext;
import org.apache.flink.tests.util.pulsar.common.FlinkContainerWithPulsarEnvironment;
import org.apache.flink.tests.util.pulsar.common.UnorderedSourceTestSuiteBase;
+import org.junit.jupiter.api.Disabled;
+
import static org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntime.container;
/**
* Pulsar E2E test based on connector testing framework. It's used for Shared & Key_Shared
* subscription.
*/
+@Disabled("FLINK-25210")
public class PulsarSourceUnorderedE2ECase extends UnorderedSourceTestSuiteBase<String> {
// Defines TestEnvironment.
[flink-connector-pulsar] 26/31: [FLINK-29709][Connector/Pulsar] Bump the Pulsar to latest 2.10.2 (#21204)
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit 90044473b9928b4dce7707f625510ab4e6bbd60f
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Mon Nov 7 11:44:15 2022 +0800
[FLINK-29709][Connector/Pulsar] Bump the Pulsar to latest 2.10.2 (#21204)
---
flink-connector-pulsar-e2e-tests/pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/flink-connector-pulsar-e2e-tests/pom.xml b/flink-connector-pulsar-e2e-tests/pom.xml
index 5ba8508..522d254 100644
--- a/flink-connector-pulsar-e2e-tests/pom.xml
+++ b/flink-connector-pulsar-e2e-tests/pom.xml
@@ -31,7 +31,7 @@ under the License.
<name>Flink : E2E Tests : Pulsar</name>
<properties>
- <pulsar.version>2.10.1</pulsar.version>
+ <pulsar.version>2.10.2</pulsar.version>
<bouncycastle.version>1.69</bouncycastle.version>
<jaxb-api.version>2.3.1</jaxb-api.version>
</properties>
[flink-connector-pulsar] 20/31: Update version to 1.17-SNAPSHOT
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit f4a7fa0616e856209544fa3294c8eb5f1d6c7a96
Author: huangxingbo <hx...@apache.org>
AuthorDate: Mon Sep 5 10:43:15 2022 +0800
Update version to 1.17-SNAPSHOT
---
flink-connector-pulsar-e2e-tests/pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/flink-connector-pulsar-e2e-tests/pom.xml b/flink-connector-pulsar-e2e-tests/pom.xml
index b9bc5d4..c105a2b 100644
--- a/flink-connector-pulsar-e2e-tests/pom.xml
+++ b/flink-connector-pulsar-e2e-tests/pom.xml
@@ -23,7 +23,7 @@ under the License.
<parent>
<artifactId>flink-end-to-end-tests</artifactId>
<groupId>org.apache.flink</groupId>
- <version>1.16-SNAPSHOT</version>
+ <version>1.17-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
[flink-connector-pulsar] 23/31: [FLINK-26182][Connector/pulsar] Create e2e tests for the Pulsar source and sink based on the connector testing framework.
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit 0ce62802e977745367c7bfe38b4c5ca1853ca604
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Wed Sep 28 15:28:14 2022 +0800
[FLINK-26182][Connector/pulsar] Create e2e tests for the Pulsar source and sink based on the connector testing framework.
---
...norderedE2ECase.java => PulsarSinkE2ECase.java} | 31 +++-----
.../util/pulsar/PulsarSourceUnorderedE2ECase.java | 4 +-
.../FlinkContainerWithPulsarEnvironment.java | 5 +-
.../common/KeyedPulsarPartitionDataWriter.java | 64 ----------------
.../source/KeySharedSubscriptionContext.java | 87 ----------------------
.../pulsar/source/SharedSubscriptionContext.java | 44 -----------
6 files changed, 15 insertions(+), 220 deletions(-)
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSinkE2ECase.java
similarity index 64%
copy from flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
copy to flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSinkE2ECase.java
index 15333a1..a19f593 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSinkE2ECase.java
@@ -19,7 +19,8 @@
package org.apache.flink.tests.util.pulsar;
import org.apache.flink.connector.pulsar.testutils.PulsarTestContextFactory;
-import org.apache.flink.connector.pulsar.testutils.source.UnorderedSourceTestSuiteBase;
+import org.apache.flink.connector.pulsar.testutils.sink.PulsarSinkTestContext;
+import org.apache.flink.connector.pulsar.testutils.sink.PulsarSinkTestSuiteBase;
import org.apache.flink.connector.testframe.junit.annotations.TestContext;
import org.apache.flink.connector.testframe.junit.annotations.TestEnv;
import org.apache.flink.connector.testframe.junit.annotations.TestExternalSystem;
@@ -27,38 +28,30 @@ import org.apache.flink.connector.testframe.junit.annotations.TestSemantics;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.tests.util.pulsar.common.FlinkContainerWithPulsarEnvironment;
import org.apache.flink.tests.util.pulsar.common.PulsarContainerTestEnvironment;
-import org.apache.flink.tests.util.pulsar.source.KeySharedSubscriptionContext;
-import org.apache.flink.tests.util.pulsar.source.SharedSubscriptionContext;
import org.apache.flink.testutils.junit.FailsOnJava11;
import org.junit.experimental.categories.Category;
-/**
- * Pulsar E2E test based on connector testing framework. It's used for Shared & Key_Shared
- * subscription.
- */
+/** Pulsar sink E2E test based on connector testing framework. */
@SuppressWarnings("unused")
@Category(value = {FailsOnJava11.class})
-public class PulsarSourceUnorderedE2ECase extends UnorderedSourceTestSuiteBase<String> {
+public class PulsarSinkE2ECase extends PulsarSinkTestSuiteBase {
- // Defines the Semantic.
@TestSemantics
- CheckpointingMode[] semantics = new CheckpointingMode[] {CheckpointingMode.EXACTLY_ONCE};
+ CheckpointingMode[] semantics =
+ new CheckpointingMode[] {
+ CheckpointingMode.EXACTLY_ONCE, CheckpointingMode.AT_LEAST_ONCE
+ };
- // Defines TestEnvironment.
+ // Defines TestEnvironment
@TestEnv
- FlinkContainerWithPulsarEnvironment flink = new FlinkContainerWithPulsarEnvironment(1, 8);
+ FlinkContainerWithPulsarEnvironment flink = new FlinkContainerWithPulsarEnvironment(1, 6);
// Defines ConnectorExternalSystem.
@TestExternalSystem
PulsarContainerTestEnvironment pulsar = new PulsarContainerTestEnvironment(flink);
- // Defines a set of external context Factories for different test cases.
- @TestContext
- PulsarTestContextFactory<String, SharedSubscriptionContext> shared =
- new PulsarTestContextFactory<>(pulsar, SharedSubscriptionContext::new);
-
@TestContext
- PulsarTestContextFactory<String, KeySharedSubscriptionContext> keyShared =
- new PulsarTestContextFactory<>(pulsar, KeySharedSubscriptionContext::new);
+ PulsarTestContextFactory<String, PulsarSinkTestContext> sinkContext =
+ new PulsarTestContextFactory<>(pulsar, PulsarSinkTestContext::new);
}
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
index 15333a1..89692d1 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
@@ -20,6 +20,8 @@ package org.apache.flink.tests.util.pulsar;
import org.apache.flink.connector.pulsar.testutils.PulsarTestContextFactory;
import org.apache.flink.connector.pulsar.testutils.source.UnorderedSourceTestSuiteBase;
+import org.apache.flink.connector.pulsar.testutils.source.cases.KeySharedSubscriptionContext;
+import org.apache.flink.connector.pulsar.testutils.source.cases.SharedSubscriptionContext;
import org.apache.flink.connector.testframe.junit.annotations.TestContext;
import org.apache.flink.connector.testframe.junit.annotations.TestEnv;
import org.apache.flink.connector.testframe.junit.annotations.TestExternalSystem;
@@ -27,8 +29,6 @@ import org.apache.flink.connector.testframe.junit.annotations.TestSemantics;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.tests.util.pulsar.common.FlinkContainerWithPulsarEnvironment;
import org.apache.flink.tests.util.pulsar.common.PulsarContainerTestEnvironment;
-import org.apache.flink.tests.util.pulsar.source.KeySharedSubscriptionContext;
-import org.apache.flink.tests.util.pulsar.source.SharedSubscriptionContext;
import org.apache.flink.testutils.junit.FailsOnJava11;
import org.junit.experimental.categories.Category;
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
index f5e862f..65e99a8 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
@@ -51,11 +51,8 @@ public class FlinkContainerWithPulsarEnvironment extends FlinkContainerTestEnvir
return ResourceTestUtils.getResource(jarName).toAbsolutePath().toString();
}
- protected static Configuration flinkConfiguration() {
+ private static Configuration flinkConfiguration() {
Configuration configuration = new Configuration();
- // Increase the off heap memory of TaskManager to avoid direct buffer memory error in Pulsar
- // e2e tests.
- configuration.set(TaskManagerOptions.TASK_OFF_HEAP_MEMORY, MemorySize.ofMebiBytes(100));
// Increase the jvm metaspace memory to avoid java.lang.OutOfMemoryError: Metaspace
configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(2048));
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/KeyedPulsarPartitionDataWriter.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/KeyedPulsarPartitionDataWriter.java
deleted file mode 100644
index d5f6e11..0000000
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/KeyedPulsarPartitionDataWriter.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.tests.util.pulsar.common;
-
-import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntimeOperator;
-import org.apache.flink.connector.testframe.external.ExternalSystemSplitDataWriter;
-
-import org.apache.pulsar.client.api.Schema;
-
-import java.util.List;
-
-import static java.util.stream.Collectors.toList;
-
-/**
- * Source split data writer for writing test data into a Pulsar topic partition. It will write the
- * message with two keys.
- */
-public class KeyedPulsarPartitionDataWriter implements ExternalSystemSplitDataWriter<String> {
-
- private final PulsarRuntimeOperator operator;
- private final String fullTopicName;
- private final String keyToRead;
- private final String keyToExclude;
-
- public KeyedPulsarPartitionDataWriter(
- PulsarRuntimeOperator operator,
- String fullTopicName,
- String keyToRead,
- String keyToExclude) {
- this.operator = operator;
- this.fullTopicName = fullTopicName;
- this.keyToRead = keyToRead;
- this.keyToExclude = keyToExclude;
- }
-
- @Override
- public void writeRecords(List<String> records) {
- List<String> newRecords = records.stream().map(a -> a + keyToRead).collect(toList());
-
- operator.sendMessages(fullTopicName, Schema.STRING, keyToExclude, newRecords);
- operator.sendMessages(fullTopicName, Schema.STRING, keyToRead, records);
- }
-
- @Override
- public void close() {
- // Nothing to do.
- }
-}
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/source/KeySharedSubscriptionContext.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/source/KeySharedSubscriptionContext.java
deleted file mode 100644
index 0cae6e5..0000000
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/source/KeySharedSubscriptionContext.java
+++ /dev/null
@@ -1,87 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.tests.util.pulsar.source;
-
-import org.apache.flink.connector.pulsar.source.PulsarSourceBuilder;
-import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange;
-import org.apache.flink.connector.pulsar.source.enumerator.topic.range.FixedRangeGenerator;
-import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
-import org.apache.flink.connector.pulsar.testutils.source.cases.MultipleTopicConsumingContext;
-import org.apache.flink.connector.testframe.external.ExternalSystemSplitDataWriter;
-import org.apache.flink.connector.testframe.external.source.TestingSourceSettings;
-import org.apache.flink.tests.util.pulsar.common.KeyedPulsarPartitionDataWriter;
-
-import org.apache.pulsar.client.api.SubscriptionType;
-import org.apache.pulsar.common.util.Murmur3_32Hash;
-
-import static java.util.Collections.singletonList;
-import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
-import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange.RANGE_SIZE;
-import static org.apache.pulsar.client.api.SubscriptionType.Key_Shared;
-
-/** We would consume from test splits by using {@link SubscriptionType#Key_Shared} subscription. */
-public class KeySharedSubscriptionContext extends MultipleTopicConsumingContext {
-
- private final String keyToRead;
- private final String keyToExclude;
-
- public KeySharedSubscriptionContext(PulsarTestEnvironment environment) {
- super(environment, Key_Shared);
-
- this.keyToRead = randomAlphabetic(8);
-
- // Make sure they have different hash code.
- int readHash = keyHash(keyToRead);
- String randomKey;
- do {
- randomKey = randomAlphabetic(8);
- } while (keyHash(randomKey) == readHash);
- this.keyToExclude = randomKey;
- }
-
- @Override
- public ExternalSystemSplitDataWriter<String> createSourceSplitDataWriter(
- TestingSourceSettings sourceSettings) {
- String partitionName = generatePartitionName();
- return new KeyedPulsarPartitionDataWriter(operator, partitionName, keyToRead, keyToExclude);
- }
-
- @Override
- protected String displayName() {
- return "consume message by Key_Shared";
- }
-
- @Override
- protected void setSourceBuilder(PulsarSourceBuilder<String> builder) {
- int keyHash = keyHash(keyToRead);
- TopicRange range = new TopicRange(keyHash, keyHash);
-
- builder.setRangeGenerator(new FixedRangeGenerator(singletonList(range)));
- }
-
- @Override
- protected String subscriptionName() {
- return "pulsar-key-shared-subscription";
- }
-
- // This method is copied from Pulsar for calculating message key hash.
- private int keyHash(String key) {
- return Murmur3_32Hash.getInstance().makeHash(key.getBytes()) % RANGE_SIZE;
- }
-}
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/source/SharedSubscriptionContext.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/source/SharedSubscriptionContext.java
deleted file mode 100644
index fe9f078..0000000
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/source/SharedSubscriptionContext.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.tests.util.pulsar.source;
-
-import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
-import org.apache.flink.connector.pulsar.testutils.source.cases.MultipleTopicConsumingContext;
-
-import org.apache.pulsar.client.api.SubscriptionType;
-
-import static org.apache.pulsar.client.api.SubscriptionType.Shared;
-
-/** We would consume from test splits by using {@link SubscriptionType#Shared} subscription. */
-public class SharedSubscriptionContext extends MultipleTopicConsumingContext {
-
- public SharedSubscriptionContext(PulsarTestEnvironment environment) {
- super(environment, Shared);
- }
-
- @Override
- protected String displayName() {
- return "consume message by Shared";
- }
-
- @Override
- protected String subscriptionName() {
- return "pulsar-shared-subscription";
- }
-}
[flink-connector-pulsar] 28/31: [FLINK-30336][Connector/Pulsar] Sync missing Pulsar e2e tests to external repo
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit 858931e29e4bcf67f2c695cc742c90652dfb6352
Author: Martijn Visser <ma...@apache.org>
AuthorDate: Fri Dec 9 12:54:22 2022 +0100
[FLINK-30336][Connector/Pulsar] Sync missing Pulsar e2e tests to external repo
---
flink-connector-pulsar-e2e-tests/pom.xml | 80 +++++++++++++++++++++++++++-----
flink-connector-pulsar/pom.xml | 1 -
pom.xml | 2 +
3 files changed, 70 insertions(+), 13 deletions(-)
diff --git a/flink-connector-pulsar-e2e-tests/pom.xml b/flink-connector-pulsar-e2e-tests/pom.xml
index 522d254..7a6e6ce 100644
--- a/flink-connector-pulsar-e2e-tests/pom.xml
+++ b/flink-connector-pulsar-e2e-tests/pom.xml
@@ -21,9 +21,9 @@ under the License.
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
- <artifactId>flink-end-to-end-tests</artifactId>
<groupId>org.apache.flink</groupId>
- <version>1.17-SNAPSHOT</version>
+ <artifactId>flink-connector-pulsar-parent</artifactId>
+ <version>3.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
@@ -31,21 +31,15 @@ under the License.
<name>Flink : E2E Tests : Pulsar</name>
<properties>
- <pulsar.version>2.10.2</pulsar.version>
- <bouncycastle.version>1.69</bouncycastle.version>
<jaxb-api.version>2.3.1</jaxb-api.version>
+ <japicmp.skip>true</japicmp.skip>
</properties>
<dependencies>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-end-to-end-tests-common</artifactId>
- <version>${project.version}</version>
- </dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-test-utils</artifactId>
- <version>${project.version}</version>
+ <version>${flink.version}</version>
<exclusions>
<exclusion>
<groupId>com.google.guava</groupId>
@@ -63,7 +57,7 @@ under the License.
<dependency>
<groupId>javax.xml.bind</groupId>
<artifactId>jaxb-api</artifactId>
- <version>${jaxb.api.version}</version>
+ <version>${jaxb-api.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
@@ -77,12 +71,74 @@ under the License.
</dependency>
</dependencies>
+ <profiles>
+ <profile>
+ <id>run-end-to-end-tests</id>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>end-to-end-tests</id>
+ <phase>integration-test</phase>
+ <goals>
+ <goal>test</goal>
+ </goals>
+ <configuration>
+ <includes>
+ <include>**/*.*</include>
+ </includes>
+ <excludedGroups>${excludeE2E}</excludedGroups>
+ <systemPropertyVariables>
+ <moduleDir>${project.basedir}</moduleDir>
+ </systemPropertyVariables>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ <profile>
+ <id>java11</id>
+ <activation>
+ <jdk>[11,)</jdk>
+ </activation>
+ <properties>
+ <excludeE2E>org.apache.flink.testutils.junit.FailsOnJava11</excludeE2E>
+ </properties>
+ </profile>
+ </profiles>
+
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
</plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-deploy-plugin</artifactId>
+ <configuration>
+ <skip>true</skip>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>default-test</id>
+ <phase>none</phase>
+ </execution>
+ <execution>
+ <id>integration-tests</id>
+ <phase>none</phase>
+ </execution>
+ </executions>
+ </plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-dependency-plugin</artifactId>
@@ -100,7 +156,7 @@ under the License.
<artifactItem>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-test-utils</artifactId>
- <version>${project.version}</version>
+ <version>${flink.version}</version>
<destFileName>flink-connector-testing.jar</destFileName>
<type>jar</type>
<outputDirectory>${project.build.directory}/dependencies
diff --git a/flink-connector-pulsar/pom.xml b/flink-connector-pulsar/pom.xml
index 5881495..68eab5a 100644
--- a/flink-connector-pulsar/pom.xml
+++ b/flink-connector-pulsar/pom.xml
@@ -35,7 +35,6 @@ under the License.
<packaging>jar</packaging>
<properties>
- <pulsar.version>2.10.2</pulsar.version>
<protoc.version>3.21.7</protoc.version>
<!-- Test Libraries -->
diff --git a/pom.xml b/pom.xml
index 64472a7..baf6aac 100644
--- a/pom.xml
+++ b/pom.xml
@@ -45,10 +45,12 @@ under the License.
<modules>
<module>flink-connector-pulsar</module>
<module>flink-sql-connector-pulsar</module>
+ <module>flink-connector-pulsar-e2e-tests</module>
</modules>
<properties>
<flink.version>1.17-SNAPSHOT</flink.version>
+ <pulsar.version>2.10.2</pulsar.version>
<jackson-bom.version>2.13.4.20221013</jackson-bom.version>
<grpc-bom.version>1.45.1</grpc-bom.version>
[flink-connector-pulsar] 10/31: [FLINK-25840][tests] Add semantics test support for connector test framework
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit 87b9f1bbeb4ee276e4809b1327003b18028913ce
Author: Hang Ruan <ru...@hotmail.com>
AuthorDate: Thu Jan 27 16:57:27 2022 +0800
[FLINK-25840][tests] Add semantics test support for connector test framework
This closes #18547
---
.../apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java | 6 ++++++
.../flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java | 6 ++++++
2 files changed, 12 insertions(+)
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
index 8641f50..7d22e80 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceOrderedE2ECase.java
@@ -23,7 +23,9 @@ import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
import org.apache.flink.connector.testframe.junit.annotations.TestContext;
import org.apache.flink.connector.testframe.junit.annotations.TestEnv;
import org.apache.flink.connector.testframe.junit.annotations.TestExternalSystem;
+import org.apache.flink.connector.testframe.junit.annotations.TestSemantics;
import org.apache.flink.connector.testframe.testsuites.SourceTestSuiteBase;
+import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.tests.util.pulsar.cases.ExclusiveSubscriptionContext;
import org.apache.flink.tests.util.pulsar.cases.FailoverSubscriptionContext;
import org.apache.flink.tests.util.pulsar.common.FlinkContainerWithPulsarEnvironment;
@@ -36,6 +38,10 @@ import static org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntime.
*/
public class PulsarSourceOrderedE2ECase extends SourceTestSuiteBase<String> {
+ // Defines the Semantic.
+ @TestSemantics
+ CheckpointingMode[] semantics = new CheckpointingMode[] {CheckpointingMode.EXACTLY_ONCE};
+
// Defines TestEnvironment.
@TestEnv
FlinkContainerWithPulsarEnvironment flink = new FlinkContainerWithPulsarEnvironment(1, 6);
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
index 4351147..d14d8f9 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceUnorderedE2ECase.java
@@ -23,6 +23,8 @@ import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
import org.apache.flink.connector.testframe.junit.annotations.TestContext;
import org.apache.flink.connector.testframe.junit.annotations.TestEnv;
import org.apache.flink.connector.testframe.junit.annotations.TestExternalSystem;
+import org.apache.flink.connector.testframe.junit.annotations.TestSemantics;
+import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.tests.util.pulsar.cases.KeySharedSubscriptionContext;
import org.apache.flink.tests.util.pulsar.cases.SharedSubscriptionContext;
import org.apache.flink.tests.util.pulsar.common.FlinkContainerWithPulsarEnvironment;
@@ -36,6 +38,10 @@ import static org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntime.
*/
public class PulsarSourceUnorderedE2ECase extends UnorderedSourceTestSuiteBase<String> {
+ // Defines the Semantic.
+ @TestSemantics
+ CheckpointingMode[] semantics = new CheckpointingMode[] {CheckpointingMode.EXACTLY_ONCE};
+
// Defines TestEnvironment.
@TestEnv
FlinkContainerWithPulsarEnvironment flink = new FlinkContainerWithPulsarEnvironment(1, 8);
[flink-connector-pulsar] 18/31: [FLINK-27199][Connector/Pulsar] Bump the pulsar-client-all to latest 2.10.0.
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
commit 65ae7f7595b0d372ebf4777db3775b2671ace83e
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Tue Jul 5 02:48:18 2022 +0800
[FLINK-27199][Connector/Pulsar] Bump the pulsar-client-all to latest 2.10.0.
---
flink-connector-pulsar-e2e-tests/pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/flink-connector-pulsar-e2e-tests/pom.xml b/flink-connector-pulsar-e2e-tests/pom.xml
index 13bf409..b9bc5d4 100644
--- a/flink-connector-pulsar-e2e-tests/pom.xml
+++ b/flink-connector-pulsar-e2e-tests/pom.xml
@@ -31,7 +31,7 @@ under the License.
<name>Flink : E2E Tests : Pulsar</name>
<properties>
- <pulsar.version>2.9.1</pulsar.version>
+ <pulsar.version>2.10.0</pulsar.version>
<bouncycastle.version>1.69</bouncycastle.version>
</properties>