You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ar...@apache.org on 2021/08/26 11:25:15 UTC
[flink] branch master updated (a8ea3c3 -> 3ef9323)
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from a8ea3c3 [FLINK-23862][runtime] Replace usages of AbstractInvokable with interface
new d476c5f [FLINK-23877][connector/pulsar] Provide an embedded pulsar server for testing.
new 2a55242 [FLINK-23877][connector/pulsar] Drop the ConfigurationDataCustomizer, use ClientBuilder and PulsarAdminBuilder for creating instance.
new 069e437 [FLINK-23877][connector/pulsar] Change pulsar config option's type, drop PulsarJsonUtils, make it strong type support.
new 0fc9008 [FLINK-23877][connector/pulsar] Change the config setter in pulsar source builder. Drop Properties. Remove useless config options.
new 111b7ec [FLINK-23877][connector/pulsar] Since we have use the StartCursor, it's no need to expose the resetIncludeHead option for users.
new 3ef9323 [FLINK-23877][connector/pulsar] Enrich the config options' description, make sure all the options have a detailed description.
The 6 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
flink-connectors/flink-connector-pulsar/pom.xml | 67 ++++-
.../common/config/ConfigurationDataCustomizer.java | 58 ----
.../pulsar/common/config/PulsarConfigUtils.java | 291 ++++++++++-----------
.../pulsar/common/config/PulsarOptions.java | 283 +++++++++++++++-----
.../pulsar/common/utils/PulsarJsonUtils.java | 113 --------
.../connector/pulsar/source/PulsarSource.java | 28 +-
.../pulsar/source/PulsarSourceBuilder.java | 205 ++++-----------
.../pulsar/source/PulsarSourceOptions.java | 230 ++++++++++------
.../source/config/PulsarSourceConfigUtils.java | 211 ++++++---------
.../source/enumerator/PulsarSourceEnumerator.java | 5 +-
.../source/reader/PulsarSourceReaderFactory.java | 14 +-
.../split/PulsarOrderedPartitionSplitReader.java | 12 +-
.../split/PulsarPartitionSplitReaderBase.java | 41 +--
.../split/PulsarUnorderedPartitionSplitReader.java | 12 +-
.../pulsar/source/PulsarSourceITCase.java | 15 +-
.../enumerator/PulsarSourceEnumeratorTest.java | 23 +-
.../subscriber/PulsarSubscriberTest.java | 24 +-
.../pulsar/testutils/PulsarContainerContext.java | 126 ---------
.../testutils/PulsarContainerEnvironment.java | 132 ----------
.../testutils/PulsarPartitionDataWriter.java | 13 +-
.../pulsar/testutils/PulsarTestContext.java | 65 +++++
...tFactory.java => PulsarTestContextFactory.java} | 18 +-
.../pulsar/testutils/PulsarTestEnvironment.java | 116 ++++++++
.../pulsar/testutils/PulsarTestSuiteBase.java | 77 +-----
.../cases/MultipleTopicConsumingContext.java | 17 +-
.../cases/SingleTopicConsumingContext.java | 19 +-
.../pulsar/testutils/runtime/PulsarRuntime.java | 50 ++++
.../PulsarRuntimeOperator.java} | 10 +-
.../testutils/runtime/PulsarRuntimeProvider.java | 37 +++
.../runtime/container/PulsarContainerProvider.java | 95 +++++++
.../runtime/mock/BlankBrokerInterceptor.java | 61 +++++
.../runtime/mock/MockBookKeeperClientFactory.java | 74 ++++++
.../testutils/runtime/mock/MockPulsarService.java | 87 ++++++
.../runtime/mock/MockZooKeeperClientFactory.java | 73 ++++++
.../runtime/mock/NonClosableMockBookKeeper.java | 55 ++++
.../testutils/runtime/mock/PortBindingUtils.java | 67 +++++
.../testutils/runtime/mock/PulsarMockProvider.java | 140 ++++++++++
.../mock/SameThreadOrderedSafeExecutor.java | 56 ++++
tools/maven/suppressions.xml | 6 +-
39 files changed, 1783 insertions(+), 1243 deletions(-)
delete mode 100644 flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/ConfigurationDataCustomizer.java
delete mode 100644 flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/utils/PulsarJsonUtils.java
delete mode 100644 flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarContainerContext.java
delete mode 100644 flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarContainerEnvironment.java
create mode 100644 flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContext.java
rename flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/{PulsarContainerContextFactory.java => PulsarTestContextFactory.java} (68%)
create mode 100644 flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestEnvironment.java
create mode 100644 flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntime.java
rename flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/{PulsarContainerOperator.java => runtime/PulsarRuntimeOperator.java} (96%)
create mode 100644 flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntimeProvider.java
create mode 100644 flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/container/PulsarContainerProvider.java
create mode 100644 flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/BlankBrokerInterceptor.java
create mode 100644 flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/MockBookKeeperClientFactory.java
create mode 100644 flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/MockPulsarService.java
create mode 100644 flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/MockZooKeeperClientFactory.java
create mode 100644 flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/NonClosableMockBookKeeper.java
create mode 100644 flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/PortBindingUtils.java
create mode 100644 flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/PulsarMockProvider.java
create mode 100644 flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/SameThreadOrderedSafeExecutor.java
[flink] 01/06: [FLINK-23877][connector/pulsar] Provide an embedded
pulsar server for testing.
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d476c5facf7baac5b3c70295599d4fc41c52eedd
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Tue Aug 24 06:19:02 2021 +0800
[FLINK-23877][connector/pulsar] Provide an embedded pulsar server for testing.
---
flink-connectors/flink-connector-pulsar/pom.xml | 67 ++++++++--
.../pulsar/source/PulsarSourceITCase.java | 15 +--
.../enumerator/PulsarSourceEnumeratorTest.java | 12 +-
.../subscriber/PulsarSubscriberTest.java | 24 ++--
.../pulsar/testutils/PulsarContainerContext.java | 126 -------------------
.../testutils/PulsarContainerEnvironment.java | 132 -------------------
.../testutils/PulsarPartitionDataWriter.java | 13 +-
.../pulsar/testutils/PulsarTestContext.java | 65 ++++++++++
...tFactory.java => PulsarTestContextFactory.java} | 18 +--
.../pulsar/testutils/PulsarTestEnvironment.java | 116 +++++++++++++++++
.../pulsar/testutils/PulsarTestSuiteBase.java | 77 +++---------
.../cases/MultipleTopicConsumingContext.java | 17 +--
.../cases/SingleTopicConsumingContext.java | 19 +--
.../pulsar/testutils/runtime/PulsarRuntime.java | 50 ++++++++
.../PulsarRuntimeOperator.java} | 10 +-
.../testutils/runtime/PulsarRuntimeProvider.java | 37 ++++++
.../runtime/container/PulsarContainerProvider.java | 95 ++++++++++++++
.../runtime/mock/BlankBrokerInterceptor.java | 61 +++++++++
.../runtime/mock/MockBookKeeperClientFactory.java | 74 +++++++++++
.../testutils/runtime/mock/MockPulsarService.java | 87 +++++++++++++
.../runtime/mock/MockZooKeeperClientFactory.java | 73 +++++++++++
.../runtime/mock/NonClosableMockBookKeeper.java | 55 ++++++++
.../testutils/runtime/mock/PortBindingUtils.java | 67 ++++++++++
.../testutils/runtime/mock/PulsarMockProvider.java | 140 +++++++++++++++++++++
.../mock/SameThreadOrderedSafeExecutor.java | 56 +++++++++
tools/maven/suppressions.xml | 6 +-
26 files changed, 1119 insertions(+), 393 deletions(-)
diff --git a/flink-connectors/flink-connector-pulsar/pom.xml b/flink-connectors/flink-connector-pulsar/pom.xml
index bc9b379..fe13f3a 100644
--- a/flink-connectors/flink-connector-pulsar/pom.xml
+++ b/flink-connectors/flink-connector-pulsar/pom.xml
@@ -41,6 +41,8 @@ under the License.
<!-- Test Libraries -->
<protobuf-maven-plugin.version>0.6.1</protobuf-maven-plugin.version>
<assertj-core.version>3.20.2</assertj-core.version>
+ <commons-lang3.version>3.11</commons-lang3.version>
+ <grpc.version>1.33.0</grpc.version>
</properties>
<dependencies>
@@ -62,14 +64,6 @@ under the License.
<scope>provided</scope>
</dependency>
- <!-- Add Pulsar 2.x as a dependency -->
-
- <dependency>
- <groupId>org.apache.pulsar</groupId>
- <artifactId>pulsar-client-all</artifactId>
- <version>${pulsar.version}</version>
- </dependency>
-
<!-- Protobuf & Protobuf Native Schema support. Add it to your pom if you need protobuf -->
<dependency>
@@ -127,15 +121,70 @@ under the License.
<scope>test</scope>
</dependency>
- <!-- TestContainers for Pulsar environment -->
+ <!-- Pulsar testing environment -->
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>pulsar</artifactId>
<scope>test</scope>
</dependency>
+
+ <!-- Pulsar bundles the latest bookkeeper & zookeeper, -->
+ <!-- we don't override the version here. -->
+ <dependency>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>testmocks</artifactId>
+ <version>${pulsar.version}</version>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>org.testng</groupId>
+ <artifactId>testng</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.powermock</groupId>
+ <artifactId>powermock-module-testng</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar-broker</artifactId>
+ <version>${pulsar.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <!-- Pulsar use a newer commons-lang3 in broker. -->
+ <!-- Bump the version only for testing. -->
+ <dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-lang3</artifactId>
+ <version>${commons-lang3.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- Add Pulsar 2.x as a dependency. -->
+ <!-- Move this to button for avoiding class conflicts with pulsar-broker. -->
+
+ <dependency>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar-client-all</artifactId>
+ <version>${pulsar.version}</version>
+ </dependency>
</dependencies>
+ <!-- gRPC use version range which don't support by flink ci. -->
+ <dependencyManagement>
+ <dependencies>
+ <dependency>
+ <groupId>io.grpc</groupId>
+ <artifactId>grpc-bom</artifactId>
+ <version>${grpc.version}</version>
+ <type>pom</type>
+ <scope>import</scope>
+ </dependency>
+ </dependencies>
+ </dependencyManagement>
+
<build>
<extensions>
<extension>
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarSourceITCase.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarSourceITCase.java
index c94b208..89457dc 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarSourceITCase.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarSourceITCase.java
@@ -18,10 +18,11 @@
package org.apache.flink.connector.pulsar.source;
-import org.apache.flink.connector.pulsar.testutils.PulsarContainerContextFactory;
-import org.apache.flink.connector.pulsar.testutils.PulsarContainerEnvironment;
+import org.apache.flink.connector.pulsar.testutils.PulsarTestContextFactory;
+import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
import org.apache.flink.connector.pulsar.testutils.cases.MultipleTopicConsumingContext;
import org.apache.flink.connector.pulsar.testutils.cases.SingleTopicConsumingContext;
+import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntime;
import org.apache.flink.connectors.test.common.environment.MiniClusterTestEnvironment;
import org.apache.flink.connectors.test.common.junit.annotations.ExternalContextFactory;
import org.apache.flink.connectors.test.common.junit.annotations.ExternalSystem;
@@ -36,15 +37,15 @@ class PulsarSourceITCase extends SourceTestSuiteBase<String> {
@TestEnv MiniClusterTestEnvironment flink = new MiniClusterTestEnvironment();
// Defines pulsar running environment
- @ExternalSystem PulsarContainerEnvironment pulsar = new PulsarContainerEnvironment();
+ @ExternalSystem PulsarTestEnvironment pulsar = new PulsarTestEnvironment(PulsarRuntime.MOCK);
// Defines a external context Factories,
// so test cases will be invoked using this external contexts.
@ExternalContextFactory
- PulsarContainerContextFactory<String, SingleTopicConsumingContext> singleTopic =
- new PulsarContainerContextFactory<>(pulsar, SingleTopicConsumingContext::new);
+ PulsarTestContextFactory<String, SingleTopicConsumingContext> singleTopic =
+ new PulsarTestContextFactory<>(pulsar, SingleTopicConsumingContext::new);
@ExternalContextFactory
- PulsarContainerContextFactory<String, MultipleTopicConsumingContext> multipleTopic =
- new PulsarContainerContextFactory<>(pulsar, MultipleTopicConsumingContext::new);
+ PulsarTestContextFactory<String, MultipleTopicConsumingContext> multipleTopic =
+ new PulsarTestContextFactory<>(pulsar, MultipleTopicConsumingContext::new);
}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/PulsarSourceEnumeratorTest.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/PulsarSourceEnumeratorTest.java
index 320479b..9622ee3 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/PulsarSourceEnumeratorTest.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/PulsarSourceEnumeratorTest.java
@@ -72,14 +72,14 @@ class PulsarSourceEnumeratorTest extends PulsarTestSuiteBase {
// method could be non-static.
@BeforeAll
void beforeAll() {
- setupTopic(TOPIC1);
- setupTopic(TOPIC2);
+ operator().setupTopic(TOPIC1);
+ operator().setupTopic(TOPIC2);
}
@AfterAll
void afterAll() {
- deleteTopic(TOPIC1, true);
- deleteTopic(TOPIC2, true);
+ operator().deleteTopic(TOPIC1, true);
+ operator().deleteTopic(TOPIC2, true);
}
@Test
@@ -133,8 +133,8 @@ class PulsarSourceEnumeratorTest extends PulsarTestSuiteBase {
topics.add(DYNAMIC_TOPIC_NAME);
}
Configuration configuration = new Configuration();
- configuration.set(PULSAR_ADMIN_URL, adminUrl());
- configuration.set(PULSAR_SERVICE_URL, serviceUrl());
+ configuration.set(PULSAR_ADMIN_URL, operator().adminUrl());
+ configuration.set(PULSAR_SERVICE_URL, operator().serviceUrl());
configuration.set(PULSAR_SUBSCRIPTION_TYPE, SubscriptionType.Failover);
configuration.set(PULSAR_TOPIC_NAMES, PulsarJsonUtils.toString(topics));
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/PulsarSubscriberTest.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/PulsarSubscriberTest.java
index 2056afb..bdfbb42 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/PulsarSubscriberTest.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/PulsarSubscriberTest.java
@@ -48,13 +48,13 @@ class PulsarSubscriberTest extends PulsarTestSuiteBase {
@Test
void topicListSubscriber() {
- createTopic(TOPIC1, NUM_PARTITIONS_PER_TOPIC);
- createTopic(TOPIC2, NUM_PARTITIONS_PER_TOPIC);
+ operator().createTopic(TOPIC1, NUM_PARTITIONS_PER_TOPIC);
+ operator().createTopic(TOPIC2, NUM_PARTITIONS_PER_TOPIC);
PulsarSubscriber subscriber = getTopicListSubscriber(Arrays.asList(TOPIC1, TOPIC2));
Set<TopicPartition> topicPartitions =
subscriber.getSubscribedTopicPartitions(
- admin(), new FullRangeGenerator(), NUM_PARALLELISM);
+ operator().admin(), new FullRangeGenerator(), NUM_PARALLELISM);
Set<TopicPartition> expectedPartitions = new HashSet<>();
for (int i = 0; i < NUM_PARTITIONS_PER_TOPIC; i++) {
@@ -64,15 +64,15 @@ class PulsarSubscriberTest extends PulsarTestSuiteBase {
assertEquals(expectedPartitions, topicPartitions);
- deleteTopic(TOPIC1, true);
- deleteTopic(TOPIC2, true);
+ operator().deleteTopic(TOPIC1, true);
+ operator().deleteTopic(TOPIC2, true);
}
@Test
void topicPatternSubscriber() {
- createTopic(TOPIC1, NUM_PARTITIONS_PER_TOPIC);
- createTopic(TOPIC2, NUM_PARTITIONS_PER_TOPIC);
- createTopic(TOPIC3, NUM_PARTITIONS_PER_TOPIC);
+ operator().createTopic(TOPIC1, NUM_PARTITIONS_PER_TOPIC);
+ operator().createTopic(TOPIC2, NUM_PARTITIONS_PER_TOPIC);
+ operator().createTopic(TOPIC3, NUM_PARTITIONS_PER_TOPIC);
PulsarSubscriber subscriber =
getTopicPatternSubscriber(
@@ -80,7 +80,7 @@ class PulsarSubscriberTest extends PulsarTestSuiteBase {
Set<TopicPartition> topicPartitions =
subscriber.getSubscribedTopicPartitions(
- admin(), new FullRangeGenerator(), NUM_PARALLELISM);
+ operator().admin(), new FullRangeGenerator(), NUM_PARALLELISM);
Set<TopicPartition> expectedPartitions = new HashSet<>();
@@ -91,8 +91,8 @@ class PulsarSubscriberTest extends PulsarTestSuiteBase {
assertEquals(expectedPartitions, topicPartitions);
- deleteTopic(TOPIC1, true);
- deleteTopic(TOPIC2, true);
- deleteTopic(TOPIC3, true);
+ operator().deleteTopic(TOPIC1, true);
+ operator().deleteTopic(TOPIC2, true);
+ operator().deleteTopic(TOPIC3, true);
}
}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarContainerContext.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarContainerContext.java
deleted file mode 100644
index ddef4ba..0000000
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarContainerContext.java
+++ /dev/null
@@ -1,126 +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.connector.pulsar.testutils;
-
-import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
-import org.apache.flink.connectors.test.common.external.ExternalContext;
-
-import org.apache.pulsar.client.admin.PulsarAdmin;
-import org.apache.pulsar.client.api.MessageId;
-import org.apache.pulsar.client.api.PulsarClient;
-import org.apache.pulsar.client.api.Schema;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-import java.util.Random;
-import java.util.function.Supplier;
-
-import static org.apache.commons.lang3.RandomStringUtils.randomAlphanumeric;
-
-/** Common test context for pulsar container based test. */
-public abstract class PulsarContainerContext<T> implements ExternalContext<T> {
-
- private static final int NUM_RECORDS_UPPER_BOUND = 500;
- private static final int NUM_RECORDS_LOWER_BOUND = 100;
-
- private final String displayName;
- protected final PulsarContainerOperator operator;
-
- protected PulsarContainerContext(String displayName, PulsarContainerEnvironment environment) {
- this.displayName = displayName;
- this.operator = environment.operator();
- }
-
- // Helper methods for generating data.
-
- protected List<String> generateStringTestData(long seed) {
- Random random = new Random(seed);
- int recordNum =
- random.nextInt(NUM_RECORDS_UPPER_BOUND - NUM_RECORDS_LOWER_BOUND)
- + NUM_RECORDS_LOWER_BOUND;
- List<String> records = new ArrayList<>(recordNum);
-
- for (int i = 0; i < recordNum; i++) {
- int stringLength = random.nextInt(50) + 1;
- records.add(randomAlphanumeric(stringLength));
- }
-
- return records;
- }
-
- // Helper methods for operating pulsar.
-
- protected void setupTopic(String topic) {
- operator.setupTopic(topic);
- }
-
- protected void setupTopic(String topic, Schema<T> schema, Supplier<T> supplier) {
- operator.setupTopic(topic, schema, supplier);
- }
-
- protected void createTopic(String topic, int numberOfPartitions) {
- operator.createTopic(topic, numberOfPartitions);
- }
-
- protected void increaseTopicPartitions(String topic, int newPartitionsNum) {
- operator.increaseTopicPartitions(topic, newPartitionsNum);
- }
-
- protected void deleteTopic(String topic, boolean isPartitioned) {
- operator.deleteTopic(topic, isPartitioned);
- }
-
- protected List<TopicPartition> topicInfo(String topic) {
- return operator.topicInfo(topic);
- }
-
- protected List<TopicPartition> topicsInfo(Collection<String> topics) {
- return operator.topicsInfo(topics);
- }
-
- protected MessageId sendMessage(String topic, Schema<T> schema, T message) {
- return operator.sendMessage(topic, schema, message);
- }
-
- protected List<MessageId> sendMessages(String topic, Schema<T> schema, Collection<T> messages) {
- return operator.sendMessages(topic, schema, messages);
- }
-
- protected String serviceUrl() {
- return operator.serviceUrl();
- }
-
- protected String adminUrl() {
- return operator.adminUrl();
- }
-
- protected PulsarClient client() {
- return operator.client();
- }
-
- protected PulsarAdmin admin() {
- return operator.admin();
- }
-
- @Override
- public String toString() {
- return displayName;
- }
-}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarContainerEnvironment.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarContainerEnvironment.java
deleted file mode 100644
index 3e6ee7f..0000000
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarContainerEnvironment.java
+++ /dev/null
@@ -1,132 +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.connector.pulsar.testutils;
-
-import org.apache.flink.connectors.test.common.TestResource;
-import org.apache.flink.connectors.test.common.junit.annotations.ExternalSystem;
-
-import org.junit.jupiter.api.extension.AfterAllCallback;
-import org.junit.jupiter.api.extension.BeforeAllCallback;
-import org.junit.jupiter.api.extension.Extension;
-import org.junit.jupiter.api.extension.ExtensionContext;
-import org.junit.rules.ExternalResource;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.testcontainers.containers.BindMode;
-import org.testcontainers.containers.PulsarContainer;
-import org.testcontainers.containers.output.Slf4jLogConsumer;
-import org.testcontainers.containers.wait.strategy.HttpWaitStrategy;
-import org.testcontainers.utility.DockerImageName;
-
-import java.time.Duration;
-
-import static org.apache.flink.util.DockerImageVersions.PULSAR;
-import static org.apache.flink.util.Preconditions.checkNotNull;
-import static org.testcontainers.containers.PulsarContainer.BROKER_HTTP_PORT;
-
-/**
- * A JUnit 5 {@link Extension} for supporting running of pulsar test container. This class is also a
- * {@link ExternalSystem} for {@code flink-connector-testing} tools.
- *
- * <p>Some old flink tests are based on JUint 4, this class is also support it. The follow code
- * snippet shows how to use this class in JUnit 4.
- *
- * <pre>{@code
- * @ClassRule
- * public static PulsarContainerEnvironment environment = new PulsarContainerEnvironment();
- * }</pre>
- *
- * <p>If you want to use this class in JUnit 5, just simply extends {@link PulsarTestSuiteBase}, all
- * the helper methods in {@code PulsarContainerOperator} is also exposing there.
- */
-public class PulsarContainerEnvironment extends ExternalResource
- implements BeforeAllCallback, AfterAllCallback, TestResource {
- private static final Logger LOG = LoggerFactory.getLogger(PulsarContainerEnvironment.class);
-
- private PulsarContainer container;
- private PulsarContainerOperator operator;
-
- /** JUnit 4 Rule setup method. */
- @Override
- protected void before() {
- startUp();
- }
-
- /** JUnit 5 Extension setup method. */
- @Override
- public void beforeAll(ExtensionContext context) {
- startUp();
- }
-
- /** flink-connector-testing setup method. */
- @Override
- public void startUp() {
- DockerImageName imageName = DockerImageName.parse(PULSAR);
- this.container = new PulsarContainer(imageName);
-
- // Prepare Pulsar Container.
- container.withClasspathResourceMapping(
- "containers/txnStandalone.conf",
- "/pulsar/conf/standalone.conf",
- BindMode.READ_ONLY);
- container.addExposedPort(2181);
- container.waitingFor(
- new HttpWaitStrategy()
- .forPort(BROKER_HTTP_PORT)
- .forStatusCode(200)
- .forPath("/admin/v2/namespaces/public/default")
- .withStartupTimeout(Duration.ofMinutes(5)));
-
- // Start Pulsar Container.
- container.start();
- container.followOutput(new Slf4jLogConsumer(LOG).withSeparateOutputStreams());
-
- this.operator =
- new PulsarContainerOperator(
- container.getPulsarBrokerUrl(), container.getHttpServiceUrl());
- }
-
- /** JUnit 4 Rule shutdown method. */
- @Override
- protected void after() {
- try {
- tearDown();
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- /** JUnit 5 Extension shutdown method. */
- @Override
- public void afterAll(ExtensionContext context) throws Exception {
- tearDown();
- }
-
- /** flink-connector-testing shutdown method. */
- @Override
- public void tearDown() throws Exception {
- container.stop();
- operator.close();
- }
-
- /** Get a common supported set of method for operating pulsar which is in container. */
- public PulsarContainerOperator operator() {
- return checkNotNull(operator, "You should start the pulsar container first.");
- }
-}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarPartitionDataWriter.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarPartitionDataWriter.java
index a259fdf..ea9c4ab 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarPartitionDataWriter.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarPartitionDataWriter.java
@@ -23,6 +23,7 @@ import org.apache.flink.connectors.test.common.external.SourceSplitDataWriter;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import java.util.Collection;
@@ -35,12 +36,12 @@ public class PulsarPartitionDataWriter implements SourceSplitDataWriter<String>
private final Producer<String> producer;
public PulsarPartitionDataWriter(PulsarClient client, TopicPartition partition) {
- this.producer =
- sneakyClient(
- () ->
- client.newProducer(Schema.STRING)
- .topic(partition.getFullTopicName())
- .create());
+ try {
+ this.producer =
+ client.newProducer(Schema.STRING).topic(partition.getFullTopicName()).create();
+ } catch (PulsarClientException e) {
+ throw new IllegalStateException(e);
+ }
}
@Override
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContext.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContext.java
new file mode 100644
index 0000000..6733439
--- /dev/null
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContext.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.testutils;
+
+import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntimeOperator;
+import org.apache.flink.connectors.test.common.external.ExternalContext;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+
+import static org.apache.commons.lang3.RandomStringUtils.randomAlphanumeric;
+
+/** Common test context for pulsar based test. */
+public abstract class PulsarTestContext<T> implements ExternalContext<T> {
+
+ private static final int NUM_RECORDS_UPPER_BOUND = 500;
+ private static final int NUM_RECORDS_LOWER_BOUND = 100;
+
+ private final String displayName;
+ protected final PulsarRuntimeOperator operator;
+
+ protected PulsarTestContext(String displayName, PulsarTestEnvironment environment) {
+ this.displayName = displayName;
+ this.operator = environment.operator();
+ }
+
+ // Helper methods for generating data.
+
+ protected List<String> generateStringTestData(long seed) {
+ Random random = new Random(seed);
+ int recordNum =
+ random.nextInt(NUM_RECORDS_UPPER_BOUND - NUM_RECORDS_LOWER_BOUND)
+ + NUM_RECORDS_LOWER_BOUND;
+ List<String> records = new ArrayList<>(recordNum);
+
+ for (int i = 0; i < recordNum; i++) {
+ int stringLength = random.nextInt(50) + 1;
+ records.add(randomAlphanumeric(stringLength));
+ }
+
+ return records;
+ }
+
+ @Override
+ public String toString() {
+ return displayName;
+ }
+}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarContainerContextFactory.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContextFactory.java
similarity index 68%
rename from flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarContainerContextFactory.java
rename to flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContextFactory.java
index 1916db5..c0517c6 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarContainerContextFactory.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContextFactory.java
@@ -19,21 +19,21 @@
package org.apache.flink.connector.pulsar.testutils;
import org.apache.flink.connectors.test.common.external.ExternalContext;
-import org.apache.flink.runtime.util.SerializableFunction;
+
+import java.util.function.Function;
/**
- * Factory for creating all the test context that extends {@link PulsarContainerContext}. Test
- * context class should have a constructor with {@link PulsarContainerEnvironment} arg.
+ * Factory for creating all the test context that extends {@link PulsarTestContext}. Test context
+ * class should have a constructor with {@link PulsarTestEnvironment} arg.
*/
-public class PulsarContainerContextFactory<F, T extends PulsarContainerContext<F>>
+public class PulsarTestContextFactory<F, T extends PulsarTestContext<F>>
implements ExternalContext.Factory<F> {
- private final PulsarContainerEnvironment environment;
- private final SerializableFunction<PulsarContainerEnvironment, T> contextFactory;
+ private final PulsarTestEnvironment environment;
+ private final Function<PulsarTestEnvironment, T> contextFactory;
- public PulsarContainerContextFactory(
- PulsarContainerEnvironment environment,
- SerializableFunction<PulsarContainerEnvironment, T> contextFactory) {
+ public PulsarTestContextFactory(
+ PulsarTestEnvironment environment, Function<PulsarTestEnvironment, T> contextFactory) {
this.environment = environment;
this.contextFactory = contextFactory;
}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestEnvironment.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestEnvironment.java
new file mode 100644
index 0000000..7b31c7c
--- /dev/null
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestEnvironment.java
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.testutils;
+
+import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntime;
+import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntimeOperator;
+import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntimeProvider;
+import org.apache.flink.connectors.test.common.TestResource;
+import org.apache.flink.connectors.test.common.junit.annotations.ExternalSystem;
+
+import org.junit.jupiter.api.extension.AfterAllCallback;
+import org.junit.jupiter.api.extension.BeforeAllCallback;
+import org.junit.jupiter.api.extension.Extension;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.junit.rules.TestRule;
+import org.junit.runner.Description;
+import org.junit.runners.model.MultipleFailureException;
+import org.junit.runners.model.Statement;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * A JUnit 5 {@link Extension} for supporting running a pulsar instance before executing tests. This
+ * class is also a {@link ExternalSystem} for {@code flink-connector-testing} tools.
+ *
+ * <p>Some old flink tests are based on JUint 4, this class is also support it. The follow code
+ * snippet shows how to use this class in JUnit 4.
+ *
+ * <pre>{@code
+ * @ClassRule
+ * public static PulsarContainerEnvironment environment = new PulsarContainerEnvironment(MOCK);
+ * }</pre>
+ *
+ * <p>If you want to use this class in JUnit 5, just simply extends {@link PulsarTestSuiteBase}, all
+ * the helper methods in {@code PulsarContainerOperator} is also exposed there.
+ */
+public class PulsarTestEnvironment
+ implements BeforeAllCallback, AfterAllCallback, TestResource, TestRule {
+
+ private final PulsarRuntimeProvider provider;
+
+ public PulsarTestEnvironment(PulsarRuntime runtime) {
+ this.provider = runtime.provider();
+ }
+
+ /** JUnit 4 Rule based test logic. */
+ @Override
+ public Statement apply(Statement base, Description description) {
+ return new Statement() {
+ @Override
+ public void evaluate() throws Throwable {
+ provider.startUp();
+
+ List<Throwable> errors = new ArrayList<>();
+ try {
+ base.evaluate();
+ } catch (Throwable t) {
+ errors.add(t);
+ } finally {
+ try {
+ provider.tearDown();
+ } catch (Throwable t) {
+ errors.add(t);
+ }
+ }
+ MultipleFailureException.assertEmpty(errors);
+ }
+ };
+ }
+
+ /** JUnit 5 Extension setup method. */
+ @Override
+ public void beforeAll(ExtensionContext context) {
+ provider.startUp();
+ }
+
+ /** flink-connector-testing setup method. */
+ @Override
+ public void startUp() {
+ provider.startUp();
+ }
+
+ /** JUnit 5 Extension shutdown method. */
+ @Override
+ public void afterAll(ExtensionContext context) {
+ provider.tearDown();
+ }
+
+ /** flink-connector-testing shutdown method. */
+ @Override
+ public void tearDown() {
+ provider.tearDown();
+ }
+
+ /** Get a common supported set of method for operating pulsar which is in container. */
+ public PulsarRuntimeOperator operator() {
+ return provider.operator();
+ }
+}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestSuiteBase.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestSuiteBase.java
index 65a7888..2321bd4 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestSuiteBase.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestSuiteBase.java
@@ -18,21 +18,14 @@
package org.apache.flink.connector.pulsar.testutils;
-import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
+import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntime;
+import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntimeOperator;
import org.apache.flink.connectors.test.common.junit.extensions.TestLoggerExtension;
-import org.apache.pulsar.client.admin.PulsarAdmin;
-import org.apache.pulsar.client.api.MessageId;
-import org.apache.pulsar.client.api.PulsarClient;
-import org.apache.pulsar.client.api.Schema;
import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.api.extension.RegisterExtension;
-import java.util.Collection;
-import java.util.List;
-import java.util.function.Supplier;
-
/**
* The base class for the all Pulsar related test sites. It brings up:
*
@@ -49,67 +42,25 @@ import java.util.function.Supplier;
* you have some setup logic, such as create topic or send message, just place then in a setup
* method with annotation {@code @BeforeAll}. This setup method would not require {@code static}.
*
- * @see PulsarContainerOperator for how to use the helper methods in this class.
+ * @see PulsarRuntimeOperator for how to use the helper methods in this class.
*/
@ExtendWith(TestLoggerExtension.class)
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
public abstract class PulsarTestSuiteBase {
@RegisterExtension
- final PulsarContainerEnvironment environment = new PulsarContainerEnvironment();
-
- // Helper method for real test classes.
-
- protected void setupTopic(String topic) {
- environment.operator().setupTopic(topic);
- }
-
- public <T> void setupTopic(String topic, Schema<T> schema, Supplier<T> supplier) {
- environment.operator().setupTopic(topic, schema, supplier);
- }
-
- protected void createTopic(String topic, int numberOfPartitions) {
- environment.operator().createTopic(topic, numberOfPartitions);
- }
-
- protected void increaseTopicPartitions(String topic, int newPartitionsNum) {
- environment.operator().increaseTopicPartitions(topic, newPartitionsNum);
- }
-
- protected void deleteTopic(String topic, boolean isPartitioned) {
- environment.operator().deleteTopic(topic, isPartitioned);
- }
-
- protected List<TopicPartition> topicInfo(String topic) {
- return environment.operator().topicInfo(topic);
- }
-
- protected List<TopicPartition> topicsInfo(Collection<String> topics) {
- return environment.operator().topicsInfo(topics);
- }
-
- protected <T> MessageId sendMessage(String topic, Schema<T> schema, T message) {
- return environment.operator().sendMessage(topic, schema, message);
- }
-
- protected <T> List<MessageId> sendMessages(
- String topic, Schema<T> schema, Collection<T> messages) {
- return environment.operator().sendMessages(topic, schema, messages);
- }
-
- protected String serviceUrl() {
- return environment.operator().serviceUrl();
- }
-
- protected String adminUrl() {
- return environment.operator().adminUrl();
- }
-
- protected PulsarClient client() {
- return environment.operator().client();
+ final PulsarTestEnvironment environment = new PulsarTestEnvironment(runtime());
+
+ /**
+ * Choose the desired pulsar runtime as the test backend. The default test backend is a mocked
+ * pulsar broker. Override this method when needs.
+ */
+ protected PulsarRuntime runtime() {
+ return PulsarRuntime.MOCK;
}
- protected PulsarAdmin admin() {
- return environment.operator().admin();
+ /** Operate pulsar by acquiring a runtime operator. */
+ protected PulsarRuntimeOperator operator() {
+ return environment.operator();
}
}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/cases/MultipleTopicConsumingContext.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/cases/MultipleTopicConsumingContext.java
index ec0f952..60a0bfba 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/cases/MultipleTopicConsumingContext.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/cases/MultipleTopicConsumingContext.java
@@ -25,9 +25,9 @@ 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.TopicPartition;
-import org.apache.flink.connector.pulsar.testutils.PulsarContainerContext;
-import org.apache.flink.connector.pulsar.testutils.PulsarContainerEnvironment;
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;
@@ -46,7 +46,7 @@ import static org.apache.pulsar.client.api.SubscriptionType.Exclusive;
* Pulsar external context that will create multiple topics with only one partitions as source
* splits.
*/
-public class MultipleTopicConsumingContext extends PulsarContainerContext<String> {
+public class MultipleTopicConsumingContext extends PulsarTestContext<String> {
private int numTopics = 0;
@@ -55,7 +55,7 @@ public class MultipleTopicConsumingContext extends PulsarContainerContext<String
private final Map<String, SourceSplitDataWriter<String>> topicNameToSplitWriters =
new HashMap<>();
- public MultipleTopicConsumingContext(PulsarContainerEnvironment environment) {
+ public MultipleTopicConsumingContext(PulsarTestEnvironment environment) {
super("consuming message on multiple topic", environment);
this.topicPattern =
"pulsar-multiple-topic-[0-9]+-"
@@ -67,8 +67,8 @@ public class MultipleTopicConsumingContext extends PulsarContainerContext<String
PulsarSourceBuilder<String> builder =
PulsarSource.builder()
.setDeserializationSchema(pulsarSchema(STRING))
- .setServiceUrl(serviceUrl())
- .setAdminUrl(adminUrl())
+ .setServiceUrl(operator.serviceUrl())
+ .setAdminUrl(operator.adminUrl())
.setTopicPattern(topicPattern, RegexSubscriptionMode.AllTopics)
.setSubscriptionType(Exclusive)
.setSubscriptionName("flink-pulsar-multiple-topic-test");
@@ -84,11 +84,12 @@ public class MultipleTopicConsumingContext extends PulsarContainerContext<String
@Override
public SourceSplitDataWriter<String> createSourceSplitDataWriter() {
String topicName = topicPattern.replace("[0-9]+", String.valueOf(numTopics));
- createTopic(topicName, 1);
+ operator.createTopic(topicName, 1);
String partitionName = TopicNameUtils.topicNameWithPartition(topicName, 0);
TopicPartition partition = new TopicPartition(partitionName, 0, createFullRange());
- PulsarPartitionDataWriter writer = new PulsarPartitionDataWriter(client(), partition);
+ PulsarPartitionDataWriter writer =
+ new PulsarPartitionDataWriter(operator.client(), partition);
topicNameToSplitWriters.put(partitionName, writer);
numTopics++;
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/cases/SingleTopicConsumingContext.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/cases/SingleTopicConsumingContext.java
index 01507d9..288ecf3 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/cases/SingleTopicConsumingContext.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/cases/SingleTopicConsumingContext.java
@@ -24,9 +24,9 @@ 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.TopicPartition;
-import org.apache.flink.connector.pulsar.testutils.PulsarContainerContext;
-import org.apache.flink.connector.pulsar.testutils.PulsarContainerEnvironment;
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 java.util.Collection;
@@ -43,7 +43,7 @@ import static org.apache.pulsar.client.api.SubscriptionType.Exclusive;
* A Pulsar external context that will create only one topic and use partitions in that topic as
* source splits.
*/
-public class SingleTopicConsumingContext extends PulsarContainerContext<String> {
+public class SingleTopicConsumingContext extends PulsarTestContext<String> {
private static final String TOPIC_NAME_PREFIX = "pulsar-single-topic";
private final String topicName;
@@ -52,7 +52,7 @@ public class SingleTopicConsumingContext extends PulsarContainerContext<String>
private int numSplits = 0;
- public SingleTopicConsumingContext(PulsarContainerEnvironment environment) {
+ public SingleTopicConsumingContext(PulsarTestEnvironment environment) {
super("consuming message on single topic", environment);
this.topicName =
TOPIC_NAME_PREFIX + "-" + ThreadLocalRandom.current().nextLong(Long.MAX_VALUE);
@@ -63,8 +63,8 @@ public class SingleTopicConsumingContext extends PulsarContainerContext<String>
PulsarSourceBuilder<String> builder =
PulsarSource.builder()
.setDeserializationSchema(pulsarSchema(STRING))
- .setServiceUrl(serviceUrl())
- .setAdminUrl(adminUrl())
+ .setServiceUrl(operator.serviceUrl())
+ .setAdminUrl(operator.adminUrl())
.setTopics(topicName)
.setSubscriptionType(Exclusive)
.setSubscriptionName("pulsar-single-topic");
@@ -81,15 +81,16 @@ public class SingleTopicConsumingContext extends PulsarContainerContext<String>
public SourceSplitDataWriter<String> createSourceSplitDataWriter() {
if (numSplits == 0) {
// Create the topic first.
- createTopic(topicName, 1);
+ operator.createTopic(topicName, 1);
numSplits++;
} else {
numSplits++;
- increaseTopicPartitions(topicName, numSplits);
+ operator.increaseTopicPartitions(topicName, numSplits);
}
TopicPartition partition = new TopicPartition(topicName, numSplits - 1, createFullRange());
- PulsarPartitionDataWriter writer = new PulsarPartitionDataWriter(client(), partition);
+ PulsarPartitionDataWriter writer =
+ new PulsarPartitionDataWriter(operator.client(), partition);
partitionToSplitWriter.put(numSplits - 1, writer);
return writer;
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntime.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntime.java
new file mode 100644
index 0000000..986f4bd
--- /dev/null
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntime.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.testutils.runtime;
+
+import org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerProvider;
+import org.apache.flink.connector.pulsar.testutils.runtime.mock.PulsarMockProvider;
+
+import java.util.function.Supplier;
+
+/**
+ * A enum class for providing a operable pulsar runtime. We support two types of runtime, the
+ * container and mock.
+ */
+public enum PulsarRuntime {
+
+ /**
+ * The whole pulsar cluster would run in a docker container, provide the full fledged test
+ * backend.
+ */
+ CONTAINER(PulsarContainerProvider::new),
+
+ /** The bookkeeper and zookeeper would use a mock backend, and start a single pulsar broker. */
+ MOCK(PulsarMockProvider::new);
+
+ private final Supplier<PulsarRuntimeProvider> provider;
+
+ PulsarRuntime(Supplier<PulsarRuntimeProvider> provider) {
+ this.provider = provider;
+ }
+
+ public PulsarRuntimeProvider provider() {
+ return provider.get();
+ }
+}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarContainerOperator.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntimeOperator.java
similarity index 96%
rename from flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarContainerOperator.java
rename to flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntimeOperator.java
index 7142b54..83fe58f 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarContainerOperator.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntimeOperator.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.flink.connector.pulsar.testutils;
+package org.apache.flink.connector.pulsar.testutils.runtime;
import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils;
import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
@@ -55,10 +55,10 @@ import static org.apache.flink.connector.pulsar.common.utils.PulsarExceptionUtil
import static org.apache.flink.util.Preconditions.checkArgument;
/**
- * A pulsar container operator for operating pulsar instance. It's serializable for using in {@link
- * ExternalContext}.
+ * A pulsar cluster operator is used for operating pulsar instance. It's serializable for using in
+ * {@link ExternalContext}.
*/
-public class PulsarContainerOperator implements Serializable, Closeable {
+public class PulsarRuntimeOperator implements Serializable, Closeable {
private static final long serialVersionUID = -630646912412751301L;
public static final int DEFAULT_PARTITIONS = 10;
@@ -70,7 +70,7 @@ public class PulsarContainerOperator implements Serializable, Closeable {
private transient PulsarClient client;
private transient PulsarAdmin admin;
- public PulsarContainerOperator(String serviceUrl, String adminUrl) {
+ public PulsarRuntimeOperator(String serviceUrl, String adminUrl) {
this.serviceUrl = serviceUrl;
this.adminUrl = adminUrl;
initializeClients();
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntimeProvider.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntimeProvider.java
new file mode 100644
index 0000000..d8ad718
--- /dev/null
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntimeProvider.java
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.testutils.runtime;
+
+import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
+
+/**
+ * A abstraction for different pulsar runtimes. Providing the common methods for {@link
+ * PulsarTestEnvironment}.
+ */
+public interface PulsarRuntimeProvider {
+
+ /** Start up this pulsar runtime, block the thread until everytime is ready for this runtime. */
+ void startUp();
+
+ /** Shutdown this pulsar runtime. */
+ void tearDown();
+
+ /** Return a operator for operating this pulsar runtime. */
+ PulsarRuntimeOperator operator();
+}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/container/PulsarContainerProvider.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/container/PulsarContainerProvider.java
new file mode 100644
index 0000000..06be3cb
--- /dev/null
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/container/PulsarContainerProvider.java
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.testutils.runtime.container;
+
+import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntimeOperator;
+import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntimeProvider;
+import org.apache.flink.util.DockerImageVersions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.BindMode;
+import org.testcontainers.containers.PulsarContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.HttpWaitStrategy;
+import org.testcontainers.utility.DockerImageName;
+
+import java.io.IOException;
+import java.time.Duration;
+
+import static org.apache.flink.util.DockerImageVersions.PULSAR;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.testcontainers.containers.PulsarContainer.BROKER_HTTP_PORT;
+
+/**
+ * {@link PulsarRuntimeProvider} implementation, use the TestContainers as the backend. We would
+ * start a pulsar container by this provider.
+ */
+public class PulsarContainerProvider implements PulsarRuntimeProvider {
+ private static final Logger LOG = LoggerFactory.getLogger(PulsarContainerProvider.class);
+
+ /**
+ * Create a pulsar container provider by a predefined version, this constance {@link
+ * DockerImageVersions#PULSAR} should be bumped after the new pulsar release.
+ */
+ private final PulsarContainer container = new PulsarContainer(DockerImageName.parse(PULSAR));
+
+ private PulsarRuntimeOperator operator;
+
+ @Override
+ public void startUp() {
+ // Prepare Pulsar Container.
+ container.withClasspathResourceMapping(
+ "containers/txnStandalone.conf",
+ "/pulsar/conf/standalone.conf",
+ BindMode.READ_ONLY);
+ container.addExposedPort(2181);
+ container.waitingFor(
+ new HttpWaitStrategy()
+ .forPort(BROKER_HTTP_PORT)
+ .forStatusCode(200)
+ .forPath("/admin/v2/namespaces/public/default")
+ .withStartupTimeout(Duration.ofMinutes(5)));
+
+ // Start the Pulsar Container.
+ container.start();
+ container.followOutput(new Slf4jLogConsumer(LOG).withSeparateOutputStreams());
+
+ // Create the operator.
+ this.operator =
+ new PulsarRuntimeOperator(
+ container.getPulsarBrokerUrl(), container.getHttpServiceUrl());
+ }
+
+ @Override
+ public void tearDown() {
+ try {
+ operator.close();
+ this.operator = null;
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ container.stop();
+ }
+
+ @Override
+ public PulsarRuntimeOperator operator() {
+ return checkNotNull(operator, "You should start this pulsar container first.");
+ }
+}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/BlankBrokerInterceptor.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/BlankBrokerInterceptor.java
new file mode 100644
index 0000000..8355a23
--- /dev/null
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/BlankBrokerInterceptor.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.connector.pulsar.testutils.runtime.mock;
+
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.intercept.BrokerInterceptor;
+import org.apache.pulsar.broker.service.ServerCnx;
+import org.apache.pulsar.common.api.proto.BaseCommand;
+
+import javax.servlet.ServletRequest;
+import javax.servlet.ServletResponse;
+
+/** No operation for this BrokerInterceptor implementation. */
+public class BlankBrokerInterceptor implements BrokerInterceptor {
+
+ @Override
+ public void onPulsarCommand(BaseCommand command, ServerCnx cnx) {
+ // no-op
+ }
+
+ @Override
+ public void onConnectionClosed(ServerCnx cnx) {
+ // no-op
+ }
+
+ @Override
+ public void onWebserviceRequest(ServletRequest request) {
+ // no-op
+ }
+
+ @Override
+ public void onWebserviceResponse(ServletRequest request, ServletResponse response) {
+ // no-op
+ }
+
+ @Override
+ public void initialize(PulsarService pulsarService) {
+ // no-op
+ }
+
+ @Override
+ public void close() {
+ // no-op
+ }
+}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/MockBookKeeperClientFactory.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/MockBookKeeperClientFactory.java
new file mode 100644
index 0000000..41fad54
--- /dev/null
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/MockBookKeeperClientFactory.java
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.testutils.runtime.mock;
+
+import io.netty.channel.EventLoopGroup;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.EnsemblePlacementPolicy;
+import org.apache.bookkeeper.common.util.OrderedExecutor;
+import org.apache.bookkeeper.stats.StatsLogger;
+import org.apache.pulsar.broker.BookKeeperClientFactory;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.zookeeper.ZooKeeper;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Optional;
+
+/** A BookKeeperClientFactory implementation which returns a mocked bookkeeper. */
+public class MockBookKeeperClientFactory implements BookKeeperClientFactory {
+
+ private final OrderedExecutor executor =
+ OrderedExecutor.newBuilder().numThreads(1).name("mock-pulsar-bookkeeper").build();
+
+ private final BookKeeper bookKeeper = NonClosableMockBookKeeper.create(executor);
+
+ @Override
+ public BookKeeper create(
+ ServiceConfiguration conf,
+ ZooKeeper zkClient,
+ EventLoopGroup eventLoopGroup,
+ Optional<Class<? extends EnsemblePlacementPolicy>> ensemblePlacementPolicyClass,
+ Map<String, Object> ensemblePlacementPolicyProperties)
+ throws IOException {
+ return bookKeeper;
+ }
+
+ @Override
+ public BookKeeper create(
+ ServiceConfiguration conf,
+ ZooKeeper zkClient,
+ EventLoopGroup eventLoopGroup,
+ Optional<Class<? extends EnsemblePlacementPolicy>> ensemblePlacementPolicyClass,
+ Map<String, Object> ensemblePlacementPolicyProperties,
+ StatsLogger statsLogger)
+ throws IOException {
+ return bookKeeper;
+ }
+
+ @Override
+ public void close() {
+ try {
+ bookKeeper.close();
+ executor.shutdown();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/MockPulsarService.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/MockPulsarService.java
new file mode 100644
index 0000000..6b6c412
--- /dev/null
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/MockPulsarService.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.connector.pulsar.testutils.runtime.mock;
+
+import org.apache.bookkeeper.common.util.OrderedExecutor;
+import org.apache.pulsar.broker.BookKeeperClientFactory;
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.intercept.BrokerInterceptor;
+import org.apache.pulsar.broker.namespace.NamespaceService;
+import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
+import org.apache.pulsar.metadata.impl.ZKMetadataStore;
+import org.apache.pulsar.zookeeper.ZooKeeperClientFactory;
+import org.apache.zookeeper.MockZooKeeperSession;
+
+import java.util.function.Supplier;
+
+/** A Mock pulsar service which would use the mocked zookeeper and bookkeeper. */
+public class MockPulsarService extends PulsarService {
+
+ private final int brokerServicePort;
+
+ private final MockZooKeeperClientFactory zooKeeperClientFactory =
+ new MockZooKeeperClientFactory();
+
+ private final MockZooKeeperSession zooKeeperSession =
+ MockZooKeeperSession.newInstance(zooKeeperClientFactory.getZooKeeper());
+
+ private final SameThreadOrderedSafeExecutor orderedExecutor =
+ new SameThreadOrderedSafeExecutor();
+
+ public MockPulsarService(ServiceConfiguration config) {
+ super(config);
+ this.brokerServicePort =
+ config.getBrokerServicePort().orElseThrow(IllegalArgumentException::new);
+ }
+
+ public ZooKeeperClientFactory getZooKeeperClientFactory() {
+ return zooKeeperClientFactory;
+ }
+
+ public BookKeeperClientFactory newBookKeeperClientFactory() {
+ return new MockBookKeeperClientFactory();
+ }
+
+ public MetadataStoreExtended createLocalMetadataStore() {
+ return new ZKMetadataStore(zooKeeperSession);
+ }
+
+ public MetadataStoreExtended createConfigurationMetadataStore() {
+ return new ZKMetadataStore(zooKeeperSession);
+ }
+
+ public Supplier<NamespaceService> getNamespaceServiceProvider() {
+ return () -> new NamespaceService(this);
+ }
+
+ @Override
+ public OrderedExecutor getOrderedExecutor() {
+ return orderedExecutor;
+ }
+
+ @Override
+ public BrokerInterceptor getBrokerInterceptor() {
+ return new BlankBrokerInterceptor();
+ }
+
+ public int getBrokerServicePort() {
+ return brokerServicePort;
+ }
+}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/MockZooKeeperClientFactory.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/MockZooKeeperClientFactory.java
new file mode 100644
index 0000000..3c89484
--- /dev/null
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/MockZooKeeperClientFactory.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.testutils.runtime.mock;
+
+import org.apache.flink.shaded.guava30.com.google.common.util.concurrent.MoreExecutors;
+
+import org.apache.bookkeeper.util.ZkUtils;
+import org.apache.pulsar.zookeeper.ZooKeeperClientFactory;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.MockZooKeeper;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.data.ACL;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+import static org.apache.pulsar.zookeeper.ZookeeperClientFactoryImpl.ENCODING_SCHEME;
+import static org.apache.zookeeper.CreateMode.PERSISTENT;
+
+/** A ZooKeeperClientFactory implementation which returns mocked zookeeper instead of normal zk. */
+public class MockZooKeeperClientFactory implements ZooKeeperClientFactory {
+
+ private final MockZooKeeper zooKeeper;
+
+ public MockZooKeeperClientFactory() {
+ this.zooKeeper = MockZooKeeper.newInstance(MoreExecutors.newDirectExecutorService());
+ List<ACL> dummyAclList = new ArrayList<>(0);
+
+ try {
+ ZkUtils.createFullPathOptimistic(
+ zooKeeper,
+ "/ledgers/available/192.168.1.1:" + 5000,
+ "".getBytes(ENCODING_SCHEME),
+ dummyAclList,
+ PERSISTENT);
+
+ zooKeeper.create(
+ "/ledgers/LAYOUT",
+ "1\nflat:1".getBytes(ENCODING_SCHEME),
+ dummyAclList,
+ PERSISTENT);
+ } catch (KeeperException | InterruptedException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ @Override
+ public CompletableFuture<ZooKeeper> create(
+ String serverList, SessionType sessionType, int zkSessionTimeoutMillis) {
+ return CompletableFuture.completedFuture(zooKeeper);
+ }
+
+ MockZooKeeper getZooKeeper() {
+ return zooKeeper;
+ }
+}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/NonClosableMockBookKeeper.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/NonClosableMockBookKeeper.java
new file mode 100644
index 0000000..b7001b8
--- /dev/null
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/NonClosableMockBookKeeper.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.connector.pulsar.testutils.runtime.mock;
+
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.PulsarMockBookKeeper;
+import org.apache.bookkeeper.common.util.OrderedExecutor;
+
+/**
+ * Prevent the MockBookKeeper instance from being closed when the broker is restarted within a test.
+ */
+public class NonClosableMockBookKeeper extends PulsarMockBookKeeper {
+
+ private NonClosableMockBookKeeper(OrderedExecutor executor) throws Exception {
+ super(executor);
+ }
+
+ @Override
+ public void close() {
+ // no-op
+ }
+
+ @Override
+ public void shutdown() {
+ // no-op
+ }
+
+ public void reallyShutdown() {
+ super.shutdown();
+ }
+
+ public static BookKeeper create(OrderedExecutor executor) {
+ try {
+ return new NonClosableMockBookKeeper(executor);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/PortBindingUtils.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/PortBindingUtils.java
new file mode 100644
index 0000000..41035db
--- /dev/null
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/PortBindingUtils.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.testutils.runtime.mock;
+
+import javax.net.ServerSocketFactory;
+
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.util.concurrent.ThreadLocalRandom;
+
+/**
+ * This class is used for finding a available socket port. Use for running multiple instances of the
+ * bookkeeper tests at once.
+ */
+public final class PortBindingUtils {
+
+ private static final int MIN_PORT = 1024;
+ private static final int MAX_PORT = 65535;
+
+ private PortBindingUtils() {
+ // No public constructor
+ }
+
+ public static int findAvailablePort() {
+ return findAvailablePort(MIN_PORT, MAX_PORT);
+ }
+
+ public static int findAvailablePort(int minPort, int maxPort) {
+ int port;
+ do {
+ port = randomPort(minPort, maxPort);
+ } while (!isPortAvailable(port));
+
+ return port;
+ }
+
+ private static int randomPort(int minPort, int maxPort) {
+ int delta = maxPort - minPort;
+ return ThreadLocalRandom.current().nextInt(delta) + minPort;
+ }
+
+ public static boolean isPortAvailable(int port) {
+ try (ServerSocket serverSocket =
+ ServerSocketFactory.getDefault()
+ .createServerSocket(port, 1, InetAddress.getByName("localhost"))) {
+ return true;
+ } catch (Exception ex) {
+ return false;
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/PulsarMockProvider.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/PulsarMockProvider.java
new file mode 100644
index 0000000..1cb8ce9
--- /dev/null
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/PulsarMockProvider.java
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.testutils.runtime.mock;
+
+import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntimeOperator;
+import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntimeProvider;
+
+import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableSet;
+
+import org.apache.pulsar.broker.PulsarServerException;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.RetentionPolicies;
+import org.apache.pulsar.common.policies.data.TenantInfo;
+
+import java.util.Optional;
+
+import static org.apache.commons.lang3.RandomStringUtils.randomAlphanumeric;
+import static org.apache.flink.connector.pulsar.common.utils.PulsarExceptionUtils.sneakyAdmin;
+import static org.apache.flink.connector.pulsar.testutils.runtime.mock.PortBindingUtils.findAvailablePort;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/** Providing a mocked pulsar server. */
+public class PulsarMockProvider implements PulsarRuntimeProvider {
+
+ private static final String CLUSTER_NAME = "mock-pulsar-" + randomAlphanumeric(6);
+ private final MockPulsarService pulsarService;
+ private PulsarRuntimeOperator operator;
+
+ public PulsarMockProvider() {
+ this(createConfig());
+ }
+
+ public PulsarMockProvider(ServiceConfiguration configuration) {
+ this.pulsarService = new MockPulsarService(configuration);
+ }
+
+ @Override
+ public void startUp() {
+ try {
+ pulsarService.start();
+ } catch (PulsarServerException e) {
+ throw new IllegalStateException(e);
+ }
+ this.operator =
+ new PulsarRuntimeOperator(
+ pulsarService.getBrokerServiceUrl(), pulsarService.getWebServiceAddress());
+
+ // Successfully start a pulsar broker, we have to create the required resources.
+ sneakyAdmin(this::createTestResource);
+ }
+
+ @Override
+ public void tearDown() {
+ try {
+ pulsarService.close();
+ operator.close();
+ this.operator = null;
+ } catch (Exception e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ @Override
+ public PulsarRuntimeOperator operator() {
+ return checkNotNull(operator, "You should start this mock pulsar first.");
+ }
+
+ private void createTestResource() throws PulsarAdminException {
+ PulsarAdmin admin = operator().admin();
+ if (!admin.clusters().getClusters().contains(CLUSTER_NAME)) {
+ // Make clients can test short names
+ ClusterData data =
+ ClusterData.builder()
+ .serviceUrl("http://127.0.0.1:" + pulsarService.getBrokerServicePort())
+ .build();
+ admin.clusters().createCluster(CLUSTER_NAME, data);
+ }
+
+ TenantInfo info =
+ TenantInfo.builder()
+ .adminRoles(ImmutableSet.of("appid1", "appid2"))
+ .allowedClusters(ImmutableSet.of(CLUSTER_NAME))
+ .build();
+ if (!admin.tenants().getTenants().contains("public")) {
+ admin.tenants().createTenant("public", info);
+ } else {
+ admin.tenants().updateTenant("public", info);
+ }
+
+ if (!admin.namespaces().getNamespaces("public").contains("public/default")) {
+ admin.namespaces().createNamespace("public/default");
+ admin.namespaces().setRetention("public/default", new RetentionPolicies(60, 1000));
+ }
+ }
+
+ private static ServiceConfiguration createConfig() {
+ ServiceConfiguration configuration = new ServiceConfiguration();
+
+ configuration.setAdvertisedAddress("localhost");
+ configuration.setClusterName(CLUSTER_NAME);
+
+ configuration.setManagedLedgerCacheSizeMB(8);
+ configuration.setActiveConsumerFailoverDelayTimeMillis(0);
+ configuration.setDefaultRetentionTimeInMinutes(7);
+ configuration.setDefaultNumberOfNamespaceBundles(1);
+ configuration.setZookeeperServers("localhost:2181");
+ configuration.setConfigurationStoreServers("localhost:3181");
+
+ configuration.setAuthenticationEnabled(false);
+ configuration.setAuthorizationEnabled(false);
+ configuration.setAllowAutoTopicCreation(true);
+ configuration.setBrokerDeleteInactiveTopicsEnabled(false);
+
+ configuration.setWebSocketServiceEnabled(false);
+ // Binding Ports
+ configuration.setBrokerServicePort(Optional.of(findAvailablePort()));
+ configuration.setWebServicePort(Optional.of(findAvailablePort()));
+
+ return configuration;
+ }
+}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/SameThreadOrderedSafeExecutor.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/SameThreadOrderedSafeExecutor.java
new file mode 100644
index 0000000..9667f08
--- /dev/null
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/mock/SameThreadOrderedSafeExecutor.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.pulsar.testutils.runtime.mock;
+
+import org.apache.bookkeeper.common.util.OrderedExecutor;
+import org.apache.bookkeeper.common.util.SafeRunnable;
+import org.apache.bookkeeper.stats.NullStatsLogger;
+import org.apache.pulsar.shade.io.netty.util.concurrent.DefaultThreadFactory;
+
+/** Override the default bookkeeper executor for executing in one thread executor. */
+public class SameThreadOrderedSafeExecutor extends OrderedExecutor {
+
+ public SameThreadOrderedSafeExecutor() {
+ super(
+ "same-thread-executor",
+ 1,
+ new DefaultThreadFactory("test"),
+ NullStatsLogger.INSTANCE,
+ false,
+ false,
+ 100000,
+ -1,
+ false);
+ }
+
+ @Override
+ public void execute(Runnable r) {
+ r.run();
+ }
+
+ @Override
+ public void executeOrdered(int orderingKey, SafeRunnable r) {
+ r.run();
+ }
+
+ @Override
+ public void executeOrdered(long orderingKey, SafeRunnable r) {
+ r.run();
+ }
+}
diff --git a/tools/maven/suppressions.xml b/tools/maven/suppressions.xml
index 9cb8d3a..84e5da3 100644
--- a/tools/maven/suppressions.xml
+++ b/tools/maven/suppressions.xml
@@ -61,7 +61,11 @@ under the License.
<suppress
files="FanOutRecordPublisherTest.java|FanOutShardSubscriber.java|FanOutShardSubscriberTest.java"
checks="IllegalImport"/>
- <!-- Classes copied from Hadoop -->
+ <!-- Pulsar broker required to handle Netty EventLoopGroup -->
+ <suppress
+ files="MockBookKeeperClientFactory.java"
+ checks="IllegalImport"/>
+ <!-- Classes copied from Hadoop -->
<suppress
files="org[\\/]apache[\\/]hadoop[\\/]conf[\\/]Configuration.java"
checks=".*"/>
[flink] 04/06: [FLINK-23877][connector/pulsar] Change the config
setter in pulsar source builder. Drop Properties. Remove useless config
options.
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 0fc900849db32fc495fad482b0e47eb59d02bfe8
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Thu Aug 26 16:03:21 2021 +0800
[FLINK-23877][connector/pulsar] Change the config setter in pulsar source builder. Drop Properties. Remove useless config options.
---
.../pulsar/common/config/PulsarOptions.java | 6 -
.../pulsar/source/PulsarSourceBuilder.java | 134 +++++++--------------
.../pulsar/source/PulsarSourceOptions.java | 48 +-------
3 files changed, 45 insertions(+), 143 deletions(-)
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarOptions.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarOptions.java
index 0891e7b..ed71c91 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarOptions.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarOptions.java
@@ -164,12 +164,6 @@ public final class PulsarOptions {
.withDescription(
"Whether to use TCP no-delay flag on the connection to disable Nagle algorithm.");
- public static final ConfigOption<Boolean> PULSAR_USE_TLS =
- ConfigOptions.key(CLIENT_CONFIG_PREFIX + "useTls")
- .booleanType()
- .defaultValue(false)
- .withDescription("Whether to use TLS encryption on the connection.");
-
public static final ConfigOption<String> PULSAR_TLS_TRUST_CERTS_FILE_PATH =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "tlsTrustCertsFilePath")
.stringType()
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
index 20064f5..b52f08a 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
@@ -22,9 +22,9 @@ import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.configuration.ConfigOption;
-import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.UnmodifiableConfiguration;
+import org.apache.flink.connector.pulsar.common.config.PulsarOptions;
import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
import org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber;
@@ -40,10 +40,11 @@ import org.apache.pulsar.client.api.SubscriptionType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
-import java.util.Properties;
import java.util.regex.Pattern;
import static java.lang.Boolean.FALSE;
@@ -52,11 +53,8 @@ import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULS
import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SERVICE_URL;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_ENABLE_AUTO_ACKNOWLEDGE_MESSAGE;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS;
-import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_REGEX_SUBSCRIPTION_MODE;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_NAME;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_TYPE;
-import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPICS_PATTERN;
-import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPIC_NAMES;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TRANSACTION_TIMEOUT_MILLIS;
import static org.apache.flink.connector.pulsar.source.config.PulsarSourceConfigUtils.checkConfigurations;
import static org.apache.flink.util.InstantiationUtil.isSerializable;
@@ -134,8 +132,7 @@ public final class PulsarSourceBuilder<OUT> {
* @return this PulsarSourceBuilder.
*/
public PulsarSourceBuilder<OUT> setAdminUrl(String adminUrl) {
- setConfiguration(PULSAR_ADMIN_URL, adminUrl);
- return this;
+ return setConfig(PULSAR_ADMIN_URL, adminUrl);
}
/**
@@ -145,8 +142,7 @@ public final class PulsarSourceBuilder<OUT> {
* @return this PulsarSourceBuilder.
*/
public PulsarSourceBuilder<OUT> setServiceUrl(String serviceUrl) {
- setConfiguration(PULSAR_SERVICE_URL, serviceUrl);
- return this;
+ return setConfig(PULSAR_SERVICE_URL, serviceUrl);
}
/**
@@ -156,8 +152,7 @@ public final class PulsarSourceBuilder<OUT> {
* @return this PulsarSourceBuilder.
*/
public PulsarSourceBuilder<OUT> setSubscriptionName(String subscriptionName) {
- setConfiguration(PULSAR_SUBSCRIPTION_NAME, subscriptionName);
- return this;
+ return setConfig(PULSAR_SUBSCRIPTION_NAME, subscriptionName);
}
/**
@@ -208,7 +203,6 @@ public final class PulsarSourceBuilder<OUT> {
*/
public PulsarSourceBuilder<OUT> setTopics(List<String> topics) {
ensureSubscriberIsNull("topics");
- configuration.set(PULSAR_TOPIC_NAMES, topics);
this.subscriber = PulsarSubscriber.getTopicListSubscriber(topics);
return this;
}
@@ -253,14 +247,19 @@ public final class PulsarSourceBuilder<OUT> {
* with {@link #setTopics} or {@link #setTopicPattern} in this builder.
*
* @param topicsPattern the pattern of the topic name to consume from.
- * @param regexSubscriptionMode The topic filter for regex subscription.
+ * @param regexSubscriptionMode When subscribing to a topic using a regular expression, you can
+ * pick a certain type of topics.
+ * <ul>
+ * <li>PersistentOnly: only subscribe to persistent topics.
+ * <li>NonPersistentOnly: only subscribe to non-persistent topics.
+ * <li>AllTopics: subscribe to both persistent and non-persistent topics.
+ * </ul>
+ *
* @return this PulsarSourceBuilder.
*/
public PulsarSourceBuilder<OUT> setTopicPattern(
Pattern topicsPattern, RegexSubscriptionMode regexSubscriptionMode) {
ensureSubscriberIsNull("topic pattern");
- configuration.set(PULSAR_TOPICS_PATTERN, topicsPattern.toString());
- configuration.set(PULSAR_REGEX_SUBSCRIPTION_MODE, regexSubscriptionMode);
this.subscriber =
PulsarSubscriber.getTopicPatternSubscriber(topicsPattern, regexSubscriptionMode);
return this;
@@ -284,7 +283,7 @@ public final class PulsarSourceBuilder<OUT> {
LOG.warn("No subscription type provided, set it to Key_Shared.");
setSubscriptionType(SubscriptionType.Key_Shared);
}
- this.rangeGenerator = rangeGenerator;
+ this.rangeGenerator = checkNotNull(rangeGenerator);
return this;
}
@@ -360,64 +359,52 @@ public final class PulsarSourceBuilder<OUT> {
/**
* Set an arbitrary property for the PulsarSource and PulsarConsumer. The valid keys can be
- * found in {@link PulsarSourceOptions}.
+ * found in {@link PulsarSourceOptions} and {@link PulsarOptions}.
*
- * @param key the key of the property.
- * @param value the value of the property.
- * @return this PulsarSourceBuilder.
- */
- public PulsarSourceBuilder<OUT> setProperty(String key, String value) {
- checkNotNull(key);
- checkNotNull(value);
- if (configuration.containsKey(key)) {
- ConfigOption<String> rawOption = ConfigOptions.key(key).stringType().noDefaultValue();
- LOG.warn(
- "Config option {} already has a value {}, override to new value {}",
- key,
- configuration.getString(rawOption),
- value);
- }
- configuration.setString(key, value);
- return this;
- }
-
- /**
- * Set an arbitrary property for the PulsarSource and PulsarConsumer. The valid keys can be
- * found in {@link PulsarSourceOptions}.
+ * <p>Make sure the option could be set only once or with same value.
*
* @param key the key of the property.
* @param value the value of the property.
* @return this PulsarSourceBuilder.
*/
- public <T> PulsarSourceBuilder<OUT> setProperty(ConfigOption<T> key, T value) {
+ public <T> PulsarSourceBuilder<OUT> setConfig(ConfigOption<T> key, T value) {
checkNotNull(key);
checkNotNull(value);
if (configuration.contains(key)) {
T oldValue = configuration.get(key);
- LOG.warn(
- "Config option {} already has a value {}, override to new value {}",
- key,
- oldValue,
- value);
+ checkArgument(
+ Objects.equals(oldValue, value),
+ "This option %s has been already set to value %s.",
+ key.key(),
+ oldValue);
+ } else {
+ configuration.set(key, value);
}
- configuration.set(key, value);
return this;
}
/**
* Set arbitrary properties for the PulsarSource and PulsarConsumer. The valid keys can be found
- * in {@link PulsarSourceOptions}.
+ * in {@link PulsarSourceOptions} and {@link PulsarOptions}.
*
- * @param properties the properties to set for the PulsarSource.
+ * @param config the config to set for the PulsarSource.
* @return this PulsarSourceBuilder.
*/
- public PulsarSourceBuilder<OUT> setProperties(Properties properties) {
- for (String name : properties.stringPropertyNames()) {
- String value = properties.getProperty(name);
- if (value != null) {
- setProperty(name, value);
+ public PulsarSourceBuilder<OUT> setConfig(Configuration config) {
+ Map<String, String> existedConfigs = configuration.toMap();
+ List<String> duplicatedKeys = new ArrayList<>();
+ for (Map.Entry<String, String> entry : config.toMap().entrySet()) {
+ String key = entry.getKey();
+ String value2 = existedConfigs.get(key);
+ if (value2 != null && !value2.equals(entry.getValue())) {
+ duplicatedKeys.add(key);
}
}
+ checkArgument(
+ duplicatedKeys.isEmpty(),
+ "Invalid configuration, these keys %s are already exist with different config value.",
+ duplicatedKeys);
+ configuration.addAll(config);
return this;
}
@@ -431,31 +418,10 @@ public final class PulsarSourceBuilder<OUT> {
// Check builder configuration.
checkConfigurations(configuration);
- SubscriptionType subscriptionType = configuration.get(PULSAR_SUBSCRIPTION_TYPE);
-
// Ensure the topic subscriber for pulsar.
- if (subscriber == null) {
- if (configuration.contains(PULSAR_TOPIC_NAMES)) {
- List<String> topics = configuration.get(PULSAR_TOPIC_NAMES);
- this.subscriber = PulsarSubscriber.getTopicListSubscriber(topics);
- } else if (configuration.contains(PULSAR_TOPICS_PATTERN)) {
- String topicPatternStr = configuration.get(PULSAR_TOPICS_PATTERN);
- Pattern topicPattern = Pattern.compile(topicPatternStr);
- RegexSubscriptionMode regexSubscriptionMode =
- configuration.get(PULSAR_REGEX_SUBSCRIPTION_MODE);
-
- this.subscriber =
- PulsarSubscriber.getTopicPatternSubscriber(
- topicPattern, regexSubscriptionMode);
- } else {
- throw new IllegalArgumentException(
- "No "
- + PULSAR_TOPIC_NAMES.key()
- + " or "
- + PULSAR_TOPICS_PATTERN.key()
- + " was provided.");
- }
- }
+ checkNotNull(subscriber, "No topic names or topic pattern are provided.");
+
+ SubscriptionType subscriptionType = configuration.get(PULSAR_SUBSCRIPTION_TYPE);
if (subscriptionType == SubscriptionType.Key_Shared) {
if (rangeGenerator == null) {
LOG.warn(
@@ -523,20 +489,6 @@ public final class PulsarSourceBuilder<OUT> {
// ------------- private helpers --------------
- /** Make sure this option is set only once or with same value. */
- private <T> void setConfiguration(ConfigOption<T> option, T value) {
- if (configuration.contains(option)) {
- T oldValue = configuration.get(option);
- checkArgument(
- Objects.equals(oldValue, value),
- "This option %s has been set to value %s.",
- option.key(),
- oldValue);
- } else {
- configuration.set(option, value);
- }
- }
-
/** Helper method for java compiler recognize the generic type. */
@SuppressWarnings("unchecked")
private <T extends OUT> PulsarSourceBuilder<T> specialized() {
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java
index 1e45a3d..34938f3 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java
@@ -26,13 +26,11 @@ import org.apache.flink.connector.pulsar.common.config.PulsarOptions;
import org.apache.flink.connector.pulsar.source.config.CursorVerification;
import org.apache.pulsar.client.api.ConsumerCryptoFailureAction;
-import org.apache.pulsar.client.api.RegexSubscriptionMode;
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
import org.apache.pulsar.client.api.SubscriptionMode;
import org.apache.pulsar.client.api.SubscriptionType;
import java.time.Duration;
-import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
@@ -42,8 +40,8 @@ import static org.apache.flink.configuration.description.TextElement.text;
/**
* Configurations for PulsarSource. All the options list here could be configured in {@link
- * PulsarSourceBuilder#setProperty(ConfigOption, Object)}. The {@link PulsarOptions} is also
- * required for pulsar source.
+ * PulsarSourceBuilder#setConfig(ConfigOption, Object)}. The {@link PulsarOptions} is also required
+ * for pulsar source.
*
* @see PulsarOptions
*/
@@ -161,19 +159,6 @@ public final class PulsarSourceOptions {
//
///////////////////////////////////////////////////////////////////////////////
- public static final ConfigOption<List<String>> PULSAR_TOPIC_NAMES =
- ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "topicNames")
- .stringType()
- .asList()
- .defaultValues()
- .withDescription("Topic name.");
-
- public static final ConfigOption<String> PULSAR_TOPICS_PATTERN =
- ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "topicsPattern")
- .stringType()
- .noDefaultValue()
- .withDescription("Topic pattern.");
-
public static final ConfigOption<String> PULSAR_SUBSCRIPTION_NAME =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "subscriptionName")
.stringType()
@@ -414,35 +399,6 @@ public final class PulsarSourceOptions {
.withDescription(
"Initial position at which to set cursor when subscribing to a topic at first time.");
- public static final ConfigOption<Integer> PULSAR_PATTERN_AUTO_DISCOVERY_PERIOD =
- ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "patternAutoDiscoveryPeriod")
- .intType()
- .defaultValue(60)
- .withDescription(
- Description.builder()
- .text(
- "Topic auto discovery period when using a pattern for topic's consumer.")
- .linebreak()
- .text("The default and minimum value is 1 minute.")
- .build());
-
- public static final ConfigOption<RegexSubscriptionMode> PULSAR_REGEX_SUBSCRIPTION_MODE =
- ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "regexSubscriptionMode")
- .enumType(RegexSubscriptionMode.class)
- .defaultValue(RegexSubscriptionMode.PersistentOnly)
- .withDescription(
- Description.builder()
- .text(
- "When subscribing to a topic using a regular expression, you can pick a certain type of topics.")
- .list(
- text(
- "PersistentOnly: only subscribe to persistent topics."),
- text(
- "NonPersistentOnly: only subscribe to non-persistent topics."),
- text(
- "AllTopics: subscribe to both persistent and non-persistent topics."))
- .build());
-
// The config set for DeadLetterPolicy
/**
[flink] 02/06: [FLINK-23877][connector/pulsar] Drop the
ConfigurationDataCustomizer,
use ClientBuilder and PulsarAdminBuilder for creating instance.
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 2a55242f37a47a49c42b6b7b107475e5b443e49c
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Thu Aug 26 04:24:23 2021 +0800
[FLINK-23877][connector/pulsar] Drop the ConfigurationDataCustomizer, use ClientBuilder and PulsarAdminBuilder for creating instance.
---
.../common/config/ConfigurationDataCustomizer.java | 58 ----
.../pulsar/common/config/PulsarConfigUtils.java | 299 +++++++++++----------
.../pulsar/common/utils/PulsarJsonUtils.java | 15 --
.../connector/pulsar/source/PulsarSource.java | 28 +-
.../pulsar/source/PulsarSourceBuilder.java | 65 +----
.../source/config/PulsarSourceConfigUtils.java | 217 ++++++---------
.../source/enumerator/PulsarSourceEnumerator.java | 5 +-
.../source/reader/PulsarSourceReaderFactory.java | 14 +-
.../split/PulsarOrderedPartitionSplitReader.java | 12 +-
.../split/PulsarPartitionSplitReaderBase.java | 41 +--
.../split/PulsarUnorderedPartitionSplitReader.java | 12 +-
.../enumerator/PulsarSourceEnumeratorTest.java | 11 +-
12 files changed, 273 insertions(+), 504 deletions(-)
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/ConfigurationDataCustomizer.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/ConfigurationDataCustomizer.java
deleted file mode 100644
index 9605c2b..0000000
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/ConfigurationDataCustomizer.java
+++ /dev/null
@@ -1,58 +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.connector.pulsar.common.config;
-
-import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.configuration.Configuration;
-
-import java.io.Serializable;
-
-/**
- * Pulsar's config have some non-serializable fields which should be provided by user. You can add
- * modify the config instance by using this interface.
- *
- * @param <T> The class type for related pulsar config.
- */
-@PublicEvolving
-@FunctionalInterface
-public interface ConfigurationDataCustomizer<T> extends Serializable {
-
- /**
- * Modify the given config instance, this instance is created from {@link Configuration} and
- * would be used in the later logic.
- */
- void customize(T configurationDate);
-
- /**
- * Returns a composed customizer that first applies this customizer to its input, and then
- * applies the {@code after} customizer to the result.
- */
- default ConfigurationDataCustomizer<T> compose(ConfigurationDataCustomizer<T> after) {
- return data -> {
- customize(data);
- // Apply the given customizer.
- after.customize(data);
- };
- }
-
- /** Return a blank customizer. */
- static <Y> ConfigurationDataCustomizer<Y> blankCustomizer() {
- return a -> {};
- }
-}
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarConfigUtils.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarConfigUtils.java
index 9c8bd73..fa9c143 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarConfigUtils.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarConfigUtils.java
@@ -22,23 +22,24 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils;
-import org.apache.flink.connector.pulsar.source.PulsarSourceOptions;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminBuilder;
import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.client.api.AuthenticationFactory;
import org.apache.pulsar.client.api.ClientBuilder;
+import org.apache.pulsar.client.api.ProxyProtocol;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.impl.auth.AuthenticationDisabled;
-import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
-
-import javax.annotation.Nullable;
import java.util.Map;
+import java.util.function.Consumer;
import java.util.function.Function;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static java.util.function.Function.identity;
import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL;
import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTH_PARAMS;
import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTH_PARAM_MAP;
@@ -65,7 +66,6 @@ import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULS
import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_PROXY_SERVICE_URL;
import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_READ_TIMEOUT;
import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_REQUEST_TIMEOUT;
-import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_REQUEST_TIMEOUT_MS;
import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SERVICE_URL;
import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SSL_PROVIDER;
import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_STATS_INTERVAL_SECONDS;
@@ -79,9 +79,8 @@ import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULS
import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_TLS_TRUST_STORE_TYPE;
import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_USE_KEY_STORE_TLS;
import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_USE_TCP_NO_DELAY;
-import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_USE_TLS;
import static org.apache.flink.connector.pulsar.common.utils.PulsarExceptionUtils.sneakyClient;
-import static org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils.configMap;
+import static org.apache.pulsar.client.api.SizeUnit.BYTES;
/** The util for creating pulsar configuration class from flink's {@link Configuration}. */
@Internal
@@ -91,77 +90,150 @@ public final class PulsarConfigUtils {
// No need to create instance.
}
+ /** Create a PulsarClient by using the flink Configuration and the config customizer. */
+ public static PulsarClient createClient(Configuration configuration) {
+ ClientBuilder builder = PulsarClient.builder();
+
+ setOptionValue(configuration, PULSAR_SERVICE_URL, builder::serviceUrl);
+ setOptionValue(configuration, PULSAR_LISTENER_NAME, builder::listenerName);
+ builder.authentication(createAuthentication(configuration));
+ setOptionValue(
+ configuration,
+ PULSAR_OPERATION_TIMEOUT_MS,
+ Math::toIntExact,
+ timeout -> builder.operationTimeout(timeout, MILLISECONDS));
+ setOptionValue(configuration, PULSAR_NUM_IO_THREADS, builder::ioThreads);
+ setOptionValue(configuration, PULSAR_NUM_LISTENER_THREADS, builder::listenerThreads);
+ setOptionValue(configuration, PULSAR_CONNECTIONS_PER_BROKER, builder::connectionsPerBroker);
+ setOptionValue(configuration, PULSAR_USE_TCP_NO_DELAY, builder::enableTcpNoDelay);
+ setOptionValue(
+ configuration, PULSAR_TLS_TRUST_CERTS_FILE_PATH, builder::tlsTrustCertsFilePath);
+ setOptionValue(
+ configuration,
+ PULSAR_TLS_ALLOW_INSECURE_CONNECTION,
+ builder::allowTlsInsecureConnection);
+ setOptionValue(
+ configuration,
+ PULSAR_TLS_HOSTNAME_VERIFICATION_ENABLE,
+ builder::enableTlsHostnameVerification);
+ setOptionValue(configuration, PULSAR_USE_KEY_STORE_TLS, builder::useKeyStoreTls);
+ setOptionValue(configuration, PULSAR_SSL_PROVIDER, builder::sslProvider);
+ setOptionValue(configuration, PULSAR_TLS_TRUST_STORE_TYPE, builder::tlsTrustStoreType);
+ setOptionValue(configuration, PULSAR_TLS_TRUST_STORE_PATH, builder::tlsTrustStorePath);
+ setOptionValue(
+ configuration, PULSAR_TLS_TRUST_STORE_PASSWORD, builder::tlsTrustStorePassword);
+ setOptionValue(
+ configuration,
+ PULSAR_TLS_CIPHERS,
+ v -> PulsarJsonUtils.toSet(String.class, v),
+ builder::tlsCiphers);
+ setOptionValue(
+ configuration,
+ PULSAR_TLS_PROTOCOLS,
+ v -> PulsarJsonUtils.toSet(String.class, v),
+ builder::tlsProtocols);
+ setOptionValue(
+ configuration,
+ PULSAR_MEMORY_LIMIT_BYTES,
+ bytes -> builder.memoryLimit(bytes, BYTES));
+ setOptionValue(
+ configuration,
+ PULSAR_STATS_INTERVAL_SECONDS,
+ v -> builder.statsInterval(v, SECONDS));
+ setOptionValue(
+ configuration,
+ PULSAR_CONCURRENT_LOOKUP_REQUEST,
+ builder::maxConcurrentLookupRequests);
+ setOptionValue(configuration, PULSAR_MAX_LOOKUP_REQUEST, builder::maxLookupRequests);
+ setOptionValue(configuration, PULSAR_MAX_LOOKUP_REDIRECTS, builder::maxLookupRedirects);
+ setOptionValue(
+ configuration,
+ PULSAR_MAX_NUMBER_OF_REJECTED_REQUEST_PER_CONNECTION,
+ builder::maxNumberOfRejectedRequestPerConnection);
+ setOptionValue(
+ configuration,
+ PULSAR_KEEP_ALIVE_INTERVAL_SECONDS,
+ v -> builder.keepAliveInterval(v, SECONDS));
+ setOptionValue(
+ configuration,
+ PULSAR_CONNECTION_TIMEOUT_MS,
+ v -> builder.connectionTimeout(v, MILLISECONDS));
+ setOptionValue(
+ configuration,
+ PULSAR_INITIAL_BACKOFF_INTERVAL_NANOS,
+ v -> builder.startingBackoffInterval(v, NANOSECONDS));
+ setOptionValue(
+ configuration,
+ PULSAR_MAX_BACKOFF_INTERVAL_NANOS,
+ v -> builder.maxBackoffInterval(v, NANOSECONDS));
+ setOptionValue(configuration, PULSAR_ENABLE_BUSY_WAIT, builder::enableBusyWait);
+ if (configuration.contains(PULSAR_PROXY_SERVICE_URL)) {
+ String proxyServiceUrl = configuration.get(PULSAR_PROXY_SERVICE_URL);
+ ProxyProtocol proxyProtocol = configuration.get(PULSAR_PROXY_PROTOCOL);
+ builder.proxyServiceUrl(proxyServiceUrl, proxyProtocol);
+ }
+ setOptionValue(configuration, PULSAR_ENABLE_TRANSACTION, builder::enableTransaction);
+
+ return sneakyClient(builder::build);
+ }
+
/**
- * Create a common {@link ClientConfigurationData} for both pulsar source and sink connector by
- * using the given {@link Configuration}. Add the configuration was listed in {@link
- * PulsarSourceOptions}, and we would parse it one by one.
- *
- * @param configuration The flink configuration
+ * PulsarAdmin shares almost the same configuration with PulsarClient, but we separate this
+ * create method for directly create it.
*/
- public static ClientConfigurationData createClientConfig(
- Configuration configuration,
- @Nullable ConfigurationDataCustomizer<ClientConfigurationData> customizer) {
- ClientConfigurationData data = new ClientConfigurationData();
+ public static PulsarAdmin createAdmin(Configuration configuration) {
+ PulsarAdminBuilder builder = PulsarAdmin.builder();
- // Set the properties one by one.
- data.setServiceUrl(configuration.get(PULSAR_SERVICE_URL));
- data.setAuthPluginClassName(configuration.get(PULSAR_AUTH_PLUGIN_CLASS_NAME));
- data.setAuthParams(configuration.get(PULSAR_AUTH_PARAMS));
- data.setAuthParamMap(
- getOptionValue(
- configuration,
- PULSAR_AUTH_PARAM_MAP,
- v -> PulsarJsonUtils.toMap(String.class, String.class, v)));
- data.setOperationTimeoutMs(configuration.get(PULSAR_OPERATION_TIMEOUT_MS));
- data.setStatsIntervalSeconds(configuration.get(PULSAR_STATS_INTERVAL_SECONDS));
- data.setNumIoThreads(configuration.get(PULSAR_NUM_IO_THREADS));
- data.setNumListenerThreads(configuration.get(PULSAR_NUM_LISTENER_THREADS));
- data.setConnectionsPerBroker(configuration.get(PULSAR_CONNECTIONS_PER_BROKER));
- data.setUseTcpNoDelay(configuration.get(PULSAR_USE_TCP_NO_DELAY));
- data.setUseTls(configuration.get(PULSAR_USE_TLS));
- data.setTlsTrustCertsFilePath(configuration.get(PULSAR_TLS_TRUST_CERTS_FILE_PATH));
- data.setTlsAllowInsecureConnection(configuration.get(PULSAR_TLS_ALLOW_INSECURE_CONNECTION));
- data.setTlsHostnameVerificationEnable(
- configuration.get(PULSAR_TLS_HOSTNAME_VERIFICATION_ENABLE));
- data.setConcurrentLookupRequest(configuration.get(PULSAR_CONCURRENT_LOOKUP_REQUEST));
- data.setMaxLookupRequest(configuration.get(PULSAR_MAX_LOOKUP_REQUEST));
- data.setMaxLookupRedirects(configuration.get(PULSAR_MAX_LOOKUP_REDIRECTS));
- data.setMaxNumberOfRejectedRequestPerConnection(
- configuration.get(PULSAR_MAX_NUMBER_OF_REJECTED_REQUEST_PER_CONNECTION));
- data.setKeepAliveIntervalSeconds(configuration.get(PULSAR_KEEP_ALIVE_INTERVAL_SECONDS));
- data.setConnectionTimeoutMs(configuration.get(PULSAR_CONNECTION_TIMEOUT_MS));
- data.setRequestTimeoutMs(configuration.get(PULSAR_REQUEST_TIMEOUT_MS));
- data.setInitialBackoffIntervalNanos(
- configuration.get(PULSAR_INITIAL_BACKOFF_INTERVAL_NANOS));
- data.setMaxBackoffIntervalNanos(configuration.get(PULSAR_MAX_BACKOFF_INTERVAL_NANOS));
- data.setEnableBusyWait(configuration.get(PULSAR_ENABLE_BUSY_WAIT));
- data.setListenerName(configuration.get(PULSAR_LISTENER_NAME));
- data.setUseKeyStoreTls(configuration.get(PULSAR_USE_KEY_STORE_TLS));
- data.setSslProvider(configuration.get(PULSAR_SSL_PROVIDER));
- data.setTlsTrustStoreType(configuration.get(PULSAR_TLS_TRUST_STORE_TYPE));
- data.setTlsTrustStorePath(configuration.get(PULSAR_TLS_TRUST_STORE_PATH));
- data.setTlsTrustStorePassword(configuration.get(PULSAR_TLS_TRUST_STORE_PASSWORD));
- data.setTlsCiphers(
- getOptionValue(
- configuration,
- PULSAR_TLS_CIPHERS,
- v -> PulsarJsonUtils.toSet(String.class, v)));
- data.setTlsProtocols(
- getOptionValue(
- configuration,
- PULSAR_TLS_PROTOCOLS,
- v -> PulsarJsonUtils.toSet(String.class, v)));
- data.setMemoryLimitBytes(configuration.get(PULSAR_MEMORY_LIMIT_BYTES));
- data.setProxyServiceUrl(configuration.get(PULSAR_PROXY_SERVICE_URL));
- data.setProxyProtocol(configuration.get(PULSAR_PROXY_PROTOCOL));
- data.setEnableTransaction(configuration.get(PULSAR_ENABLE_TRANSACTION));
- data.setAuthentication(createAuthentication(configuration));
-
- if (customizer != null) {
- customizer.customize(data);
- }
+ setOptionValue(configuration, PULSAR_ADMIN_URL, builder::serviceHttpUrl);
+ builder.authentication(createAuthentication(configuration));
+ setOptionValue(
+ configuration, PULSAR_TLS_TRUST_CERTS_FILE_PATH, builder::tlsTrustCertsFilePath);
+ setOptionValue(
+ configuration,
+ PULSAR_TLS_ALLOW_INSECURE_CONNECTION,
+ builder::allowTlsInsecureConnection);
+ setOptionValue(
+ configuration,
+ PULSAR_TLS_HOSTNAME_VERIFICATION_ENABLE,
+ builder::enableTlsHostnameVerification);
+ setOptionValue(configuration, PULSAR_USE_KEY_STORE_TLS, builder::useKeyStoreTls);
+ setOptionValue(configuration, PULSAR_SSL_PROVIDER, builder::sslProvider);
+ setOptionValue(configuration, PULSAR_TLS_TRUST_STORE_TYPE, builder::tlsTrustStoreType);
+ setOptionValue(configuration, PULSAR_TLS_TRUST_STORE_PATH, builder::tlsTrustStorePath);
+ setOptionValue(
+ configuration, PULSAR_TLS_TRUST_STORE_PASSWORD, builder::tlsTrustStorePassword);
+ setOptionValue(
+ configuration,
+ PULSAR_TLS_CIPHERS,
+ v -> PulsarJsonUtils.toSet(String.class, v),
+ builder::tlsCiphers);
+ setOptionValue(
+ configuration,
+ PULSAR_TLS_PROTOCOLS,
+ v -> PulsarJsonUtils.toSet(String.class, v),
+ builder::tlsProtocols);
+ setOptionValue(
+ configuration,
+ PULSAR_CONNECT_TIMEOUT,
+ Math::toIntExact,
+ v -> builder.connectionTimeout(v, MILLISECONDS));
+ setOptionValue(
+ configuration,
+ PULSAR_READ_TIMEOUT,
+ Math::toIntExact,
+ v -> builder.readTimeout(v, MILLISECONDS));
+ setOptionValue(
+ configuration,
+ PULSAR_REQUEST_TIMEOUT,
+ Math::toIntExact,
+ v -> builder.requestTimeout(v, MILLISECONDS));
+ setOptionValue(
+ configuration,
+ PULSAR_AUTO_CERT_REFRESH_TIME,
+ Math::toIntExact,
+ v -> builder.autoCertRefreshTime(v, MILLISECONDS));
- return data;
+ return sneakyClient(builder::build);
}
/**
@@ -197,7 +269,6 @@ public final class PulsarConfigUtils {
public static <F, T> T getOptionValue(
Configuration configuration, ConfigOption<F> option, Function<F, T> convertor) {
F value = configuration.get(option);
-
if (value != null) {
return convertor.apply(value);
} else {
@@ -205,75 +276,23 @@ public final class PulsarConfigUtils {
}
}
- /** Create a PulsarClient by using the flink Configuration and the config customizer. */
- public static PulsarClient createClient(
- Configuration configuration,
- ConfigurationDataCustomizer<ClientConfigurationData> customizer) {
- return createClient(createClientConfig(configuration, customizer));
+ /** Set the config option's value to a given builder. */
+ public static <T> void setOptionValue(
+ Configuration configuration, ConfigOption<T> option, Consumer<T> setter) {
+ setOptionValue(configuration, option, identity(), setter);
}
/**
- * Create a PulsarClient by using the given config, all the config could be directly loaded
- * except the authentication.
+ * Query the config option's value, convert it into a required type, set it to a given builder.
*/
- public static PulsarClient createClient(ClientConfigurationData config) {
- // PulsarClientBuilder don't support using the given ClientConfigurationData directly.
- Map<String, Object> configMap = configMap(config);
-
- // These two auth config would be serialized to a secret information by default.
- configMap.put("authParamMap", config.getAuthParamMap());
- configMap.put("authParams", config.getAuthParams());
-
- ClientBuilder clientBuilder = PulsarClient.builder().loadConf(configMap);
-
- // Set some non-serializable fields.
- if (config.getAuthentication() != null) {
- clientBuilder.authentication(config.getAuthentication());
- }
- if (config.getClock() != null) {
- clientBuilder.clock(config.getClock());
- }
- if (config.getServiceUrlProvider() != null) {
- clientBuilder.serviceUrlProvider(config.getServiceUrlProvider());
- }
-
- return sneakyClient(clientBuilder::build);
- }
-
- /**
- * PulsarAdmin shares almost the same configuration with PulsarClient, but we separate this
- * create method for directly create it.
- */
- public static PulsarAdmin createAdmin(
+ public static <T, V> void setOptionValue(
Configuration configuration,
- ConfigurationDataCustomizer<ClientConfigurationData> customizer) {
- ClientConfigurationData clientConfig = createClientConfig(configuration, customizer);
- String adminUrl = configuration.get(PULSAR_ADMIN_URL);
- int connectionTimeout = Math.toIntExact(configuration.get(PULSAR_CONNECT_TIMEOUT));
- int readTimeout = Math.toIntExact(configuration.get(PULSAR_READ_TIMEOUT));
- int requestTimeout = Math.toIntExact(configuration.get(PULSAR_REQUEST_TIMEOUT));
- int autoCertRefreshTime = Math.toIntExact(configuration.get(PULSAR_AUTO_CERT_REFRESH_TIME));
-
- PulsarAdminBuilder builder =
- PulsarAdmin.builder()
- .serviceHttpUrl(adminUrl)
- .authentication(clientConfig.getAuthentication())
- .tlsTrustCertsFilePath(clientConfig.getTlsTrustCertsFilePath())
- .allowTlsInsecureConnection(clientConfig.isTlsAllowInsecureConnection())
- .enableTlsHostnameVerification(
- clientConfig.isTlsHostnameVerificationEnable())
- .useKeyStoreTls(clientConfig.isUseKeyStoreTls())
- .sslProvider(clientConfig.getSslProvider())
- .tlsTrustStoreType(clientConfig.getTlsTrustStoreType())
- .tlsTrustStorePath(clientConfig.getTlsTrustStorePath())
- .tlsTrustStorePassword(clientConfig.getTlsTrustStorePassword())
- .tlsCiphers(clientConfig.getTlsCiphers())
- .tlsProtocols(clientConfig.getTlsProtocols())
- .connectionTimeout(connectionTimeout, MILLISECONDS)
- .readTimeout(readTimeout, MILLISECONDS)
- .requestTimeout(requestTimeout, MILLISECONDS)
- .autoCertRefreshTime(autoCertRefreshTime, MILLISECONDS);
-
- return sneakyClient(builder::build);
+ ConfigOption<T> option,
+ Function<T, V> convertor,
+ Consumer<V> setter) {
+ if (configuration.contains(option)) {
+ V value = getOptionValue(configuration, option, convertor);
+ setter.accept(value);
+ }
}
}
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/utils/PulsarJsonUtils.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/utils/PulsarJsonUtils.java
index 27587cb..64308bc 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/utils/PulsarJsonUtils.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/utils/PulsarJsonUtils.java
@@ -20,14 +20,12 @@ package org.apache.flink.connector.pulsar.common.utils;
import org.apache.flink.annotation.Internal;
-import org.apache.pulsar.client.api.ConsumerBuilder;
import org.apache.pulsar.shade.com.fasterxml.jackson.core.JsonProcessingException;
import org.apache.pulsar.shade.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.pulsar.shade.com.fasterxml.jackson.databind.type.CollectionType;
import org.apache.pulsar.shade.com.fasterxml.jackson.databind.type.MapType;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -97,17 +95,4 @@ public final class PulsarJsonUtils {
throw new IllegalArgumentException(e);
}
}
-
- /**
- * Pulsar's consumer builder use {@link ConsumerBuilder#loadConf(Map)} for user to customize
- * their conf. We have to convert the configure instance into a common config map.
- */
- public static Map<String, Object> configMap(Object config) {
- try {
- String json = mapper.writeValueAsString(config);
- return toMap(String.class, Object.class, json);
- } catch (JsonProcessingException e) {
- return Collections.emptyMap();
- }
- }
}
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSource.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSource.java
index 0c6590d..7b0e4bf 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSource.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSource.java
@@ -28,7 +28,6 @@ import org.apache.flink.api.connector.source.SplitEnumerator;
import org.apache.flink.api.connector.source.SplitEnumeratorContext;
import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer;
import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
import org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumState;
import org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumStateSerializer;
@@ -45,9 +44,6 @@ import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplitSerializer;
import org.apache.flink.core.io.SimpleVersionedSerializer;
-import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
-import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
-
/**
* The Source implementation of Pulsar. Please use a {@link PulsarSourceBuilder} to construct a
* {@link PulsarSource}. The following example shows how to create a PulsarSource emitting records
@@ -96,14 +92,6 @@ public final class PulsarSource<OUT>
/** The pulsar deserialization schema used for deserializing message. */
private final PulsarDeserializationSchema<OUT> deserializationSchema;
- /** Modify the flink generated {@link ClientConfigurationData}. */
- private final ConfigurationDataCustomizer<ClientConfigurationData>
- clientConfigurationCustomizer;
-
- /** Modify the flink generated {@link ConsumerConfigurationData}. */
- private final ConfigurationDataCustomizer<ConsumerConfigurationData<byte[]>>
- consumerConfigurationCustomizer;
-
/**
* The constructor for PulsarSource, it's package protected for forcing using {@link
* PulsarSourceBuilder}.
@@ -115,10 +103,7 @@ public final class PulsarSource<OUT>
StartCursor startCursor,
StopCursor stopCursor,
Boundedness boundedness,
- PulsarDeserializationSchema<OUT> deserializationSchema,
- ConfigurationDataCustomizer<ClientConfigurationData> clientConfigurationCustomizer,
- ConfigurationDataCustomizer<ConsumerConfigurationData<byte[]>>
- consumerConfigurationCustomizer) {
+ PulsarDeserializationSchema<OUT> deserializationSchema) {
this.configuration = configuration;
this.sourceConfiguration = new SourceConfiguration(configuration);
@@ -128,8 +113,6 @@ public final class PulsarSource<OUT>
this.stopCursor = stopCursor;
this.boundedness = boundedness;
this.deserializationSchema = deserializationSchema;
- this.clientConfigurationCustomizer = clientConfigurationCustomizer;
- this.consumerConfigurationCustomizer = consumerConfigurationCustomizer;
}
/**
@@ -155,12 +138,7 @@ public final class PulsarSource<OUT>
new PulsarDeserializationSchemaInitializationContext(readerContext));
return PulsarSourceReaderFactory.create(
- readerContext,
- deserializationSchema,
- configuration,
- sourceConfiguration,
- clientConfigurationCustomizer,
- consumerConfigurationCustomizer);
+ readerContext, deserializationSchema, configuration, sourceConfiguration);
}
@Override
@@ -173,7 +151,6 @@ public final class PulsarSource<OUT>
rangeGenerator,
configuration,
sourceConfiguration,
- clientConfigurationCustomizer,
enumContext,
assignmentState);
}
@@ -189,7 +166,6 @@ public final class PulsarSource<OUT>
rangeGenerator,
configuration,
sourceConfiguration,
- clientConfigurationCustomizer,
enumContext,
assignmentState);
}
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
index e8294a5..50d93fa 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
@@ -21,12 +21,10 @@ package org.apache.flink.connector.pulsar.source;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.connector.source.Boundedness;
-import org.apache.flink.api.java.ClosureCleaner;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.UnmodifiableConfiguration;
-import org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer;
import org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils;
import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
@@ -40,8 +38,6 @@ import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeseri
import org.apache.pulsar.client.api.RegexSubscriptionMode;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionType;
-import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
-import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -49,11 +45,9 @@ import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.Properties;
-import java.util.function.Consumer;
import java.util.regex.Pattern;
import static java.lang.Boolean.FALSE;
-import static org.apache.flink.api.common.ExecutionConfig.ClosureCleanerLevel.RECURSIVE;
import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.getOptionValue;
import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL;
import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ENABLE_TRANSACTION;
@@ -123,18 +117,16 @@ public final class PulsarSourceBuilder<OUT> {
private final Configuration configuration;
private PulsarSubscriber subscriber;
private RangeGenerator rangeGenerator;
- private StartCursor startCursor = StartCursor.defaultStartCursor();
- private StopCursor stopCursor = StopCursor.defaultStopCursor();
+ private StartCursor startCursor;
+ private StopCursor stopCursor;
private Boundedness boundedness;
private PulsarDeserializationSchema<OUT> deserializationSchema;
- private ConfigurationDataCustomizer<ClientConfigurationData> clientConfigurationCustomizer;
- private ConfigurationDataCustomizer<ConsumerConfigurationData<byte[]>>
- consumerConfigurationCustomizer;
// private builder constructor.
PulsarSourceBuilder() {
- // The default configuration holder.
this.configuration = new Configuration();
+ this.startCursor = StartCursor.defaultStartCursor();
+ this.stopCursor = StopCursor.defaultStopCursor();
}
/**
@@ -369,41 +361,6 @@ public final class PulsarSourceBuilder<OUT> {
}
/**
- * {@link ClientConfigurationData} is created from {@link Properties}, but you can just using
- * its setter method for programmatic update a config instance.
- *
- * @param consumer Modify the ClientConfigurationData instance in this consumer.
- * @return this PulsarSourceBuilder.
- */
- public PulsarSourceBuilder<OUT> modifyClientConfig(Consumer<ClientConfigurationData> consumer) {
- if (clientConfigurationCustomizer == null) {
- this.clientConfigurationCustomizer = consumer::accept;
- } else {
- this.clientConfigurationCustomizer =
- clientConfigurationCustomizer.compose(consumer::accept);
- }
- return this;
- }
-
- /**
- * {@link ConsumerConfigurationData} is created from {@link Properties}, but you can just using
- * its setter method for programmatic update a config instance.
- *
- * @param consumer Modify the ConsumerConfigurationData instance in this consumer.
- * @return this PulsarSourceBuilder.
- */
- public PulsarSourceBuilder<OUT> modifyConsumerConfig(
- Consumer<ConsumerConfigurationData<byte[]>> consumer) {
- if (consumerConfigurationCustomizer == null) {
- this.consumerConfigurationCustomizer = consumer::accept;
- } else {
- this.consumerConfigurationCustomizer =
- consumerConfigurationCustomizer.compose(consumer::accept);
- }
- return this;
- }
-
- /**
* Set an arbitrary property for the PulsarSource and PulsarConsumer. The valid keys can be
* found in {@link PulsarSourceOptions}.
*
@@ -531,14 +488,6 @@ public final class PulsarSourceBuilder<OUT> {
checkNotNull(deserializationSchema, "deserializationSchema should be set.");
- if (clientConfigurationCustomizer == null) {
- this.clientConfigurationCustomizer = ConfigurationDataCustomizer.blankCustomizer();
- }
-
- if (consumerConfigurationCustomizer == null) {
- this.consumerConfigurationCustomizer = ConfigurationDataCustomizer.blankCustomizer();
- }
-
// Enable transaction if the cursor auto commit is disabled for Key_Shared & Shared.
if (FALSE.equals(configuration.get(PULSAR_ENABLE_AUTO_ACKNOWLEDGE_MESSAGE))
&& (subscriptionType == SubscriptionType.Key_Shared
@@ -564,8 +513,6 @@ public final class PulsarSourceBuilder<OUT> {
// Since these implementation could be a lambda, make sure they are serializable.
checkState(isSerializable(startCursor), "StartCursor isn't serializable");
checkState(isSerializable(stopCursor), "StopCursor isn't serializable");
- ClosureCleaner.clean(clientConfigurationCustomizer, RECURSIVE, true);
- ClosureCleaner.clean(consumerConfigurationCustomizer, RECURSIVE, true);
// Make the configuration unmodifiable.
UnmodifiableConfiguration config = new UnmodifiableConfiguration(configuration);
@@ -577,9 +524,7 @@ public final class PulsarSourceBuilder<OUT> {
startCursor,
stopCursor,
boundedness,
- deserializationSchema,
- clientConfigurationCustomizer,
- consumerConfigurationCustomizer);
+ deserializationSchema);
}
// ------------- private helpers --------------
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/PulsarSourceConfigUtils.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/PulsarSourceConfigUtils.java
index 6687bbd..a471bc6 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/PulsarSourceConfigUtils.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/PulsarSourceConfigUtils.java
@@ -21,9 +21,7 @@ package org.apache.flink.connector.pulsar.source.config;
import org.apache.flink.annotation.Internal;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer;
import org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils;
-import org.apache.flink.connector.pulsar.source.PulsarSourceOptions;
import org.apache.flink.util.Preconditions;
import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList;
@@ -33,31 +31,26 @@ import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.ConsumerBuilder;
import org.apache.pulsar.client.api.DeadLetterPolicy;
import org.apache.pulsar.client.api.PulsarClient;
-import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
-import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
-
-import javax.annotation.Nullable;
import java.util.List;
+import java.util.Optional;
import java.util.Set;
-import java.util.SortedMap;
-import java.util.TreeMap;
-import java.util.regex.Pattern;
-import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.getOptionValue;
+import static java.util.concurrent.TimeUnit.MICROSECONDS;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.setOptionValue;
import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL;
import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTH_PARAMS;
import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTH_PARAM_MAP;
import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SERVICE_URL;
-import static org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils.configMap;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_ACKNOWLEDGEMENTS_GROUP_TIME_MICROS;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_ACK_RECEIPT_ENABLED;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_ACK_TIMEOUT_MILLIS;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_AUTO_ACK_OLDEST_CHUNKED_MESSAGE_ON_QUEUE_FULL;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_AUTO_UPDATE_PARTITIONS;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_AUTO_UPDATE_PARTITIONS_INTERVAL_SECONDS;
-import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_BATCH_INDEX_ACK_ENABLED;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_CONSUMER_NAME;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_CONSUMER_PROPERTIES;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_CRYPTO_FAILURE_ACTION;
@@ -67,12 +60,10 @@ import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSA
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_MAX_REDELIVER_COUNT;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_MAX_TOTAL_RECEIVER_QUEUE_SIZE_ACROSS_PARTITIONS;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_NEGATIVE_ACK_REDELIVERY_DELAY_MICROS;
-import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_PATTERN_AUTO_DISCOVERY_PERIOD;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_POOL_MESSAGES;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_PRIORITY_LEVEL;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_READ_COMPACTED;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_RECEIVER_QUEUE_SIZE;
-import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_REGEX_SUBSCRIPTION_MODE;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_REPLICATE_SUBSCRIPTION_STATE;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_RESET_INCLUDE_HEAD;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_RETRY_ENABLE;
@@ -82,8 +73,6 @@ import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSA
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_NAME;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_TYPE;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TICK_DURATION_MILLIS;
-import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPICS_PATTERN;
-import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPIC_NAMES;
/** Create source related {@link Consumer} and validate config. */
@Internal
@@ -96,7 +85,6 @@ public final class PulsarSourceConfigUtils {
private static final List<Set<ConfigOption<?>>> CONFLICT_SOURCE_OPTIONS =
ImmutableList.<Set<ConfigOption<?>>>builder()
.add(ImmutableSet.of(PULSAR_AUTH_PARAMS, PULSAR_AUTH_PARAM_MAP))
- .add(ImmutableSet.of(PULSAR_TOPIC_NAMES, PULSAR_TOPICS_PATTERN))
.build();
private static final Set<ConfigOption<?>> REQUIRED_SOURCE_OPTIONS =
@@ -132,127 +120,92 @@ public final class PulsarSourceConfigUtils {
});
}
- /**
- * Create a {@link ConsumerConfigurationData} by using the given {@link Configuration}. Add the
- * configuration was listed in {@link PulsarSourceOptions}, and we would parse it one by one.
- *
- * @param configuration The flink configuration
- * @param <T> The type of the pulsar message, this type is ignored.
- */
- public static <T> ConsumerConfigurationData<T> createConsumerConfig(
- Configuration configuration,
- @Nullable ConfigurationDataCustomizer<ConsumerConfigurationData<T>> customizer) {
- ConsumerConfigurationData<T> data = new ConsumerConfigurationData<>();
-
- // Set the properties one by one.
- data.setTopicNames(
- getOptionValue(
- configuration,
- PULSAR_TOPIC_NAMES,
- v -> PulsarJsonUtils.toSet(String.class, v)));
- data.setTopicsPattern(
- getOptionValue(configuration, PULSAR_TOPICS_PATTERN, Pattern::compile));
- data.setSubscriptionName(configuration.get(PULSAR_SUBSCRIPTION_NAME));
- data.setSubscriptionType(configuration.get(PULSAR_SUBSCRIPTION_TYPE));
-
- data.setSubscriptionMode(configuration.get(PULSAR_SUBSCRIPTION_MODE));
- data.setReceiverQueueSize(configuration.get(PULSAR_RECEIVER_QUEUE_SIZE));
- data.setAcknowledgementsGroupTimeMicros(
- configuration.get(PULSAR_ACKNOWLEDGEMENTS_GROUP_TIME_MICROS));
- data.setNegativeAckRedeliveryDelayMicros(
- configuration.get(PULSAR_NEGATIVE_ACK_REDELIVERY_DELAY_MICROS));
- data.setMaxTotalReceiverQueueSizeAcrossPartitions(
- configuration.get(PULSAR_MAX_TOTAL_RECEIVER_QUEUE_SIZE_ACROSS_PARTITIONS));
- data.setConsumerName(configuration.get(PULSAR_CONSUMER_NAME));
- data.setAckTimeoutMillis(configuration.get(PULSAR_ACK_TIMEOUT_MILLIS));
- data.setTickDurationMillis(configuration.get(PULSAR_TICK_DURATION_MILLIS));
- data.setPriorityLevel(configuration.get(PULSAR_PRIORITY_LEVEL));
-
- // This method is deprecated in 2.8.0, we use this method for backward compatible.
- data.setMaxPendingChunkedMessage(configuration.get(PULSAR_MAX_PENDING_CHUNKED_MESSAGE));
- data.setAutoAckOldestChunkedMessageOnQueueFull(
- configuration.get(PULSAR_AUTO_ACK_OLDEST_CHUNKED_MESSAGE_ON_QUEUE_FULL));
- data.setExpireTimeOfIncompleteChunkedMessageMillis(
- configuration.get(PULSAR_EXPIRE_TIME_OF_INCOMPLETE_CHUNKED_MESSAGE_MILLIS));
- data.setCryptoFailureAction(configuration.get(PULSAR_CRYPTO_FAILURE_ACTION));
- data.setProperties(
- getOptionValue(
- configuration,
- PULSAR_CONSUMER_PROPERTIES,
- v ->
- (SortedMap<String, String>)
- PulsarJsonUtils.toMap(
- TreeMap.class, String.class, String.class, v)));
- data.setReadCompacted(configuration.get(PULSAR_READ_COMPACTED));
- data.setSubscriptionInitialPosition(
- configuration.get(PULSAR_SUBSCRIPTION_INITIAL_POSITION));
- data.setPatternAutoDiscoveryPeriod(configuration.get(PULSAR_PATTERN_AUTO_DISCOVERY_PERIOD));
- data.setRegexSubscriptionMode(configuration.get(PULSAR_REGEX_SUBSCRIPTION_MODE));
+ /** Create a pulsar consumer builder by using the given Configuration. */
+ public static <T> ConsumerBuilder<T> createConsumerBuilder(
+ PulsarClient client, Schema<T> schema, Configuration configuration) {
+ ConsumerBuilder<T> builder = client.newConsumer(schema);
+
+ setOptionValue(configuration, PULSAR_SUBSCRIPTION_NAME, builder::subscriptionName);
+ setOptionValue(
+ configuration, PULSAR_ACK_TIMEOUT_MILLIS, v -> builder.ackTimeout(v, MILLISECONDS));
+ setOptionValue(configuration, PULSAR_ACK_RECEIPT_ENABLED, builder::isAckReceiptEnabled);
+ setOptionValue(
+ configuration,
+ PULSAR_TICK_DURATION_MILLIS,
+ v -> builder.ackTimeoutTickTime(v, MILLISECONDS));
+ setOptionValue(
+ configuration,
+ PULSAR_NEGATIVE_ACK_REDELIVERY_DELAY_MICROS,
+ v -> builder.negativeAckRedeliveryDelay(v, MICROSECONDS));
+ setOptionValue(configuration, PULSAR_SUBSCRIPTION_TYPE, builder::subscriptionType);
+ setOptionValue(configuration, PULSAR_SUBSCRIPTION_MODE, builder::subscriptionMode);
+ setOptionValue(configuration, PULSAR_CRYPTO_FAILURE_ACTION, builder::cryptoFailureAction);
+ setOptionValue(configuration, PULSAR_RECEIVER_QUEUE_SIZE, builder::receiverQueueSize);
+ setOptionValue(
+ configuration,
+ PULSAR_ACKNOWLEDGEMENTS_GROUP_TIME_MICROS,
+ v -> builder.acknowledgmentGroupTime(v, MICROSECONDS));
+ setOptionValue(
+ configuration,
+ PULSAR_REPLICATE_SUBSCRIPTION_STATE,
+ builder::replicateSubscriptionState);
+ setOptionValue(
+ configuration,
+ PULSAR_MAX_TOTAL_RECEIVER_QUEUE_SIZE_ACROSS_PARTITIONS,
+ builder::maxTotalReceiverQueueSizeAcrossPartitions);
+ setOptionValue(configuration, PULSAR_CONSUMER_NAME, builder::consumerName);
+ setOptionValue(configuration, PULSAR_READ_COMPACTED, builder::readCompacted);
+ setOptionValue(configuration, PULSAR_PRIORITY_LEVEL, builder::priorityLevel);
+ setOptionValue(
+ configuration,
+ PULSAR_CONSUMER_PROPERTIES,
+ v -> PulsarJsonUtils.toMap(String.class, String.class, v),
+ builder::properties);
+ setOptionValue(
+ configuration,
+ PULSAR_SUBSCRIPTION_INITIAL_POSITION,
+ builder::subscriptionInitialPosition);
+ createDeadLetterPolicy(configuration).ifPresent(builder::deadLetterPolicy);
+ setOptionValue(configuration, PULSAR_AUTO_UPDATE_PARTITIONS, builder::autoUpdatePartitions);
+ setOptionValue(
+ configuration,
+ PULSAR_AUTO_UPDATE_PARTITIONS_INTERVAL_SECONDS,
+ Math::toIntExact,
+ v -> builder.autoUpdatePartitionsInterval(v, SECONDS));
+ if (configuration.getOptional(PULSAR_RESET_INCLUDE_HEAD).orElse(false)) {
+ builder.startMessageIdInclusive();
+ }
+ setOptionValue(configuration, PULSAR_RETRY_ENABLE, builder::enableRetry);
+ setOptionValue(
+ configuration,
+ PULSAR_MAX_PENDING_CHUNKED_MESSAGE,
+ builder::maxPendingChunkedMessage);
+ setOptionValue(
+ configuration,
+ PULSAR_AUTO_ACK_OLDEST_CHUNKED_MESSAGE_ON_QUEUE_FULL,
+ builder::autoAckOldestChunkedMessageOnQueueFull);
+ setOptionValue(
+ configuration,
+ PULSAR_EXPIRE_TIME_OF_INCOMPLETE_CHUNKED_MESSAGE_MILLIS,
+ v -> builder.expireTimeOfIncompleteChunkedMessage(v, MILLISECONDS));
+ setOptionValue(configuration, PULSAR_POOL_MESSAGES, builder::poolMessages);
+
+ return builder;
+ }
- // Set dead letter policy
+ private static Optional<DeadLetterPolicy> createDeadLetterPolicy(Configuration configuration) {
if (configuration.contains(PULSAR_MAX_REDELIVER_COUNT)
|| configuration.contains(PULSAR_RETRY_LETTER_TOPIC)
|| configuration.contains(PULSAR_DEAD_LETTER_TOPIC)) {
- DeadLetterPolicy deadLetterPolicy =
- DeadLetterPolicy.builder()
- .maxRedeliverCount(configuration.get(PULSAR_MAX_REDELIVER_COUNT))
- .retryLetterTopic(configuration.get(PULSAR_RETRY_LETTER_TOPIC))
- .deadLetterTopic(configuration.get(PULSAR_DEAD_LETTER_TOPIC))
- .build();
- data.setDeadLetterPolicy(deadLetterPolicy);
- }
-
- data.setRetryEnable(configuration.get(PULSAR_RETRY_ENABLE));
- data.setAutoUpdatePartitions(configuration.get(PULSAR_AUTO_UPDATE_PARTITIONS));
- data.setAutoUpdatePartitionsIntervalSeconds(
- configuration.get(PULSAR_AUTO_UPDATE_PARTITIONS_INTERVAL_SECONDS));
- data.setReplicateSubscriptionState(configuration.get(PULSAR_REPLICATE_SUBSCRIPTION_STATE));
- data.setResetIncludeHead(configuration.get(PULSAR_RESET_INCLUDE_HEAD));
- data.setBatchIndexAckEnabled(configuration.get(PULSAR_BATCH_INDEX_ACK_ENABLED));
- data.setAckReceiptEnabled(configuration.get(PULSAR_ACK_RECEIPT_ENABLED));
- data.setPoolMessages(configuration.get(PULSAR_POOL_MESSAGES));
+ DeadLetterPolicy.DeadLetterPolicyBuilder builder = DeadLetterPolicy.builder();
- if (customizer != null) {
- customizer.customize(data);
- }
-
- return data;
- }
-
- /** Create a Pulsar Consumer by using the flink Configuration and the config customizer. */
- public static <T> Consumer<T> createConsumer(
- PulsarClient client,
- Schema<T> schema,
- Configuration configuration,
- ConfigurationDataCustomizer<ConsumerConfigurationData<T>> customizer)
- throws PulsarClientException {
- return createConsumer(client, schema, createConsumerConfig(configuration, customizer));
- }
-
- /** Create a pulsar consumer by using the given ConsumerConfigurationData and PulsarClient. */
- public static <T> Consumer<T> createConsumer(
- PulsarClient client, Schema<T> schema, ConsumerConfigurationData<T> config)
- throws PulsarClientException {
- // ConsumerBuilder don't support using the given ConsumerConfigurationData directly.
- ConsumerBuilder<T> consumerBuilder = client.newConsumer(schema).loadConf(configMap(config));
+ setOptionValue(configuration, PULSAR_MAX_REDELIVER_COUNT, builder::maxRedeliverCount);
+ setOptionValue(configuration, PULSAR_RETRY_LETTER_TOPIC, builder::retryLetterTopic);
+ setOptionValue(configuration, PULSAR_DEAD_LETTER_TOPIC, builder::deadLetterTopic);
- // Set some non-serializable fields.
- if (config.getMessageListener() != null) {
- consumerBuilder.messageListener(config.getMessageListener());
+ return Optional.of(builder.build());
+ } else {
+ return Optional.empty();
}
- if (config.getConsumerEventListener() != null) {
- consumerBuilder.consumerEventListener(config.getConsumerEventListener());
- }
- if (config.getCryptoKeyReader() != null) {
- consumerBuilder.cryptoKeyReader(config.getCryptoKeyReader());
- }
- if (config.getMessageCrypto() != null) {
- consumerBuilder.messageCrypto(config.getMessageCrypto());
- }
- if (config.getBatchReceivePolicy() != null) {
- consumerBuilder.batchReceivePolicy(config.getBatchReceivePolicy());
- }
-
- return consumerBuilder.subscribe();
}
}
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/PulsarSourceEnumerator.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/PulsarSourceEnumerator.java
index bfa025b..2a55e7a 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/PulsarSourceEnumerator.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/PulsarSourceEnumerator.java
@@ -22,7 +22,6 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.api.connector.source.SplitEnumerator;
import org.apache.flink.api.connector.source.SplitEnumeratorContext;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer;
import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
import org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber;
import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
@@ -31,7 +30,6 @@ import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.pulsar.client.admin.PulsarAdmin;
-import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -66,10 +64,9 @@ public class PulsarSourceEnumerator
RangeGenerator rangeGenerator,
Configuration configuration,
SourceConfiguration sourceConfiguration,
- ConfigurationDataCustomizer<ClientConfigurationData> clientConfigurationCustomizer,
SplitEnumeratorContext<PulsarPartitionSplit> context,
SplitsAssignmentState assignmentState) {
- this.pulsarAdmin = createAdmin(configuration, clientConfigurationCustomizer);
+ this.pulsarAdmin = createAdmin(configuration);
this.subscriber = subscriber;
this.rangeGenerator = rangeGenerator;
this.configuration = configuration;
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReaderFactory.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReaderFactory.java
index b15bed0..e60ef99 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReaderFactory.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReaderFactory.java
@@ -24,7 +24,6 @@ import org.apache.flink.api.connector.source.SourceReaderContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
import org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
-import org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer;
import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
import org.apache.flink.connector.pulsar.source.reader.message.PulsarMessage;
@@ -39,8 +38,6 @@ import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.api.transaction.TransactionCoordinatorClient;
import org.apache.pulsar.client.impl.PulsarClientImpl;
-import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
-import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
import java.util.function.Supplier;
@@ -69,13 +66,10 @@ public final class PulsarSourceReaderFactory {
SourceReaderContext readerContext,
PulsarDeserializationSchema<OUT> deserializationSchema,
Configuration configuration,
- SourceConfiguration sourceConfiguration,
- ConfigurationDataCustomizer<ClientConfigurationData> clientConfigurationCustomizer,
- ConfigurationDataCustomizer<ConsumerConfigurationData<byte[]>>
- consumerConfigurationCustomizer) {
+ SourceConfiguration sourceConfiguration) {
- PulsarClient pulsarClient = createClient(configuration, clientConfigurationCustomizer);
- PulsarAdmin pulsarAdmin = createAdmin(configuration, clientConfigurationCustomizer);
+ PulsarClient pulsarClient = createClient(configuration);
+ PulsarAdmin pulsarAdmin = createAdmin(configuration);
// Create a message queue with the predefined source option.
int queueSize = configuration.getInteger(ELEMENT_QUEUE_CAPACITY);
@@ -94,7 +88,6 @@ public final class PulsarSourceReaderFactory {
pulsarAdmin,
configuration,
sourceConfiguration,
- consumerConfigurationCustomizer,
deserializationSchema);
return new PulsarOrderedSourceReader<>(
@@ -121,7 +114,6 @@ public final class PulsarSourceReaderFactory {
pulsarAdmin,
configuration,
sourceConfiguration,
- consumerConfigurationCustomizer,
deserializationSchema,
coordinatorClient);
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarOrderedPartitionSplitReader.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarOrderedPartitionSplitReader.java
index 5f41d49..718966c 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarOrderedPartitionSplitReader.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarOrderedPartitionSplitReader.java
@@ -20,7 +20,6 @@ package org.apache.flink.connector.pulsar.source.reader.split;
import org.apache.flink.annotation.Internal;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer;
import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
@@ -32,7 +31,6 @@ import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.PulsarClient;
-import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -59,16 +57,8 @@ public class PulsarOrderedPartitionSplitReader<OUT> extends PulsarPartitionSplit
PulsarAdmin pulsarAdmin,
Configuration configuration,
SourceConfiguration sourceConfiguration,
- ConfigurationDataCustomizer<ConsumerConfigurationData<byte[]>>
- consumerConfigurationCustomizer,
PulsarDeserializationSchema<OUT> deserializationSchema) {
- super(
- pulsarClient,
- pulsarAdmin,
- configuration,
- sourceConfiguration,
- consumerConfigurationCustomizer,
- deserializationSchema);
+ super(pulsarClient, pulsarAdmin, configuration, sourceConfiguration, deserializationSchema);
}
@Override
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java
index 3d9122b..868a65d 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java
@@ -25,7 +25,6 @@ import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
import org.apache.flink.connector.base.source.reader.splitreader.SplitsAddition;
import org.apache.flink.connector.base.source.reader.splitreader.SplitsChange;
-import org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer;
import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
import org.apache.flink.connector.pulsar.source.enumerator.cursor.CursorPosition;
import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
@@ -39,13 +38,13 @@ import org.apache.flink.util.Preconditions;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.ConsumerBuilder;
import org.apache.pulsar.client.api.KeySharedPolicy;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionType;
-import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -56,10 +55,9 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
-import static java.util.Collections.singleton;
import static org.apache.flink.connector.pulsar.common.utils.PulsarExceptionUtils.sneakyClient;
import static org.apache.flink.connector.pulsar.source.config.CursorVerification.FAIL_ON_MISMATCH;
-import static org.apache.flink.connector.pulsar.source.config.PulsarSourceConfigUtils.createConsumer;
+import static org.apache.flink.connector.pulsar.source.config.PulsarSourceConfigUtils.createConsumerBuilder;
/**
* The common partition split reader.
@@ -74,8 +72,6 @@ abstract class PulsarPartitionSplitReaderBase<OUT>
protected final PulsarAdmin pulsarAdmin;
protected final Configuration configuration;
protected final SourceConfiguration sourceConfiguration;
- protected final ConfigurationDataCustomizer<ConsumerConfigurationData<byte[]>>
- consumerConfigurationCustomizer;
protected final PulsarDeserializationSchema<OUT> deserializationSchema;
protected final AtomicBoolean wakeup;
@@ -87,14 +83,11 @@ abstract class PulsarPartitionSplitReaderBase<OUT>
PulsarAdmin pulsarAdmin,
Configuration configuration,
SourceConfiguration sourceConfiguration,
- ConfigurationDataCustomizer<ConsumerConfigurationData<byte[]>>
- consumerConfigurationCustomizer,
PulsarDeserializationSchema<OUT> deserializationSchema) {
this.pulsarClient = pulsarClient;
this.pulsarAdmin = pulsarAdmin;
this.configuration = configuration;
this.sourceConfiguration = sourceConfiguration;
- this.consumerConfigurationCustomizer = consumerConfigurationCustomizer;
this.deserializationSchema = deserializationSchema;
this.wakeup = new AtomicBoolean(false);
}
@@ -240,25 +233,19 @@ abstract class PulsarPartitionSplitReaderBase<OUT>
/** Create a specified {@link Consumer} by the given topic partition. */
protected Consumer<byte[]> createPulsarConsumer(TopicPartition partition) {
- ConfigurationDataCustomizer<ConsumerConfigurationData<byte[]>> customizer =
- consumerConfigurationCustomizer.compose(
- config -> {
- // Override the configuration by the split information.
- config.setTopicsPattern(null);
- config.setTopicNames(singleton(partition.getFullTopicName()));
-
- // Add KeySharedPolicy for Key_Shared subscription.
- if (sourceConfiguration.getSubscriptionType()
- == SubscriptionType.Key_Shared) {
- KeySharedPolicy policy =
- KeySharedPolicy.stickyHashRange()
- .ranges(partition.getPulsarRange());
- config.setKeySharedPolicy(policy);
- }
- });
+ ConsumerBuilder<byte[]> consumerBuilder =
+ createConsumerBuilder(pulsarClient, Schema.BYTES, configuration);
+
+ consumerBuilder.topic(partition.getFullTopicName());
+
+ // Add KeySharedPolicy for Key_Shared subscription.
+ if (sourceConfiguration.getSubscriptionType() == SubscriptionType.Key_Shared) {
+ KeySharedPolicy policy =
+ KeySharedPolicy.stickyHashRange().ranges(partition.getPulsarRange());
+ consumerBuilder.keySharedPolicy(policy);
+ }
// Create the consumer configuration by using common utils.
- return sneakyClient(
- () -> createConsumer(pulsarClient, Schema.BYTES, configuration, customizer));
+ return sneakyClient(consumerBuilder::subscribe);
}
}
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarUnorderedPartitionSplitReader.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarUnorderedPartitionSplitReader.java
index 52c489a..415bdb3 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarUnorderedPartitionSplitReader.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarUnorderedPartitionSplitReader.java
@@ -20,7 +20,6 @@ package org.apache.flink.connector.pulsar.source.reader.split;
import org.apache.flink.annotation.Internal;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer;
import org.apache.flink.connector.pulsar.common.utils.PulsarTransactionUtils;
import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema;
@@ -36,7 +35,6 @@ import org.apache.pulsar.client.api.transaction.Transaction;
import org.apache.pulsar.client.api.transaction.TransactionCoordinatorClient;
import org.apache.pulsar.client.api.transaction.TransactionCoordinatorClientException;
import org.apache.pulsar.client.api.transaction.TxnID;
-import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -70,17 +68,9 @@ public class PulsarUnorderedPartitionSplitReader<OUT> extends PulsarPartitionSpl
PulsarAdmin pulsarAdmin,
Configuration configuration,
SourceConfiguration sourceConfiguration,
- ConfigurationDataCustomizer<ConsumerConfigurationData<byte[]>>
- consumerConfigurationCustomizer,
PulsarDeserializationSchema<OUT> deserializationSchema,
TransactionCoordinatorClient coordinatorClient) {
- super(
- pulsarClient,
- pulsarAdmin,
- configuration,
- sourceConfiguration,
- consumerConfigurationCustomizer,
- deserializationSchema);
+ super(pulsarClient, pulsarAdmin, configuration, sourceConfiguration, deserializationSchema);
this.coordinatorClient = coordinatorClient;
}
diff --git a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/PulsarSourceEnumeratorTest.java b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/PulsarSourceEnumeratorTest.java
index 9622ee3..9f0dfde 100644
--- a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/PulsarSourceEnumeratorTest.java
+++ b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/PulsarSourceEnumeratorTest.java
@@ -20,7 +20,6 @@ package org.apache.flink.connector.pulsar.source.enumerator;
import org.apache.flink.api.connector.source.mocks.MockSplitEnumeratorContext;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils;
import org.apache.flink.connector.pulsar.source.config.SourceConfiguration;
import org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber;
import org.apache.flink.connector.pulsar.source.enumerator.topic.range.FullRangeGenerator;
@@ -40,15 +39,12 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Set;
-import java.util.StringJoiner;
import java.util.regex.Pattern;
-import static org.apache.flink.connector.pulsar.common.config.ConfigurationDataCustomizer.blankCustomizer;
import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL;
import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SERVICE_URL;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_TYPE;
-import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TOPIC_NAMES;
import static org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor.earliest;
import static org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor.latest;
import static org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber.getTopicPatternSubscriber;
@@ -136,7 +132,6 @@ class PulsarSourceEnumeratorTest extends PulsarTestSuiteBase {
configuration.set(PULSAR_ADMIN_URL, operator().adminUrl());
configuration.set(PULSAR_SERVICE_URL, operator().serviceUrl());
configuration.set(PULSAR_SUBSCRIPTION_TYPE, SubscriptionType.Failover);
- configuration.set(PULSAR_TOPIC_NAMES, PulsarJsonUtils.toString(topics));
PulsarSourceEnumState sourceEnumState =
new PulsarSourceEnumState(
@@ -166,9 +161,8 @@ class PulsarSourceEnumeratorTest extends PulsarTestSuiteBase {
Configuration configuration) {
// Use a TopicPatternSubscriber so that no exception if a subscribed topic hasn't been
// created yet.
- StringJoiner topicNameJoiner = new StringJoiner("|");
- topicsToSubscribe.forEach(topicNameJoiner::add);
- Pattern topicPattern = Pattern.compile(topicNameJoiner.toString());
+ String topicRegex = String.join("|", topicsToSubscribe);
+ Pattern topicPattern = Pattern.compile(topicRegex);
PulsarSubscriber subscriber =
getTopicPatternSubscriber(topicPattern, RegexSubscriptionMode.AllTopics);
if (enablePeriodicPartitionDiscovery) {
@@ -186,7 +180,6 @@ class PulsarSourceEnumeratorTest extends PulsarTestSuiteBase {
new FullRangeGenerator(),
configuration,
sourceConfiguration,
- blankCustomizer(),
enumContext,
assignmentState);
}
[flink] 03/06: [FLINK-23877][connector/pulsar] Change pulsar config
option's type, drop PulsarJsonUtils, make it strong type support.
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 069e4374db57d07910885930b7efd301b730ce74
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Thu Aug 26 04:38:17 2021 +0800
[FLINK-23877][connector/pulsar] Change pulsar config option's type, drop PulsarJsonUtils, make it strong type support.
---
.../pulsar/common/config/PulsarConfigUtils.java | 48 +++--------
.../pulsar/common/config/PulsarOptions.java | 55 ++++++------
.../pulsar/common/utils/PulsarJsonUtils.java | 98 ----------------------
.../pulsar/source/PulsarSourceBuilder.java | 10 +--
.../pulsar/source/PulsarSourceOptions.java | 20 +++--
.../source/config/PulsarSourceConfigUtils.java | 8 +-
6 files changed, 56 insertions(+), 183 deletions(-)
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarConfigUtils.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarConfigUtils.java
index fa9c143..ae5d784 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarConfigUtils.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarConfigUtils.java
@@ -21,7 +21,6 @@ package org.apache.flink.connector.pulsar.common.config;
import org.apache.flink.annotation.Internal;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminBuilder;
@@ -33,6 +32,7 @@ import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.impl.auth.AuthenticationDisabled;
import java.util.Map;
+import java.util.TreeSet;
import java.util.function.Consumer;
import java.util.function.Function;
@@ -100,7 +100,6 @@ public final class PulsarConfigUtils {
setOptionValue(
configuration,
PULSAR_OPERATION_TIMEOUT_MS,
- Math::toIntExact,
timeout -> builder.operationTimeout(timeout, MILLISECONDS));
setOptionValue(configuration, PULSAR_NUM_IO_THREADS, builder::ioThreads);
setOptionValue(configuration, PULSAR_NUM_LISTENER_THREADS, builder::listenerThreads);
@@ -122,16 +121,8 @@ public final class PulsarConfigUtils {
setOptionValue(configuration, PULSAR_TLS_TRUST_STORE_PATH, builder::tlsTrustStorePath);
setOptionValue(
configuration, PULSAR_TLS_TRUST_STORE_PASSWORD, builder::tlsTrustStorePassword);
- setOptionValue(
- configuration,
- PULSAR_TLS_CIPHERS,
- v -> PulsarJsonUtils.toSet(String.class, v),
- builder::tlsCiphers);
- setOptionValue(
- configuration,
- PULSAR_TLS_PROTOCOLS,
- v -> PulsarJsonUtils.toSet(String.class, v),
- builder::tlsProtocols);
+ setOptionValue(configuration, PULSAR_TLS_CIPHERS, TreeSet::new, builder::tlsCiphers);
+ setOptionValue(configuration, PULSAR_TLS_PROTOCOLS, TreeSet::new, builder::tlsProtocols);
setOptionValue(
configuration,
PULSAR_MEMORY_LIMIT_BYTES,
@@ -202,35 +193,21 @@ public final class PulsarConfigUtils {
setOptionValue(configuration, PULSAR_TLS_TRUST_STORE_PATH, builder::tlsTrustStorePath);
setOptionValue(
configuration, PULSAR_TLS_TRUST_STORE_PASSWORD, builder::tlsTrustStorePassword);
- setOptionValue(
- configuration,
- PULSAR_TLS_CIPHERS,
- v -> PulsarJsonUtils.toSet(String.class, v),
- builder::tlsCiphers);
- setOptionValue(
- configuration,
- PULSAR_TLS_PROTOCOLS,
- v -> PulsarJsonUtils.toSet(String.class, v),
- builder::tlsProtocols);
+ setOptionValue(configuration, PULSAR_TLS_CIPHERS, TreeSet::new, builder::tlsCiphers);
+ setOptionValue(configuration, PULSAR_TLS_PROTOCOLS, TreeSet::new, builder::tlsProtocols);
setOptionValue(
configuration,
PULSAR_CONNECT_TIMEOUT,
- Math::toIntExact,
v -> builder.connectionTimeout(v, MILLISECONDS));
setOptionValue(
- configuration,
- PULSAR_READ_TIMEOUT,
- Math::toIntExact,
- v -> builder.readTimeout(v, MILLISECONDS));
+ configuration, PULSAR_READ_TIMEOUT, v -> builder.readTimeout(v, MILLISECONDS));
setOptionValue(
configuration,
PULSAR_REQUEST_TIMEOUT,
- Math::toIntExact,
v -> builder.requestTimeout(v, MILLISECONDS));
setOptionValue(
configuration,
PULSAR_AUTO_CERT_REFRESH_TIME,
- Math::toIntExact,
v -> builder.autoCertRefreshTime(v, MILLISECONDS));
return sneakyClient(builder::build);
@@ -243,7 +220,7 @@ public final class PulsarConfigUtils {
*
* <p>This method behavior is the same as the pulsar command line tools.
*/
- public static Authentication createAuthentication(Configuration configuration) {
+ private static Authentication createAuthentication(Configuration configuration) {
if (configuration.contains(PULSAR_AUTH_PLUGIN_CLASS_NAME)) {
String authPluginClassName = configuration.get(PULSAR_AUTH_PLUGIN_CLASS_NAME);
@@ -252,13 +229,14 @@ public final class PulsarConfigUtils {
return sneakyClient(
() -> AuthenticationFactory.create(authPluginClassName, authParamsString));
} else if (configuration.contains(PULSAR_AUTH_PARAM_MAP)) {
- Map<String, String> paramsMap =
- getOptionValue(
- configuration,
- PULSAR_AUTH_PARAM_MAP,
- v -> PulsarJsonUtils.toMap(String.class, String.class, v));
+ Map<String, String> paramsMap = configuration.get(PULSAR_AUTH_PARAM_MAP);
return sneakyClient(
() -> AuthenticationFactory.create(authPluginClassName, paramsMap));
+ } else {
+ throw new IllegalArgumentException(
+ String.format(
+ "No %s or %s provided",
+ PULSAR_AUTH_PARAMS.key(), PULSAR_AUTH_PARAM_MAP.key()));
}
}
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarOptions.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarOptions.java
index 88c1304..0891e7b 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarOptions.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarOptions.java
@@ -25,8 +25,11 @@ import org.apache.flink.configuration.description.Description;
import org.apache.pulsar.client.api.ProxyProtocol;
+import java.util.List;
+import java.util.Map;
import java.util.concurrent.TimeUnit;
+import static java.util.Collections.emptyMap;
import static org.apache.flink.configuration.description.TextElement.code;
import static org.apache.flink.configuration.description.TextElement.text;
@@ -108,16 +111,15 @@ public final class PulsarOptions {
.add(code("key1:val1,key2:val2"))
.build());
- // The real config type is Map<String, String>, you should provided a json str here.
- public static final ConfigOption<String> PULSAR_AUTH_PARAM_MAP =
+ public static final ConfigOption<Map<String, String>> PULSAR_AUTH_PARAM_MAP =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "authParamMap")
- .stringType()
- .defaultValue("{}");
+ .mapType()
+ .defaultValue(emptyMap());
- public static final ConfigOption<Long> PULSAR_OPERATION_TIMEOUT_MS =
+ public static final ConfigOption<Integer> PULSAR_OPERATION_TIMEOUT_MS =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "operationTimeoutMs")
- .longType()
- .defaultValue(30000L)
+ .intType()
+ .defaultValue(30000)
.withDescription("Operation timeout.");
public static final ConfigOption<Long> PULSAR_STATS_INTERVAL_SECONDS =
@@ -235,6 +237,7 @@ public final class PulsarOptions {
"If the duration passes without a response from a broker, the connection attempt is dropped.")
.build());
+ // TODO This option would be exposed by Pulsar's ClientBuilder in next release.
public static final ConfigOption<Integer> PULSAR_REQUEST_TIMEOUT_MS =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "requestTimeoutMs")
.intType()
@@ -284,15 +287,17 @@ public final class PulsarOptions {
.stringType()
.noDefaultValue();
- // The real config type is Set<String>, you should provided a json str here.
- public static final ConfigOption<String> PULSAR_TLS_CIPHERS =
- ConfigOptions.key(CLIENT_CONFIG_PREFIX + "tlsCiphers").stringType().defaultValue("[]");
+ public static final ConfigOption<List<String>> PULSAR_TLS_CIPHERS =
+ ConfigOptions.key(CLIENT_CONFIG_PREFIX + "tlsCiphers")
+ .stringType()
+ .asList()
+ .defaultValues();
- // The real config type is Set<String>, you should provided a json str here.
- public static final ConfigOption<String> PULSAR_TLS_PROTOCOLS =
+ public static final ConfigOption<List<String>> PULSAR_TLS_PROTOCOLS =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "tlsProtocols")
.stringType()
- .defaultValue("[]");
+ .asList()
+ .defaultValues();
public static final ConfigOption<Long> PULSAR_MEMORY_LIMIT_BYTES =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "memoryLimitBytes")
@@ -328,26 +333,22 @@ public final class PulsarOptions {
.withDescription("Admin URL for Pulsar service.");
// The network connect timeout in millis.
- public static final ConfigOption<Long> PULSAR_CONNECT_TIMEOUT =
- ConfigOptions.key(ADMIN_CONFIG_PREFIX + "connectTimeout")
- .longType()
- .defaultValue(TimeUnit.SECONDS.toMillis(60));
+ public static final ConfigOption<Integer> PULSAR_CONNECT_TIMEOUT =
+ ConfigOptions.key(ADMIN_CONFIG_PREFIX + "connectTimeout").intType().defaultValue(60000);
// The read timeout in millis.
- public static final ConfigOption<Long> PULSAR_READ_TIMEOUT =
- ConfigOptions.key(ADMIN_CONFIG_PREFIX + "readTimeout")
- .longType()
- .defaultValue(TimeUnit.SECONDS.toMillis(60));
+ public static final ConfigOption<Integer> PULSAR_READ_TIMEOUT =
+ ConfigOptions.key(ADMIN_CONFIG_PREFIX + "readTimeout").intType().defaultValue(60000);
// The request timeout in millis.
- public static final ConfigOption<Long> PULSAR_REQUEST_TIMEOUT =
+ public static final ConfigOption<Integer> PULSAR_REQUEST_TIMEOUT =
ConfigOptions.key(ADMIN_CONFIG_PREFIX + "requestTimeout")
- .longType()
- .defaultValue(TimeUnit.SECONDS.toMillis(300));
+ .intType()
+ .defaultValue(300000);
// The auto refresh time for certification in millis.
- public static final ConfigOption<Long> PULSAR_AUTO_CERT_REFRESH_TIME =
+ public static final ConfigOption<Integer> PULSAR_AUTO_CERT_REFRESH_TIME =
ConfigOptions.key(ADMIN_CONFIG_PREFIX + "autoCertRefreshTime")
- .longType()
- .defaultValue(TimeUnit.SECONDS.toMillis(300));
+ .intType()
+ .defaultValue(300000);
}
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/utils/PulsarJsonUtils.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/utils/PulsarJsonUtils.java
deleted file mode 100644
index 64308bc..0000000
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/utils/PulsarJsonUtils.java
+++ /dev/null
@@ -1,98 +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.connector.pulsar.common.utils;
-
-import org.apache.flink.annotation.Internal;
-
-import org.apache.pulsar.shade.com.fasterxml.jackson.core.JsonProcessingException;
-import org.apache.pulsar.shade.com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.pulsar.shade.com.fasterxml.jackson.databind.type.CollectionType;
-import org.apache.pulsar.shade.com.fasterxml.jackson.databind.type.MapType;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.TreeSet;
-
-/** json util for pulsar configuration. */
-@Internal
-public final class PulsarJsonUtils {
-
- /**
- * Pulsar client shade a jackson and use jackson annotation in its config class. We have to use
- * this shaded jackson here.
- */
- private static final ObjectMapper mapper = new ObjectMapper();
-
- private PulsarJsonUtils() {
- // No need to create instance.
- }
-
- public static <K, V> Map<K, V> toMap(Class<K> kClass, Class<V> vClass, String json) {
- return toMap(HashMap.class, kClass, vClass, json);
- }
-
- public static <K, V> Map<K, V> toMap(
- Class<? extends Map> mapClass, Class<K> kClass, Class<V> vClass, String json) {
- MapType mapType = mapper.getTypeFactory().constructMapType(mapClass, kClass, vClass);
- try {
- return mapper.readValue(json, mapType);
- } catch (JsonProcessingException e) {
- throw new IllegalArgumentException(e);
- }
- }
-
- public static <T> Set<T> toSet(Class<T> clazz, String json) {
- return toSet(TreeSet.class, clazz, json);
- }
-
- public static <T> Set<T> toSet(Class<? extends Set> setClass, Class<T> clazz, String json) {
- CollectionType collectionType =
- mapper.getTypeFactory().constructCollectionType(setClass, clazz);
- try {
- return mapper.readValue(json, collectionType);
- } catch (JsonProcessingException e) {
- throw new IllegalArgumentException(e);
- }
- }
-
- public static <T> List<T> toList(Class<T> clazz, String json) {
- return toList(ArrayList.class, clazz, json);
- }
-
- public static <T> List<T> toList(Class<? extends List> listClass, Class<T> clazz, String json) {
- CollectionType collectionType =
- mapper.getTypeFactory().constructCollectionType(listClass, clazz);
- try {
- return mapper.readValue(json, collectionType);
- } catch (JsonProcessingException e) {
- throw new IllegalArgumentException(e);
- }
- }
-
- public static String toString(Object o) {
- try {
- return mapper.writeValueAsString(o);
- } catch (JsonProcessingException e) {
- throw new IllegalArgumentException(e);
- }
- }
-}
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
index 50d93fa..20064f5 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
@@ -25,7 +25,6 @@ import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.UnmodifiableConfiguration;
-import org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils;
import org.apache.flink.connector.pulsar.source.enumerator.cursor.StartCursor;
import org.apache.flink.connector.pulsar.source.enumerator.cursor.StopCursor;
import org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber;
@@ -48,7 +47,6 @@ import java.util.Properties;
import java.util.regex.Pattern;
import static java.lang.Boolean.FALSE;
-import static org.apache.flink.connector.pulsar.common.config.PulsarConfigUtils.getOptionValue;
import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL;
import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ENABLE_TRANSACTION;
import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SERVICE_URL;
@@ -210,7 +208,7 @@ public final class PulsarSourceBuilder<OUT> {
*/
public PulsarSourceBuilder<OUT> setTopics(List<String> topics) {
ensureSubscriberIsNull("topics");
- configuration.set(PULSAR_TOPIC_NAMES, PulsarJsonUtils.toString(topics));
+ configuration.set(PULSAR_TOPIC_NAMES, topics);
this.subscriber = PulsarSubscriber.getTopicListSubscriber(topics);
return this;
}
@@ -438,11 +436,7 @@ public final class PulsarSourceBuilder<OUT> {
// Ensure the topic subscriber for pulsar.
if (subscriber == null) {
if (configuration.contains(PULSAR_TOPIC_NAMES)) {
- List<String> topics =
- getOptionValue(
- configuration,
- PULSAR_TOPIC_NAMES,
- names -> PulsarJsonUtils.toList(String.class, names));
+ List<String> topics = configuration.get(PULSAR_TOPIC_NAMES);
this.subscriber = PulsarSubscriber.getTopicListSubscriber(topics);
} else if (configuration.contains(PULSAR_TOPICS_PATTERN)) {
String topicPatternStr = configuration.get(PULSAR_TOPICS_PATTERN);
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java
index 3cac072..1e45a3d 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java
@@ -32,8 +32,11 @@ import org.apache.pulsar.client.api.SubscriptionMode;
import org.apache.pulsar.client.api.SubscriptionType;
import java.time.Duration;
+import java.util.List;
+import java.util.Map;
import java.util.concurrent.TimeUnit;
+import static java.util.Collections.emptyMap;
import static org.apache.flink.configuration.description.TextElement.code;
import static org.apache.flink.configuration.description.TextElement.text;
@@ -158,10 +161,11 @@ public final class PulsarSourceOptions {
//
///////////////////////////////////////////////////////////////////////////////
- public static final ConfigOption<String> PULSAR_TOPIC_NAMES =
+ public static final ConfigOption<List<String>> PULSAR_TOPIC_NAMES =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "topicNames")
.stringType()
- .defaultValue("[]")
+ .asList()
+ .defaultValues()
.withDescription("Topic name.");
public static final ConfigOption<String> PULSAR_TOPICS_PATTERN =
@@ -364,10 +368,10 @@ public final class PulsarSourceOptions {
code("EncryptionContext"))
.build());
- public static final ConfigOption<String> PULSAR_CONSUMER_PROPERTIES =
+ public static final ConfigOption<Map<String, String>> PULSAR_CONSUMER_PROPERTIES =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "properties")
- .stringType()
- .defaultValue("{}")
+ .mapType()
+ .defaultValue(emptyMap())
.withDescription(
Description.builder()
.text("A name or value property of this consumer.")
@@ -498,10 +502,10 @@ public final class PulsarSourceOptions {
.text("Note: this is only for partitioned consumers.\t")
.build());
- public static final ConfigOption<Long> PULSAR_AUTO_UPDATE_PARTITIONS_INTERVAL_SECONDS =
+ public static final ConfigOption<Integer> PULSAR_AUTO_UPDATE_PARTITIONS_INTERVAL_SECONDS =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "autoUpdatePartitionsIntervalSeconds")
- .longType()
- .defaultValue(60L);
+ .intType()
+ .defaultValue(60);
public static final ConfigOption<Boolean> PULSAR_REPLICATE_SUBSCRIPTION_STATE =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "replicateSubscriptionState")
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/PulsarSourceConfigUtils.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/PulsarSourceConfigUtils.java
index a471bc6..e19dcd5 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/PulsarSourceConfigUtils.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/PulsarSourceConfigUtils.java
@@ -21,7 +21,6 @@ package org.apache.flink.connector.pulsar.source.config;
import org.apache.flink.annotation.Internal;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.connector.pulsar.common.utils.PulsarJsonUtils;
import org.apache.flink.util.Preconditions;
import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList;
@@ -156,11 +155,7 @@ public final class PulsarSourceConfigUtils {
setOptionValue(configuration, PULSAR_CONSUMER_NAME, builder::consumerName);
setOptionValue(configuration, PULSAR_READ_COMPACTED, builder::readCompacted);
setOptionValue(configuration, PULSAR_PRIORITY_LEVEL, builder::priorityLevel);
- setOptionValue(
- configuration,
- PULSAR_CONSUMER_PROPERTIES,
- v -> PulsarJsonUtils.toMap(String.class, String.class, v),
- builder::properties);
+ setOptionValue(configuration, PULSAR_CONSUMER_PROPERTIES, builder::properties);
setOptionValue(
configuration,
PULSAR_SUBSCRIPTION_INITIAL_POSITION,
@@ -170,7 +165,6 @@ public final class PulsarSourceConfigUtils {
setOptionValue(
configuration,
PULSAR_AUTO_UPDATE_PARTITIONS_INTERVAL_SECONDS,
- Math::toIntExact,
v -> builder.autoUpdatePartitionsInterval(v, SECONDS));
if (configuration.getOptional(PULSAR_RESET_INCLUDE_HEAD).orElse(false)) {
builder.startMessageIdInclusive();
[flink] 06/06: [FLINK-23877][connector/pulsar] Enrich the config
options' description, make sure all the options have a detailed description.
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 3ef9323436fbd7a20742b52120b34a53a82a9675
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Thu Aug 26 16:14:17 2021 +0800
[FLINK-23877][connector/pulsar] Enrich the config options' description, make sure all the options have a detailed description.
---
.../pulsar/common/config/PulsarOptions.java | 242 +++++++++++++++++----
.../pulsar/source/PulsarSourceOptions.java | 162 ++++++++++++--
2 files changed, 344 insertions(+), 60 deletions(-)
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarOptions.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarOptions.java
index ed71c91..4c33a70 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarOptions.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarOptions.java
@@ -19,6 +19,8 @@
package org.apache.flink.connector.pulsar.common.config;
import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.docs.ConfigGroup;
+import org.apache.flink.annotation.docs.ConfigGroups;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.description.Description;
@@ -30,20 +32,28 @@ import java.util.Map;
import java.util.concurrent.TimeUnit;
import static java.util.Collections.emptyMap;
+import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.flink.configuration.description.TextElement.code;
import static org.apache.flink.configuration.description.TextElement.text;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.ADMIN_CONFIG_PREFIX;
+import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.CLIENT_CONFIG_PREFIX;
/**
* Configuration for Pulsar Client, these config options would be used for both source, sink and
* table.
*/
@PublicEvolving
+@ConfigGroups(
+ groups = {
+ @ConfigGroup(name = "PulsarClient", keyPrefix = CLIENT_CONFIG_PREFIX),
+ @ConfigGroup(name = "PulsarAdmin", keyPrefix = ADMIN_CONFIG_PREFIX)
+ })
public final class PulsarOptions {
// Pulsar client API config prefix.
- private static final String CLIENT_CONFIG_PREFIX = "pulsar.client.";
+ public static final String CLIENT_CONFIG_PREFIX = "pulsar.client.";
// Pulsar admin API config prefix.
- private static final String ADMIN_CONFIG_PREFIX = "pulsar.admin.";
+ public static final String ADMIN_CONFIG_PREFIX = "pulsar.admin.";
private PulsarOptions() {
// This is a constant class
@@ -114,13 +124,21 @@ public final class PulsarOptions {
public static final ConfigOption<Map<String, String>> PULSAR_AUTH_PARAM_MAP =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "authParamMap")
.mapType()
- .defaultValue(emptyMap());
+ .defaultValue(emptyMap())
+ .withDescription(
+ "Map which represents parameters for the authentication plugin.");
public static final ConfigOption<Integer> PULSAR_OPERATION_TIMEOUT_MS =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "operationTimeoutMs")
.intType()
.defaultValue(30000)
- .withDescription("Operation timeout.");
+ .withDescription(
+ Description.builder()
+ .text("Operation timeout (in millis).")
+ .text(
+ "Producer-create, subscribe and unsubscribe operations will be retried until this interval,"
+ + " after which the operation will be marked as failed.")
+ .build());
public static final ConfigOption<Long> PULSAR_STATS_INTERVAL_SECONDS =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "statsIntervalSeconds")
@@ -150,19 +168,52 @@ public final class PulsarOptions {
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "numListenerThreads")
.intType()
.defaultValue(1)
- .withDescription("The number of threads used for handling message listeners.");
+ .withDescription(
+ Description.builder()
+ .text(
+ "The number of threads used for handling message listeners.")
+ .text(
+ "The listener thread pool is shared across all the consumers and readers that are using a %s model to get messages.",
+ code("listener"))
+ .text(
+ "For a given consumer, the listener will be always invoked from the same thread, to ensure ordering.")
+ .build());
public static final ConfigOption<Integer> PULSAR_CONNECTIONS_PER_BROKER =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "connectionsPerBroker")
.intType()
- .defaultValue(1);
+ .defaultValue(1)
+ .withDescription(
+ Description.builder()
+ .text(
+ "Sets the max number of connection that the client library will open to a single broker.")
+ .linebreak()
+ .text(
+ "By default, the connection pool will use a single connection for all the producers and consumers.")
+ .text(
+ "Increasing this parameter may improve throughput when using many producers over a high latency connection.")
+ .build());
public static final ConfigOption<Boolean> PULSAR_USE_TCP_NO_DELAY =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "useTcpNoDelay")
.booleanType()
.defaultValue(true)
.withDescription(
- "Whether to use TCP no-delay flag on the connection to disable Nagle algorithm.");
+ Description.builder()
+ .text(
+ "Whether to use TCP no-delay flag on the connection to disable Nagle algorithm.")
+ .linebreak()
+ .text(
+ "No-delay features make sure packets are sent out on the network as soon as possible,")
+ .text("and it's critical to achieve low latency publishes.")
+ .text(
+ "On the other hand, sending out a huge number of small packets might limit the overall throughput,")
+ .text(
+ "so if latency is not a concern, it's advisable to set the %s flag to false.",
+ code("useTcpNoDelay"))
+ .linebreak()
+ .text("Default value is true.")
+ .build());
public static final ConfigOption<String> PULSAR_TLS_TRUST_CERTS_FILE_PATH =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "tlsTrustCertsFilePath")
@@ -181,26 +232,52 @@ public final class PulsarOptions {
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "tlsHostnameVerificationEnable")
.booleanType()
.defaultValue(false)
- .withDescription("Whether to enable TLS hostname verification.");
+ .withDescription(
+ Description.builder()
+ .text("Whether to enable TLS hostname verification.")
+ .text(
+ "It allows to validate hostname verification when client connects to broker over tls.")
+ .text(
+ "It validates incoming x509 certificate and matches provided hostname(CN/SAN) with expected broker's host name.")
+ .text(
+ "It follows RFC 2818, 3.1. Server Identity hostname verification.")
+ .build());
public static final ConfigOption<Integer> PULSAR_CONCURRENT_LOOKUP_REQUEST =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "concurrentLookupRequest")
.intType()
.defaultValue(5000)
.withDescription(
- "The number of concurrent lookup requests allowed to send on each broker connection to prevent overload on broker.");
+ Description.builder()
+ .text(
+ "The number of concurrent lookup requests allowed to send on each broker connection to prevent overload on broker.")
+ .text(
+ "It should be configured with higher value only in case of it requires"
+ + " to produce/subscribe on thousands of topic using created %s",
+ code("PulsarClient"))
+ .build());
public static final ConfigOption<Integer> PULSAR_MAX_LOOKUP_REQUEST =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "maxLookupRequest")
.intType()
.defaultValue(50000)
.withDescription(
- "The maximum number of lookup requests allowed on each broker connection to prevent overload on broker.");
+ Description.builder()
+ .text(
+ "The maximum number of lookup requests allowed on each broker connection to prevent overload on broker.")
+ .text("It should be bigger than maxConcurrentLookupRequests.")
+ .text(
+ "Requests that inside maxConcurrentLookupRequests already send to broker,")
+ .text(
+ "and requests beyond maxConcurrentLookupRequests and under maxLookupRequests will wait in each client cnx.")
+ .build());
public static final ConfigOption<Integer> PULSAR_MAX_LOOKUP_REDIRECTS =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "maxLookupRedirects")
.intType()
- .defaultValue(20);
+ .defaultValue(20)
+ .withDescription(
+ "Set the maximum number of times a lookup-request to a broker will be redirected.");
public static final ConfigOption<Integer> PULSAR_MAX_NUMBER_OF_REJECTED_REQUEST_PER_CONNECTION =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "maxNumberOfRejectedRequestPerConnection")
@@ -225,7 +302,7 @@ public final class PulsarOptions {
.withDescription(
Description.builder()
.text(
- "Duration of waiting for a connection to a broker to be established.")
+ "Duration (in millis) of waiting for a connection to a broker to be established.")
.linebreak()
.text(
"If the duration passes without a response from a broker, the connection attempt is dropped.")
@@ -236,82 +313,157 @@ public final class PulsarOptions {
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "requestTimeoutMs")
.intType()
.defaultValue(60000)
- .withDescription("Maximum duration for completing a request.");
+ .withDescription("Maximum duration (in millis) for completing a request.");
public static final ConfigOption<Long> PULSAR_INITIAL_BACKOFF_INTERVAL_NANOS =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "initialBackoffIntervalNanos")
.longType()
.defaultValue(TimeUnit.MILLISECONDS.toNanos(100))
- .withDescription("Default duration for a backoff interval.");
+ .withDescription("Default duration (in nanoseconds) for a backoff interval.");
public static final ConfigOption<Long> PULSAR_MAX_BACKOFF_INTERVAL_NANOS =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "maxBackoffIntervalNanos")
.longType()
- .defaultValue(TimeUnit.SECONDS.toNanos(60))
- .withDescription("Maximum duration for a backoff interval.");
+ .defaultValue(SECONDS.toNanos(60))
+ .withDescription("Maximum duration (in nanoseconds) for a backoff interval.");
public static final ConfigOption<Boolean> PULSAR_ENABLE_BUSY_WAIT =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "enableBusyWait")
.booleanType()
- .defaultValue(false);
+ .defaultValue(false)
+ .withDescription(
+ Description.builder()
+ .text("Option to enable busy-wait settings.")
+ .linebreak()
+ .text(
+ "This option will enable spin-waiting on executors and IO threads in order to reduce latency during context switches.")
+ .text(
+ "The spinning will consume 100% CPU even when the broker is not doing any work.")
+ .text(
+ "It is recommended to reduce the number of IO threads and BK client threads to only have few CPU cores busy.")
+ .build());
public static final ConfigOption<String> PULSAR_LISTENER_NAME =
- ConfigOptions.key(CLIENT_CONFIG_PREFIX + "listenerName").stringType().noDefaultValue();
+ ConfigOptions.key(CLIENT_CONFIG_PREFIX + "listenerName")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "Configure the listenerName that the broker will return the corresponding %s.",
+ code("advertisedListener"))
+ .build());
public static final ConfigOption<Boolean> PULSAR_USE_KEY_STORE_TLS =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "useKeyStoreTls")
.booleanType()
- .defaultValue(false);
+ .defaultValue(false)
+ .withDescription(
+ Description.builder()
+ .text(
+ "If Tls is enabled, whether use KeyStore type as tls configuration parameter.")
+ .text("False means use default pem type configuration.")
+ .build());
public static final ConfigOption<String> PULSAR_SSL_PROVIDER =
- ConfigOptions.key(CLIENT_CONFIG_PREFIX + "sslProvider").stringType().noDefaultValue();
+ ConfigOptions.key(CLIENT_CONFIG_PREFIX + "sslProvider")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text(
+ "The name of the security provider used for SSL connections.")
+ .text(
+ "Default value is the default security provider of the JVM.")
+ .build());
public static final ConfigOption<String> PULSAR_TLS_TRUST_STORE_TYPE =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "tlsTrustStoreType")
.stringType()
- .defaultValue("JKS");
+ .defaultValue("JKS")
+ .withDescription("The file format of the trust store file.");
public static final ConfigOption<String> PULSAR_TLS_TRUST_STORE_PATH =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "tlsTrustStorePath")
.stringType()
- .noDefaultValue();
+ .noDefaultValue()
+ .withDescription("The location of the trust store file.");
public static final ConfigOption<String> PULSAR_TLS_TRUST_STORE_PASSWORD =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "tlsTrustStorePassword")
.stringType()
- .noDefaultValue();
+ .noDefaultValue()
+ .withDescription("The store password for the key store file.");
+ // The real config type is Set<String>, you should provided a json str here.
public static final ConfigOption<List<String>> PULSAR_TLS_CIPHERS =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "tlsCiphers")
.stringType()
.asList()
- .defaultValues();
+ .defaultValues()
+ .withDescription(
+ Description.builder()
+ .text("A list of cipher suites.")
+ .text(
+ "This is a named combination of authentication, encryption,"
+ + " MAC and key exchange algorithm used to negotiate the security"
+ + " settings for a network connection using TLS or SSL network protocol.")
+ .text(
+ "By default all the available cipher suites are supported.")
+ .build());
public static final ConfigOption<List<String>> PULSAR_TLS_PROTOCOLS =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "tlsProtocols")
.stringType()
.asList()
- .defaultValues();
+ .defaultValues()
+ .withDescription(
+ Description.builder()
+ .text("The SSL protocol used to generate the SSLContext.")
+ .text("Default setting is TLS, which is fine for most cases.")
+ .text(
+ "Allowed values in recent JVMs are TLS, TLSv1.3, TLSv1.2 and TLSv1.1.")
+ .build());
public static final ConfigOption<Long> PULSAR_MEMORY_LIMIT_BYTES =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "memoryLimitBytes")
.longType()
- .defaultValue(0L);
+ .defaultValue(0L)
+ .withDescription(
+ Description.builder()
+ .text(
+ "Configure a limit on the amount of direct memory that will be allocated by this client instance. Its unit is byte.")
+ .linebreak()
+ .text(
+ "Note: at this moment this is only limiting the memory for producers.")
+ .text("Setting this to 0 will disable the limit.")
+ .build());
public static final ConfigOption<String> PULSAR_PROXY_SERVICE_URL =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "proxyServiceUrl")
.stringType()
- .noDefaultValue();
+ .noDefaultValue()
+ .withDescription(
+ "Proxy-service url when client would like to connect to broker via proxy."
+ + " Client can choose type of proxy-routing.");
public static final ConfigOption<ProxyProtocol> PULSAR_PROXY_PROTOCOL =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "proxyProtocol")
.enumType(ProxyProtocol.class)
- .noDefaultValue();
+ .defaultValue(ProxyProtocol.SNI)
+ .withDescription(
+ Description.builder()
+ .text(
+ "Protocol type to determine type of proxy routing when client connects to proxy using %s.",
+ code(CLIENT_CONFIG_PREFIX + "proxyServiceUrl"))
+ .build());
public static final ConfigOption<Boolean> PULSAR_ENABLE_TRANSACTION =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "enableTransaction")
.booleanType()
- .defaultValue(false);
+ .defaultValue(false)
+ .withDescription(
+ "If enable transaction, start the transactionCoordinatorClient with pulsar client.");
///////////////////////////////////////////////////////////////////////////////
//
@@ -324,25 +476,39 @@ public final class PulsarOptions {
ConfigOptions.key(ADMIN_CONFIG_PREFIX + "adminUrl")
.stringType()
.noDefaultValue()
- .withDescription("Admin URL for Pulsar service.");
+ .withDescription(
+ Description.builder()
+ .text(
+ "Set the Pulsar service HTTP URL for the admin endpoint. eg. %s, or %s for TLS.",
+ code("http://my-broker.example.com:8080"),
+ code("https://my-broker.example.com:8443"))
+ .build());
- // The network connect timeout in millis.
public static final ConfigOption<Integer> PULSAR_CONNECT_TIMEOUT =
- ConfigOptions.key(ADMIN_CONFIG_PREFIX + "connectTimeout").intType().defaultValue(60000);
+ ConfigOptions.key(ADMIN_CONFIG_PREFIX + "connectTimeout")
+ .intType()
+ .defaultValue(60000)
+ .withDescription(
+ "This sets the connection time out (in millis) for the pulsar admin client.");
- // The read timeout in millis.
public static final ConfigOption<Integer> PULSAR_READ_TIMEOUT =
- ConfigOptions.key(ADMIN_CONFIG_PREFIX + "readTimeout").intType().defaultValue(60000);
+ ConfigOptions.key(ADMIN_CONFIG_PREFIX + "readTimeout")
+ .intType()
+ .defaultValue(60000)
+ .withDescription(
+ "This sets the server response read time out (in millis) for the pulsar admin client for any request.");
- // The request timeout in millis.
public static final ConfigOption<Integer> PULSAR_REQUEST_TIMEOUT =
ConfigOptions.key(ADMIN_CONFIG_PREFIX + "requestTimeout")
.intType()
- .defaultValue(300000);
+ .defaultValue(300000)
+ .withDescription(
+ "This sets the server request time out (in millis) for the pulsar admin client for any request.");
- // The auto refresh time for certification in millis.
public static final ConfigOption<Integer> PULSAR_AUTO_CERT_REFRESH_TIME =
ConfigOptions.key(ADMIN_CONFIG_PREFIX + "autoCertRefreshTime")
.intType()
- .defaultValue(300000);
+ .defaultValue(300000)
+ .withDescription(
+ "This sets auto cert refresh time (in millis) if Pulsar admin uses tls authentication.");
}
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java
index 2f6a779..2884f5e 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java
@@ -19,6 +19,8 @@
package org.apache.flink.connector.pulsar.source;
import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.docs.ConfigGroup;
+import org.apache.flink.annotation.docs.ConfigGroups;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.description.Description;
@@ -37,6 +39,8 @@ import java.util.concurrent.TimeUnit;
import static java.util.Collections.emptyMap;
import static org.apache.flink.configuration.description.TextElement.code;
import static org.apache.flink.configuration.description.TextElement.text;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.CONSUMER_CONFIG_PREFIX;
+import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.SOURCE_CONFIG_PREFIX;
/**
* Configurations for PulsarSource. All the options list here could be configured in {@link
@@ -46,12 +50,17 @@ import static org.apache.flink.configuration.description.TextElement.text;
* @see PulsarOptions
*/
@PublicEvolving
+@ConfigGroups(
+ groups = {
+ @ConfigGroup(name = "PulsarSource", keyPrefix = SOURCE_CONFIG_PREFIX),
+ @ConfigGroup(name = "PulsarConsumer", keyPrefix = CONSUMER_CONFIG_PREFIX)
+ })
public final class PulsarSourceOptions {
// Pulsar source connector config prefix.
- private static final String SOURCE_CONFIG_PREFIX = "pulsar.source.";
+ public static final String SOURCE_CONFIG_PREFIX = "pulsar.source.";
// Pulsar consumer API config prefix.
- private static final String CONSUMER_CONFIG_PREFIX = "pulsar.consumer.";
+ public static final String CONSUMER_CONFIG_PREFIX = "pulsar.consumer.";
private PulsarSourceOptions() {
// This is a constant class
@@ -99,8 +108,12 @@ public final class PulsarSourceOptions {
.longType()
.defaultValue(Duration.ofSeconds(5).toMillis())
.withDescription(
- "This option is used only when user disabled checkpoint"
- + " and using Exclusive or Failover subscription");
+ Description.builder()
+ .text(
+ "This option is used only when user disabled checkpoint and using Exclusive or Failover subscription.")
+ .text(
+ "We would automatically commit the cursor using the given period (in millis).")
+ .build());
public static final ConfigOption<Long> PULSAR_TRANSACTION_TIMEOUT_MILLIS =
ConfigOptions.key(SOURCE_CONFIG_PREFIX + "transactionTimeoutMillis")
@@ -115,6 +128,7 @@ public final class PulsarSourceOptions {
.linebreak()
.text(
"This value should be greater than the checkpoint interval.")
+ .text("It uses milliseconds as the unit of time.")
.build());
public static final ConfigOption<Long> PULSAR_MAX_FETCH_TIME =
@@ -124,7 +138,7 @@ public final class PulsarSourceOptions {
.withDescription(
Description.builder()
.text(
- "The max time to wait when fetching records. "
+ "The max time (in millis) to wait when fetching records. "
+ "A longer time increases throughput but also latency. "
+ "A fetch batch might be finished earlier because of %s.",
code("pulsar.source.maxFetchRecords"))
@@ -163,7 +177,9 @@ public final class PulsarSourceOptions {
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "subscriptionName")
.stringType()
.noDefaultValue()
- .withDescription("Subscription name.");
+ .withDescription(
+ "Specify the subscription name for this consumer."
+ + " This argument is required when constructing the consumer.");
public static final ConfigOption<SubscriptionType> PULSAR_SUBSCRIPTION_TYPE =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "subscriptionType")
@@ -185,7 +201,19 @@ public final class PulsarSourceOptions {
public static final ConfigOption<SubscriptionMode> PULSAR_SUBSCRIPTION_MODE =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "subscriptionMode")
.enumType(SubscriptionMode.class)
- .defaultValue(SubscriptionMode.Durable);
+ .defaultValue(SubscriptionMode.Durable)
+ .withDescription(
+ Description.builder()
+ .text(
+ "Select the subscription mode to be used when subscribing to the topic.")
+ .list(
+ text(
+ "%s: Make the subscription to be backed by a durable cursor that will retain messages and persist the current position.",
+ code("Durable")),
+ text(
+ "%s: Lightweight subscription mode that doesn't have a durable cursor associated",
+ code("NonDurable")))
+ .build());
public static final ConfigOption<Integer> PULSAR_RECEIVER_QUEUE_SIZE =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "receiverQueueSize")
@@ -209,7 +237,8 @@ public final class PulsarSourceOptions {
.defaultValue(TimeUnit.MILLISECONDS.toMicros(100))
.withDescription(
Description.builder()
- .text("Group a consumer acknowledgment for a specified time.")
+ .text(
+ "Group a consumer acknowledgment for a specified time (in microseconds).")
.linebreak()
.text(
"By default, a consumer uses 100ms grouping time to send out acknowledgments to a broker.")
@@ -228,7 +257,7 @@ public final class PulsarSourceOptions {
.withDescription(
Description.builder()
.text(
- "Delay to wait before redelivering messages that failed to be processed.")
+ "Delay (in microseconds) to wait before redelivering messages that failed to be processed.")
.linebreak()
.text(
"When an application uses %s, failed messages are redelivered after a fixed timeout.",
@@ -255,13 +284,28 @@ public final class PulsarSourceOptions {
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "consumerName")
.stringType()
.noDefaultValue()
- .withDescription("Consumer name.");
+ .withDescription(
+ "Consumer name is informative and it can be used to identify a particular consumer instance from the topic stats.");
public static final ConfigOption<Long> PULSAR_ACK_TIMEOUT_MILLIS =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "ackTimeoutMillis")
.longType()
.defaultValue(0L)
- .withDescription("Timeout of unacknowledged messages.");
+ .withDescription(
+ Description.builder()
+ .text(
+ "Set the timeout (in millis) for unacknowledged messages, truncated to the nearest millisecond."
+ + " The timeout needs to be greater than 1 second.")
+ .linebreak()
+ .text(
+ "By default, the acknowledge timeout is disabled and that means that messages delivered to a"
+ + " consumer will not be re-delivered unless the consumer crashes.")
+ .linebreak()
+ .text(
+ "When enabling ack timeout, if a message is not acknowledged within the specified timeout"
+ + " it will be re-delivered to the consumer"
+ + " (possibly to a different consumer in case of a shared subscription).")
+ .build());
public static final ConfigOption<Long> PULSAR_TICK_DURATION_MILLIS =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "tickDurationMillis")
@@ -269,7 +313,7 @@ public final class PulsarSourceOptions {
.defaultValue(1000L)
.withDescription(
Description.builder()
- .text("Granularity of the ack-timeout redelivery.")
+ .text("Granularity (in millis) of the ack-timeout redelivery.")
.linebreak()
.text(
"Using an higher %s reduces the memory overhead to track messages when setting ack-timeout to a bigger value (for example, 1 hour).",
@@ -315,17 +359,66 @@ public final class PulsarSourceOptions {
public static final ConfigOption<Integer> PULSAR_MAX_PENDING_CHUNKED_MESSAGE =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "maxPendingChunkedMessage")
.intType()
- .defaultValue(10);
+ .defaultValue(10)
+ .withDescription(
+ Description.builder()
+ .text(
+ "Consumer buffers chunk messages into memory until it receives all the chunks of the original message.")
+ .text(
+ "While consuming chunk-messages, chunks from same message might not be contiguous"
+ + " in the stream and they might be mixed with other messages' chunks.")
+ .text(
+ "So, consumer has to maintain multiple buffers to manage chunks coming from different messages.")
+ .text(
+ "This mainly happens when multiple publishers are publishing messages on the topic"
+ + " concurrently or publisher failed to publish all chunks of the messages.")
+ .linebreak()
+ .text("eg: M1-C1, M2-C1, M1-C2, M2-C2")
+ .text(
+ "Messages M1-C1 and M1-C2 belong to original message M1, M2-C1 and M2-C2 messages belong to M2 message.")
+ .linebreak()
+ .text(
+ "Buffering large number of outstanding uncompleted chunked messages can create memory"
+ + " pressure and it can be guarded by providing this %s threshold."
+ + " Once, consumer reaches this threshold, it drops the outstanding unchunked-messages"
+ + " by silently acking or asking broker to redeliver later by marking it unacked."
+ + " This behavior can be controlled by configuration %s",
+ code(
+ CONSUMER_CONFIG_PREFIX
+ + "maxPendingChunkedMessage"),
+ code(
+ CONSUMER_CONFIG_PREFIX
+ + "autoAckOldestChunkedMessageOnQueueFull"))
+ .build());
public static final ConfigOption<Boolean> PULSAR_AUTO_ACK_OLDEST_CHUNKED_MESSAGE_ON_QUEUE_FULL =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "autoAckOldestChunkedMessageOnQueueFull")
.booleanType()
- .defaultValue(false);
+ .defaultValue(false)
+ .withDescription(
+ Description.builder()
+ .text(
+ "Buffering large number of outstanding uncompleted chunked messages can create memory pressure"
+ + " and it can be guarded by providing this %s threshold."
+ + " Once, consumer reaches this threshold, it drops the outstanding unchunked-messages"
+ + " by silently acking if %s is true else it marks them for redelivery.",
+ code(
+ CONSUMER_CONFIG_PREFIX
+ + "maxPendingChunkedMessage"),
+ code(
+ CONSUMER_CONFIG_PREFIX
+ + "autoAckOldestChunkedMessageOnQueueFull"))
+ .build());
public static final ConfigOption<Long> PULSAR_EXPIRE_TIME_OF_INCOMPLETE_CHUNKED_MESSAGE_MILLIS =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "expireTimeOfIncompleteChunkedMessageMillis")
.longType()
- .defaultValue(60 * 1000L);
+ .defaultValue(60 * 1000L)
+ .withDescription(
+ "If producer fails to publish all the chunks of a message then consumer"
+ + " can expire incomplete chunks if consumer won't be able to"
+ + " receive all chunks in expire times (default 1 hour)."
+ + " It uses milliseconds as the unit of time.");
public static final ConfigOption<ConsumerCryptoFailureAction> PULSAR_CRYPTO_FAILURE_ACTION =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "cryptoFailureAction")
@@ -429,21 +522,28 @@ public final class PulsarSourceOptions {
public static final ConfigOption<Integer> PULSAR_MAX_REDELIVER_COUNT =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "deadLetterPolicy.maxRedeliverCount")
.intType()
- .defaultValue(0);
+ .defaultValue(0)
+ .withDescription(
+ "Maximum number of times that a message will be redelivered before being sent to the dead letter queue.");
public static final ConfigOption<String> PULSAR_RETRY_LETTER_TOPIC =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "deadLetterPolicy.retryLetterTopic")
.stringType()
- .noDefaultValue();
+ .noDefaultValue()
+ .withDescription(
+ "Name of the retry topic where the failing messages will be sent.");
public static final ConfigOption<String> PULSAR_DEAD_LETTER_TOPIC =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "deadLetterPolicy.deadLetterTopic")
.stringType()
- .noDefaultValue();
+ .noDefaultValue()
+ .withDescription(
+ "Name of the dead topic where the failing messages will be sent.");
public static final ConfigOption<Boolean> PULSAR_RETRY_ENABLE =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "retryEnable")
.booleanType()
- .defaultValue(false);
+ .defaultValue(false)
+ .withDescription("If enabled, the consumer will auto retry messages.");
public static final ConfigOption<Boolean> PULSAR_AUTO_UPDATE_PARTITIONS =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "autoUpdatePartitions")
@@ -461,7 +561,10 @@ public final class PulsarSourceOptions {
public static final ConfigOption<Integer> PULSAR_AUTO_UPDATE_PARTITIONS_INTERVAL_SECONDS =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "autoUpdatePartitionsIntervalSeconds")
.intType()
- .defaultValue(60);
+ .defaultValue(60)
+ .withDescription(
+ "Set the interval (in seconds) of updating partitions."
+ + " This only works if autoUpdatePartitions is enabled.");
public static final ConfigOption<Boolean> PULSAR_REPLICATE_SUBSCRIPTION_STATE =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "replicateSubscriptionState")
@@ -477,10 +580,25 @@ public final class PulsarSourceOptions {
public static final ConfigOption<Boolean> PULSAR_ACK_RECEIPT_ENABLED =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "ackReceiptEnabled")
.booleanType()
- .defaultValue(false);
+ .defaultValue(false)
+ .withDescription(
+ "Ack will return receipt but does not mean that the message will not be resent after get receipt.");
public static final ConfigOption<Boolean> PULSAR_POOL_MESSAGES =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "poolMessages")
.booleanType()
- .defaultValue(false);
+ .defaultValue(false)
+ .withDescription(
+ Description.builder()
+ .text(
+ "Enable pooling of messages and the underlying data buffers.")
+ .linebreak()
+ .text(
+ "When pooling is enabled, the application is responsible for calling"
+ + " %s after the handling of every received message. If %s"
+ + " is not called on a received message, there will be a memory leak."
+ + " If an application attempts to use and already \"released\" message,"
+ + " it might experience undefined behavior (eg: memory corruption, deserialization error, etc.).",
+ code("Message.release()"), code("release()"))
+ .build());
}
[flink] 05/06: [FLINK-23877][connector/pulsar] Since we have use
the StartCursor,
it's no need to expose the resetIncludeHead option for users.
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 111b7ec313076a4242140846b52c55260f795784
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Thu Aug 26 05:04:47 2021 +0800
[FLINK-23877][connector/pulsar] Since we have use the StartCursor, it's no need to expose the resetIncludeHead option for users.
---
.../flink/connector/pulsar/source/PulsarSourceOptions.java | 10 ----------
.../pulsar/source/config/PulsarSourceConfigUtils.java | 4 ----
2 files changed, 14 deletions(-)
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java
index 34938f3..2f6a779 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java
@@ -474,16 +474,6 @@ public final class PulsarSourceOptions {
code("replicateSubscriptionState"))
.build());
- public static final ConfigOption<Boolean> PULSAR_RESET_INCLUDE_HEAD =
- ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "resetIncludeHead")
- .booleanType()
- .defaultValue(false);
-
- public static final ConfigOption<Boolean> PULSAR_BATCH_INDEX_ACK_ENABLED =
- ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "batchIndexAckEnabled")
- .booleanType()
- .defaultValue(false);
-
public static final ConfigOption<Boolean> PULSAR_ACK_RECEIPT_ENABLED =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "ackReceiptEnabled")
.booleanType()
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/PulsarSourceConfigUtils.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/PulsarSourceConfigUtils.java
index e19dcd5..b212aa7 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/PulsarSourceConfigUtils.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/PulsarSourceConfigUtils.java
@@ -64,7 +64,6 @@ import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSA
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_READ_COMPACTED;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_RECEIVER_QUEUE_SIZE;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_REPLICATE_SUBSCRIPTION_STATE;
-import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_RESET_INCLUDE_HEAD;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_RETRY_ENABLE;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_RETRY_LETTER_TOPIC;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_INITIAL_POSITION;
@@ -166,9 +165,6 @@ public final class PulsarSourceConfigUtils {
configuration,
PULSAR_AUTO_UPDATE_PARTITIONS_INTERVAL_SECONDS,
v -> builder.autoUpdatePartitionsInterval(v, SECONDS));
- if (configuration.getOptional(PULSAR_RESET_INCLUDE_HEAD).orElse(false)) {
- builder.startMessageIdInclusive();
- }
setOptionValue(configuration, PULSAR_RETRY_ENABLE, builder::enableRetry);
setOptionValue(
configuration,