You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ti...@apache.org on 2023/07/18 07:33:59 UTC
[flink-connector-pulsar] branch main updated: [FLINK-32612] Upgrade Pulsar version to 3.0.0 (#54)
This is an automated email from the ASF dual-hosted git repository.
tison pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
The following commit(s) were added to refs/heads/main by this push:
new 279a678 [FLINK-32612] Upgrade Pulsar version to 3.0.0 (#54)
279a678 is described below
commit 279a678dd4f39e1e268d911dc4096d23ab15bc73
Author: tison <wa...@gmail.com>
AuthorDate: Tue Jul 18 15:33:53 2023 +0800
[FLINK-32612] Upgrade Pulsar version to 3.0.0 (#54)
Signed-off-by: tison <wa...@gmail.com>
Co-authored-by: Yufan Sheng <sy...@gmail.com>
Co-authored-by: Yufan Sheng <yu...@streamnative.io>
---
.../flink/tests/util/pulsar/PulsarSinkE2ECase.java | 20 +++++++++-----
.../tests/util/pulsar/PulsarSourceE2ECase.java | 25 ++++++++++++-----
...arEnvironment.java => FlinkContainerUtils.java} | 31 +++++++++++-----------
.../common/PulsarContainerTestEnvironment.java | 3 ++-
.../cursor/stop/MessageIdStopCursor.java | 3 +--
.../pulsar/sink/writer/PulsarWriterTest.java | 21 +++++++++++++++
.../pulsar/testutils/PulsarTestCommonUtils.java | 11 ++++++--
.../pulsar/testutils/PulsarTestContext.java | 12 ++++++---
.../runtime/container/PulsarContainerRuntime.java | 2 +-
flink-sql-connector-pulsar/pom.xml | 5 ++++
.../src/main/resources/META-INF/NOTICE | 7 ++---
pom.xml | 5 ++--
12 files changed, 101 insertions(+), 44 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 e2f7976..7446760 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
@@ -18,15 +18,17 @@
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.pulsar.testutils.sink.cases.SingleTopicProducingContext;
+import org.apache.flink.connector.testframe.container.FlinkContainerTestEnvironment;
+import org.apache.flink.connector.testframe.external.ExternalContextFactory;
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.FlinkContainerUtils;
import org.apache.flink.tests.util.pulsar.common.PulsarContainerTestEnvironment;
import org.junit.jupiter.api.Tag;
@@ -45,14 +47,18 @@ public class PulsarSinkE2ECase extends SinkTestSuiteBase<String> {
// Defines TestEnvironment
@TestEnv
- FlinkContainerWithPulsarEnvironment flink = new FlinkContainerWithPulsarEnvironment(1, 6);
+ FlinkContainerTestEnvironment flink =
+ new FlinkContainerTestEnvironment(FlinkContainerUtils.flinkConfiguration(), 1, 6);
// Defines ConnectorExternalSystem.
- @TestExternalSystem
- PulsarContainerTestEnvironment pulsar = new PulsarContainerTestEnvironment(flink);
+ @TestExternalSystem PulsarTestEnvironment pulsar = new PulsarContainerTestEnvironment(flink);
// Defines a set of external context Factories for different test cases.
@TestContext
- PulsarTestContextFactory<String, SingleTopicProducingContext> sinkContext =
- new PulsarTestContextFactory<>(pulsar, SingleTopicProducingContext::new);
+ ExternalContextFactory<SingleTopicProducingContext> sinkContext =
+ ignore -> {
+ final SingleTopicProducingContext context = new SingleTopicProducingContext(pulsar);
+ context.addConnectorJarPaths(FlinkContainerUtils.connectorJarPaths());
+ return context;
+ };
}
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceE2ECase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceE2ECase.java
index ac2388a..4f4cadc 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceE2ECase.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceE2ECase.java
@@ -18,16 +18,17 @@
package org.apache.flink.tests.util.pulsar;
-import org.apache.flink.connector.pulsar.testutils.PulsarTestContextFactory;
import org.apache.flink.connector.pulsar.testutils.source.cases.MultipleTopicsConsumingContext;
import org.apache.flink.connector.pulsar.testutils.source.cases.PartialKeysConsumingContext;
+import org.apache.flink.connector.testframe.container.FlinkContainerTestEnvironment;
+import org.apache.flink.connector.testframe.external.ExternalContextFactory;
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.common.FlinkContainerWithPulsarEnvironment;
+import org.apache.flink.tests.util.pulsar.common.FlinkContainerUtils;
import org.apache.flink.tests.util.pulsar.common.PulsarContainerTestEnvironment;
import org.junit.jupiter.api.Tag;
@@ -44,7 +45,8 @@ public class PulsarSourceE2ECase extends SourceTestSuiteBase<String> {
// Defines TestEnvironment.
@TestEnv
- FlinkContainerWithPulsarEnvironment flink = new FlinkContainerWithPulsarEnvironment(1, 6);
+ FlinkContainerTestEnvironment flink =
+ new FlinkContainerTestEnvironment(FlinkContainerUtils.flinkConfiguration(), 1, 6);
// Defines ConnectorExternalSystem.
@TestExternalSystem
@@ -52,10 +54,19 @@ public class PulsarSourceE2ECase extends SourceTestSuiteBase<String> {
// Defines a set of external context Factories for different test cases.
@TestContext
- PulsarTestContextFactory<String, MultipleTopicsConsumingContext> multipleTopic =
- new PulsarTestContextFactory<>(pulsar, MultipleTopicsConsumingContext::new);
+ ExternalContextFactory<MultipleTopicsConsumingContext> multipleTopic =
+ ignore -> {
+ final MultipleTopicsConsumingContext context =
+ new MultipleTopicsConsumingContext(pulsar);
+ context.addConnectorJarPaths(FlinkContainerUtils.connectorJarPaths());
+ return context;
+ };
@TestContext
- PulsarTestContextFactory<String, PartialKeysConsumingContext> partialKeys =
- new PulsarTestContextFactory<>(pulsar, PartialKeysConsumingContext::new);
+ ExternalContextFactory<PartialKeysConsumingContext> partialKeys =
+ ignore -> {
+ final PartialKeysConsumingContext context = new PartialKeysConsumingContext(pulsar);
+ context.addConnectorJarPaths(FlinkContainerUtils.connectorJarPaths());
+ return context;
+ };
}
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/FlinkContainerUtils.java
similarity index 70%
rename from flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerWithPulsarEnvironment.java
rename to flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerUtils.java
index ab018f8..43e951f 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/FlinkContainerUtils.java
@@ -22,31 +22,32 @@ 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 static org.apache.flink.connector.pulsar.testutils.PulsarTestCommonUtils.resourcePath;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.List;
-/** A Flink Container which would bundles pulsar connector in its classpath. */
-public class FlinkContainerWithPulsarEnvironment extends FlinkContainerTestEnvironment {
+import static org.apache.flink.connector.pulsar.testutils.PulsarTestCommonUtils.resourcePath;
- public FlinkContainerWithPulsarEnvironment(int numTaskManagers, int numSlotsPerTaskManager) {
- super(
- flinkConfiguration(),
- numTaskManagers,
- numSlotsPerTaskManager,
- resourcePath("pulsar-connector.jar"),
- resourcePath("flink-connector-testing.jar"));
- }
+/** Shared utilities for building Flink containers. */
+public class FlinkContainerUtils {
- private static Configuration flinkConfiguration() {
+ public static Configuration flinkConfiguration() {
Configuration configuration = new Configuration();
// 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));
+ configuration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(2560));
+ configuration.set(JobManagerOptions.JVM_METASPACE, MemorySize.ofMebiBytes(1024));
return configuration;
}
+
+ public static List<URL> connectorJarPaths() {
+ List<URL> urls = new ArrayList<>();
+ urls.add(resourcePath("pulsar-connector.jar"));
+ urls.add(resourcePath("flink-connector-testing.jar"));
+ return urls;
+ }
}
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
index 654347b..059f477 100644
--- 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
@@ -19,13 +19,14 @@
package org.apache.flink.tests.util.pulsar.common;
import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
+import org.apache.flink.connector.testframe.container.FlinkContainerTestEnvironment;
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) {
+ public PulsarContainerTestEnvironment(FlinkContainerTestEnvironment flinkEnvironment) {
super(container(flinkEnvironment.getFlinkContainers().getJobManager()));
}
}
diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/stop/MessageIdStopCursor.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/stop/MessageIdStopCursor.java
index ada2fd3..668b8ef 100644
--- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/stop/MessageIdStopCursor.java
+++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/stop/MessageIdStopCursor.java
@@ -26,7 +26,6 @@ import org.apache.pulsar.client.api.MessageId;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.pulsar.client.api.MessageId.earliest;
import static org.apache.pulsar.client.api.MessageId.latest;
-import static org.apache.pulsar.client.impl.MessageIdImpl.convertToMessageIdImpl;
/**
* Stop consuming message at a given message id. We use the {@link MessageId#compareTo(Object)} for
@@ -43,8 +42,8 @@ public class MessageIdStopCursor implements StopCursor {
checkArgument(!earliest.equals(messageId), "MessageId.earliest is not supported.");
checkArgument(!latest.equals(messageId), "Use LatestMessageStopCursor instead.");
- this.messageId = convertToMessageIdImpl(messageId);
this.inclusive = inclusive;
+ this.messageId = messageId;
}
@Override
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriterTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriterTest.java
index eba7b18..01914cc 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriterTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriterTest.java
@@ -18,9 +18,11 @@
package org.apache.flink.connector.pulsar.sink.writer;
+import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.operators.MailboxExecutor;
import org.apache.flink.api.common.operators.ProcessingTimeService;
import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.connector.sink2.Sink.InitContext;
import org.apache.flink.api.connector.sink2.SinkWriter;
import org.apache.flink.configuration.Configuration;
@@ -205,6 +207,25 @@ class PulsarWriterTest extends PulsarTestSuiteBase {
return 0;
}
+ // The following three methods are for compatibility with
+ // https://github.com/apache/flink/commit/4f5b2fb5736f5a1c098a7dc1d448a879f36f801b
+ // . Removed the commented out `@Override` when we move to 1.18.
+
+ // @Override
+ public boolean isObjectReuseEnabled() {
+ return false;
+ }
+
+ // @Override
+ public <IN> TypeSerializer<IN> createInputSerializer() {
+ return null;
+ }
+
+ // @Override
+ public JobID getJobId() {
+ return null;
+ }
+
@Override
public SinkWriterMetricGroup metricGroup() {
return metricGroup;
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestCommonUtils.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestCommonUtils.java
index 487fbdc..970dd44 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestCommonUtils.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestCommonUtils.java
@@ -25,9 +25,12 @@ import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.test.resources.ResourceTestUtils;
+import org.apache.flink.util.FlinkRuntimeException;
import org.apache.pulsar.client.api.MessageId;
+import java.net.MalformedURLException;
+import java.net.URL;
import java.util.ArrayList;
import java.util.List;
@@ -46,8 +49,12 @@ public class PulsarTestCommonUtils {
}
}
- public static String resourcePath(String jarName) {
- return ResourceTestUtils.getResource(jarName).toAbsolutePath().toString();
+ public static URL resourcePath(String jarName) {
+ try {
+ return ResourceTestUtils.getResource(jarName).toAbsolutePath().toUri().toURL();
+ } catch (MalformedURLException e) {
+ throw new FlinkRuntimeException("Couldn't find jar: " + jarName);
+ }
}
/** creates a fullRange() partitionSplit. */
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContext.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContext.java
index d283df8..531aa20 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContext.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContext.java
@@ -24,7 +24,7 @@ import org.apache.flink.connector.testframe.external.ExternalContext;
import org.apache.pulsar.client.api.Schema;
import java.net.URL;
-import java.util.Collections;
+import java.util.ArrayList;
import java.util.List;
/**
@@ -37,6 +37,8 @@ public abstract class PulsarTestContext<T> implements ExternalContext {
// The schema used for consuming and producing messages between Pulsar and tests.
protected final Schema<T> schema;
+ private final List<URL> connectorJarPaths = new ArrayList<>();
+
protected PulsarTestContext(PulsarTestEnvironment environment, Schema<T> schema) {
this.operator = environment.operator();
this.schema = schema;
@@ -52,9 +54,11 @@ public abstract class PulsarTestContext<T> implements ExternalContext {
@Override
public List<URL> getConnectorJarPaths() {
- // We don't need any test jars' definition.
- // They are provided in docker-related environments.
- return Collections.emptyList();
+ return connectorJarPaths;
+ }
+
+ public void addConnectorJarPaths(List<URL> urls) {
+ this.connectorJarPaths.addAll(urls);
}
/**
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/container/PulsarContainerRuntime.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/container/PulsarContainerRuntime.java
index 20c4ae8..097927e 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/container/PulsarContainerRuntime.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/container/PulsarContainerRuntime.java
@@ -54,7 +54,7 @@ public class PulsarContainerRuntime implements PulsarRuntime {
private static final String PULSAR_ADMIN_URL =
String.format("http://%s:%d", PULSAR_INTERNAL_HOSTNAME, BROKER_HTTP_PORT);
- private static final String CURRENT_VERSION = "2.11.0";
+ private static final String CURRENT_VERSION = "3.0.0";
private final PulsarContainer container;
private final AtomicBoolean started;
diff --git a/flink-sql-connector-pulsar/pom.xml b/flink-sql-connector-pulsar/pom.xml
index 0c2b2ce..bdc3809 100644
--- a/flink-sql-connector-pulsar/pom.xml
+++ b/flink-sql-connector-pulsar/pom.xml
@@ -44,6 +44,10 @@ under the License.
<artifactId>flink-connector-pulsar</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-annotations</artifactId>
+ </dependency>
</dependencies>
<build>
@@ -61,6 +65,7 @@ under the License.
<configuration>
<artifactSet>
<includes>
+ <include>com.fasterxml.jackson.core:jackson-annotations</include>
<include>org.apache.flink:flink-connector-base</include>
<include>org.apache.flink:flink-connector-pulsar</include>
<include>org.apache.pulsar:pulsar-client-admin-api</include>
diff --git a/flink-sql-connector-pulsar/src/main/resources/META-INF/NOTICE b/flink-sql-connector-pulsar/src/main/resources/META-INF/NOTICE
index 9aa8f42..08199ce 100644
--- a/flink-sql-connector-pulsar/src/main/resources/META-INF/NOTICE
+++ b/flink-sql-connector-pulsar/src/main/resources/META-INF/NOTICE
@@ -6,9 +6,10 @@ The Apache Software Foundation (http://www.apache.org/).
This project bundles the following dependencies under the Apache Software License 2.0 (http://www.apache.org/licenses/LICENSE-2.0.txt)
-- org.apache.pulsar:pulsar-client-admin-api:2.11.0
-- org.apache.pulsar:pulsar-client-all:2.11.0
-- org.apache.pulsar:pulsar-client-api:2.11.0
+- com.fasterxml.jackson.core:jackson-annotations:2.13.4
+- org.apache.pulsar:pulsar-client-admin-api:3.0.0
+- org.apache.pulsar:pulsar-client-all:3.0.0
+- org.apache.pulsar:pulsar-client-api:3.0.0
This project bundles the following dependencies under the Bouncy Castle license.
See bundled license files for details.
diff --git a/pom.xml b/pom.xml
index 75c3083..b67e1e3 100644
--- a/pom.xml
+++ b/pom.xml
@@ -52,7 +52,8 @@ under the License.
<properties>
<flink.version>1.17.0</flink.version>
- <pulsar.version>2.11.0</pulsar.version>
+ <flink-ci-tools.version>1.18-SNAPSHOT</flink-ci-tools.version>
+ <pulsar.version>3.0.0</pulsar.version>
<bouncycastle.version>1.69</bouncycastle.version>
<jsr305.version>1.3.9</jsr305.version>
@@ -457,7 +458,7 @@ under the License.
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-ci-tools</artifactId>
- <version>${flink.version}</version>
+ <version>${flink-ci-tools.version}</version>
</dependency>
</dependencies>
</plugin>