You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ti...@apache.org on 2023/02/08 14:58:05 UTC
[flink-connector-pulsar] branch main updated: [FLINK-28351][Connector/Pulsar] Add dynamic sink topic support for Pulsar connector. (#14)
This is an automated email from the ASF dual-hosted git repository.
tison pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
The following commit(s) were added to refs/heads/main by this push:
new 39c1a15 [FLINK-28351][Connector/Pulsar] Add dynamic sink topic support for Pulsar connector. (#14)
39c1a15 is described below
commit 39c1a156a795f67aaf143f096f489e4ef0ca6304
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Wed Feb 8 22:57:59 2023 +0800
[FLINK-28351][Connector/Pulsar] Add dynamic sink topic support for Pulsar connector. (#14)
---
.../docs/connectors/datastream/pulsar.md | 23 +-
docs/content/docs/connectors/datastream/pulsar.md | 26 ++-
.../generated/pulsar_source_configuration.html | 2 +-
.../flink/tests/util/pulsar/PulsarSinkE2ECase.java | 16 +-
.../tests/util/pulsar/PulsarSourceE2ECase.java | 16 +-
.../flink/connector/pulsar/sink/PulsarSink.java | 6 +-
.../connector/pulsar/sink/PulsarSinkBuilder.java | 8 +-
.../connector/pulsar/sink/writer/PulsarWriter.java | 68 +++---
.../sink/writer/context/PulsarSinkContext.java | 16 +-
.../sink/writer/context/PulsarSinkContextImpl.java | 18 +-
.../sink/writer/message/PulsarMessageBuilder.java | 6 +-
.../sink/writer/router/KeyHashTopicRouter.java | 4 +-
.../sink/writer/router/RoundRobinTopicRouter.java | 4 +-
.../pulsar/sink/writer/router/TopicRouter.java | 4 +-
.../pulsar/sink/writer/topic/MetadataListener.java | 235 +++++++++++++++++++++
...ProducerRegister.java => ProducerRegister.java} | 25 ++-
.../sink/writer/topic/TopicMetadataListener.java | 181 ----------------
.../pulsar/source/PulsarSourceOptions.java | 2 +-
.../PulsarSourceEnumStateSerializer.java | 8 +-
.../subscriber/impl/BasePulsarSubscriber.java | 2 +-
.../source/enumerator/topic/TopicPartition.java | 35 ++-
.../pulsar/source/enumerator/topic/TopicRange.java | 7 +-
.../split/PulsarPartitionSplitSerializer.java | 2 -
.../connector/pulsar/sink/PulsarSinkITCase.java | 29 ++-
.../pulsar/sink/writer/PulsarWriterTest.java | 50 ++++-
.../sink/writer/router/KeyHashTopicRouterTest.java | 33 +--
.../writer/router/RoundRobinTopicRouterTest.java | 18 +-
...ListenerTest.java => MetadataListenerTest.java} | 74 +++----
...RegisterTest.java => ProducerRegisterTest.java} | 16 +-
.../pulsar/source/PulsarSourceITCase.java | 17 +-
.../enumerator/assigner/SplitAssignerImplTest.java | 3 +-
.../source/enumerator/cursor/StopCursorTest.java | 2 +-
.../subscriber/PulsarSubscriberTest.java | 26 ++-
.../source/reader/PulsarSourceReaderTest.java | 11 +-
.../PulsarDeserializationSchemaTest.java | 4 +-
.../pulsar/testutils/PulsarTestContext.java | 7 +-
.../pulsar/testutils/PulsarTestSuiteBase.java | 6 +-
.../pulsar/testutils/runtime/PulsarRuntime.java | 10 +
.../testutils/runtime/PulsarRuntimeOperator.java | 11 +-
.../runtime/container/PulsarContainerRuntime.java | 6 +-
.../runtime/singleton/PulsarSingletonRuntime.java | 87 ++++++++
.../sink/{cases => }/PulsarSinkTestContext.java | 108 ++++++----
.../AutoCreateTopicProducingContext.java} | 31 +--
....java => EncryptedMessageProducingContext.java} | 24 +--
.../sink/cases/MultipleTopicsProducingContext.java | 50 +++++
.../SingleTopicProducingContext.java} | 23 +-
.../sink/reader/PulsarEncryptDataReader.java | 6 +-
.../sink/reader/PulsarPartitionDataReader.java | 13 +-
.../testutils/source/PulsarSourceTestContext.java | 2 +-
...java => EncryptedMessagesConsumingContext.java} | 4 +-
...xt.java => MultipleTopicsConsumingContext.java} | 4 +-
.../source/cases/PartialKeysConsumingContext.java | 2 +-
.../source/cases/SingleTopicConsumingContext.java | 2 +-
53 files changed, 915 insertions(+), 478 deletions(-)
diff --git a/docs/content.zh/docs/connectors/datastream/pulsar.md b/docs/content.zh/docs/connectors/datastream/pulsar.md
index 2f732bd..288b5bf 100644
--- a/docs/content.zh/docs/connectors/datastream/pulsar.md
+++ b/docs/content.zh/docs/connectors/datastream/pulsar.md
@@ -30,7 +30,7 @@ Flink 当前提供 [Apache Pulsar](https://pulsar.apache.org) Source 和 Sink
当前支持 Pulsar 2.10.0 及其之后的版本，建议在总是将 Pulsar 升级至最新版。如果想要了解更多关于 Pulsar API 兼容性设计，可以阅读文档 [PIP-72](https://github.com/apache/pulsar/wiki/PIP-72%3A-Introduce-Pulsar-Interface-Taxonomy%3A-Audience-and-Stability-Classification)。
-{{< artifact flink-connector-pulsar >}}
+{{< artifact flink-connector-pulsar 4.0.0-SNAPSHOT >}}
{{< py_download_link "pulsar" >}}
@@ -519,7 +519,9 @@ PulsarSource.builder()
{{< /tabs >}}
{{< hint warning >}}
-默认情况下，Pulsar 启用动态分区发现，查询间隔为 30 秒。用户可以给定一个负数，将该功能禁用。如果使用批的方式消费数据，将无法启用该功能。
+- 默认情况下，Pulsar 启用分区发现，查询间隔为 5 分钟。用户可以给定一个负数，将该功能禁用。如果使用批的方式消费数据，将无法启用该功能。
+- 如果需要禁用分区发现功能，你需要将查询间隔设置为负值。
+- 在 bounded 消费模式下，即使将分区发现的查询间隔设置为正值，也会被禁用。
{{< /hint >}}
### 事件时间和水位线
@@ -662,6 +664,21 @@ PulsarSink.builder().set_topics(["topic-a-partition-0", "topic-a-partition-2", "
举个例子，如果通过 `PulsarSink.builder().setTopics("some-topic1", "some-topic1-partition-0")` 来指定写入的 Topic，那么其结果等价于 `PulsarSink.builder().setTopics("some-topic1")`。
{{< /hint >}}
+#### 基于消息实例的动态 Topic 指定
+
+除了前面说的一开始就指定 Topic 或者是 Topic 分区，你还可以在程序启动后基于消息内容动态指定 Topic，只需要实现 `TopicRouter` 接口即可。使用 `PulsarSinkContext.topicMetadata(String)` 方法来查询某个 Topic 在 Pulsar 上有多少个分区，查询结果会缓存并在 `PulsarSinkOptions.PULSAR_TOPIC_METADATA_REFRESH_INTERVAL` 毫秒之后失效。
+
+此方法同样支持将消息写入一个不存在的 Topic，可以在 `TopicRouter` 内返回想要创建的 Topic，连接器将会**尝试**创建。
+
+{{< hint warning >}}
+如果使用 Topic 自动创建功能，需要在 Pulsar 的 `broker.conf` 配置文件内配置 `allowAutoTopicCreation=true` 来启用对应的功能。
+
+`broker.conf` 配置文件的 `allowAutoTopicCreationType` 选项可以控制自动创建的 Topic 的类型。
+
+- `non-partitioned`: 默认配置，创建的 Topic 没有分区，并且不可以手动创建分区。
+- `partitioned`: 创建的 Topic 将按照 `defaultNumPartitions` 选项定义的个数创建对应的分区。
+{{< /hint >}}
+
### 序列化器
序列化器（`PulsarSerializationSchema`）负责将 Flink 中的每条记录序列化成 byte 数组，并通过网络发送至指定的写入 Topic。和 Pulsar Source 类似的是，序列化器同时支持使用基于 Flink 的 `SerializationSchema` 接口实现序列化器和使用 Pulsar 原生的 `Schema` 类型实现的序列化器。不过序列化器并不支持 Pulsar 的 `Schema.AUTO_PRODUCE_BYTES()`。
@@ -774,7 +791,7 @@ Pulsar 的 topic 至少会包含一种 Schema，`Schema.BYTES` 是默认的 Sche
@PublicEvolving
public interface TopicRouter<IN> extends Serializable {
- String route(IN in, List<String> partitions, PulsarSinkContext context);
+ TopicPartition route(IN in, List<TopicPartition> partitions, PulsarSinkContext context);
default void open(SinkConfiguration sinkConfiguration) {
// 默认无操作
diff --git a/docs/content/docs/connectors/datastream/pulsar.md b/docs/content/docs/connectors/datastream/pulsar.md
index 78bfb7a..20d41ab 100644
--- a/docs/content/docs/connectors/datastream/pulsar.md
+++ b/docs/content/docs/connectors/datastream/pulsar.md
@@ -31,7 +31,7 @@ Flink provides an [Apache Pulsar](https://pulsar.apache.org) connector for readi
You can use the connector with the Pulsar 2.10.0 or higher. It is recommended to always use the latest Pulsar version.
The details on Pulsar compatibility can be found in [PIP-72](https://github.com/apache/pulsar/wiki/PIP-72%3A-Introduce-Pulsar-Interface-Taxonomy%3A-Audience-and-Stability-Classification).
-{{< artifact flink-connector-pulsar >}}
+{{< artifact flink-connector-pulsar 4.0.0-SNAPSHOT >}}
{{< py_download_link "pulsar" >}}
@@ -596,7 +596,7 @@ PulsarSource.builder()
{{< /tabs >}}
{{< hint warning >}}
-- Partition discovery is **enabled** by default. The Pulsar connector queries the topic metadata every 30 seconds.
+- Partition discovery is **enabled** by default. The Pulsar connector queries the topic metadata every 5 minutes.
- To disable partition discovery, you need to set a negative partition discovery interval.
- Partition discovery is disabled for bounded data even if you set this option with a non-negative value.
{{< /hint >}}
@@ -760,6 +760,26 @@ For example, when using the `PulsarSink.builder().setTopics("some-topic1", "some
this is simplified to `PulsarSink.builder().setTopics("some-topic1")`.
{{< /hint >}}
+#### Dynamic Topics by incoming messages
+
+Topics could be defined by the incoming messages instead of providing the fixed topic set in builder. You can dynamically
+provide the topic by in a custom `TopicRouter`. The topic metadata can be queried by using `PulsarSinkContext.topicMetadata(String)`
+and the query result would be cached and expire in `PulsarSinkOptions.PULSAR_TOPIC_METADATA_REFRESH_INTERVAL` milliseconds.
+
+If you want to write to a non-existed topic, just return it in `TopicRouter`. Pulsar connector will **try to create it**.
+
+{{< hint warning >}}
+You need to enable the topic auto creation in Pulsar's `broker.conf` when you want to write messages to a non-existed topic.
+Set the `allowAutoTopicCreation=true` to enable it.
+
+The `allowAutoTopicCreationType` option in `broker.conf` is used to control the type of topic that is allowed to be automatically created.
+
+- `non-partitioned`: The default type for the auto-created topic.
+ It doesn't have any partition and can't be converted to a partitioned topic.
+- `partitioned`: The topic will be created as a partitioned topic.
+ Set the `defaultNumPartitions` option to control the auto created partition size.
+{{< /hint >}}
+
### Serializer
A serializer (`PulsarSerializationSchema`) is required for serializing the record instance into bytes.
@@ -897,7 +917,7 @@ Thus, you do not need to specify topics using the `PulsarSinkBuilder.setTopics`
@PublicEvolving
public interface TopicRouter<IN> extends Serializable {
- String route(IN in, List<String> partitions, PulsarSinkContext context);
+ TopicPartition route(IN in, List<TopicPartition> partitions, PulsarSinkContext context);
default void open(SinkConfiguration sinkConfiguration) {
// Nothing to do by default.
diff --git a/docs/layouts/shortcodes/generated/pulsar_source_configuration.html b/docs/layouts/shortcodes/generated/pulsar_source_configuration.html
index 94c2bd8..42b6bbf 100644
--- a/docs/layouts/shortcodes/generated/pulsar_source_configuration.html
+++ b/docs/layouts/shortcodes/generated/pulsar_source_configuration.html
@@ -58,7 +58,7 @@
</tr>
<tr>
<td><h5>pulsar.source.partitionDiscoveryIntervalMs</h5></td>
- <td style="word-wrap: break-word;">30000</td>
+ <td style="word-wrap: break-word;">300000</td>
<td>Long</td>
<td>The interval (in ms) for the Pulsar source to discover the new partitions. A non-positive value disables the partition discovery.</td>
</tr>
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSinkE2ECase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSinkE2ECase.java
index cdf8a23..e2f7976 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSinkE2ECase.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSinkE2ECase.java
@@ -19,7 +19,7 @@
package org.apache.flink.tests.util.pulsar;
import org.apache.flink.connector.pulsar.testutils.PulsarTestContextFactory;
-import org.apache.flink.connector.pulsar.testutils.sink.cases.PulsarSinkTestContext;
+import org.apache.flink.connector.pulsar.testutils.sink.cases.SingleTopicProducingContext;
import org.apache.flink.connector.testframe.junit.annotations.TestContext;
import org.apache.flink.connector.testframe.junit.annotations.TestEnv;
import org.apache.flink.connector.testframe.junit.annotations.TestExternalSystem;
@@ -31,17 +31,17 @@ import org.apache.flink.tests.util.pulsar.common.PulsarContainerTestEnvironment;
import org.junit.jupiter.api.Tag;
-/** Pulsar sink E2E test based on connector testing framework. */
+import static org.apache.flink.streaming.api.CheckpointingMode.AT_LEAST_ONCE;
+import static org.apache.flink.streaming.api.CheckpointingMode.EXACTLY_ONCE;
+
+/** Pulsar sink E2E test based on the connector testing framework. */
@SuppressWarnings("unused")
@Tag("org.apache.flink.testutils.junit.FailsOnJava11")
public class PulsarSinkE2ECase extends SinkTestSuiteBase<String> {
// Defines the Semantic.
@TestSemantics
- CheckpointingMode[] semantics =
- new CheckpointingMode[] {
- CheckpointingMode.EXACTLY_ONCE, CheckpointingMode.AT_LEAST_ONCE
- };
+ CheckpointingMode[] semantics = new CheckpointingMode[] {AT_LEAST_ONCE, EXACTLY_ONCE};
// Defines TestEnvironment
@TestEnv
@@ -53,6 +53,6 @@ public class PulsarSinkE2ECase extends SinkTestSuiteBase<String> {
// Defines a set of external context Factories for different test cases.
@TestContext
- PulsarTestContextFactory<String, PulsarSinkTestContext> sinkContext =
- new PulsarTestContextFactory<>(pulsar, PulsarSinkTestContext::new);
+ PulsarTestContextFactory<String, SingleTopicProducingContext> sinkContext =
+ new PulsarTestContextFactory<>(pulsar, SingleTopicProducingContext::new);
}
diff --git a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceE2ECase.java b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceE2ECase.java
index 385e58a..ac2388a 100644
--- a/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceE2ECase.java
+++ b/flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceE2ECase.java
@@ -19,7 +19,7 @@
package org.apache.flink.tests.util.pulsar;
import org.apache.flink.connector.pulsar.testutils.PulsarTestContextFactory;
-import org.apache.flink.connector.pulsar.testutils.source.cases.MultipleTopicConsumingContext;
+import org.apache.flink.connector.pulsar.testutils.source.cases.MultipleTopicsConsumingContext;
import org.apache.flink.connector.pulsar.testutils.source.cases.PartialKeysConsumingContext;
import org.apache.flink.connector.testframe.junit.annotations.TestContext;
import org.apache.flink.connector.testframe.junit.annotations.TestEnv;
@@ -32,17 +32,15 @@ import org.apache.flink.tests.util.pulsar.common.PulsarContainerTestEnvironment;
import org.junit.jupiter.api.Tag;
-/**
- * Pulsar E2E test based on connector testing framework. It's used for Failover & Exclusive
- * subscription.
- */
+import static org.apache.flink.streaming.api.CheckpointingMode.EXACTLY_ONCE;
+
+/** Pulsar source E2E test based on the connector testing framework. */
@SuppressWarnings("unused")
@Tag("org.apache.flink.testutils.junit.FailsOnJava11")
public class PulsarSourceE2ECase extends SourceTestSuiteBase<String> {
// Defines the Semantic.
- @TestSemantics
- CheckpointingMode[] semantics = new CheckpointingMode[] {CheckpointingMode.EXACTLY_ONCE};
+ @TestSemantics CheckpointingMode[] semantics = new CheckpointingMode[] {EXACTLY_ONCE};
// Defines TestEnvironment.
@TestEnv
@@ -54,8 +52,8 @@ public class PulsarSourceE2ECase extends SourceTestSuiteBase<String> {
// Defines a set of external context Factories for different test cases.
@TestContext
- PulsarTestContextFactory<String, MultipleTopicConsumingContext> multipleTopic =
- new PulsarTestContextFactory<>(pulsar, MultipleTopicConsumingContext::new);
+ PulsarTestContextFactory<String, MultipleTopicsConsumingContext> multipleTopic =
+ new PulsarTestContextFactory<>(pulsar, MultipleTopicsConsumingContext::new);
@TestContext
PulsarTestContextFactory<String, PartialKeysConsumingContext> partialKeys =
diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/PulsarSink.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/PulsarSink.java
index b4f7d6e..883c8ea 100644
--- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/PulsarSink.java
+++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/PulsarSink.java
@@ -35,7 +35,7 @@ import org.apache.flink.connector.pulsar.sink.writer.router.RoundRobinTopicRoute
import org.apache.flink.connector.pulsar.sink.writer.router.TopicRouter;
import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
import org.apache.flink.connector.pulsar.sink.writer.serializer.PulsarSerializationSchema;
-import org.apache.flink.connector.pulsar.sink.writer.topic.TopicMetadataListener;
+import org.apache.flink.connector.pulsar.sink.writer.topic.MetadataListener;
import org.apache.flink.core.io.SimpleVersionedSerializer;
import javax.annotation.Nullable;
@@ -85,7 +85,7 @@ public class PulsarSink<IN> implements TwoPhaseCommittingSink<IN, PulsarCommitta
private final SinkConfiguration sinkConfiguration;
private final PulsarSerializationSchema<IN> serializationSchema;
- private final TopicMetadataListener metadataListener;
+ private final MetadataListener metadataListener;
private final TopicRouter<IN> topicRouter;
private final MessageDelayer<IN> messageDelayer;
private final PulsarCrypto pulsarCrypto;
@@ -93,7 +93,7 @@ public class PulsarSink<IN> implements TwoPhaseCommittingSink<IN, PulsarCommitta
PulsarSink(
SinkConfiguration sinkConfiguration,
PulsarSerializationSchema<IN> serializationSchema,
- TopicMetadataListener metadataListener,
+ MetadataListener metadataListener,
TopicRoutingMode topicRoutingMode,
@Nullable TopicRouter<IN> topicRouter,
MessageDelayer<IN> messageDelayer,
diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/PulsarSinkBuilder.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/PulsarSinkBuilder.java
index f394a09..a04093f 100644
--- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/PulsarSinkBuilder.java
+++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/PulsarSinkBuilder.java
@@ -33,7 +33,7 @@ import org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode;
import org.apache.flink.connector.pulsar.sink.writer.serializer.PulsarSchemaWrapper;
import org.apache.flink.connector.pulsar.sink.writer.serializer.PulsarSerializationSchema;
import org.apache.flink.connector.pulsar.sink.writer.serializer.PulsarSerializationSchemaWrapper;
-import org.apache.flink.connector.pulsar.sink.writer.topic.TopicMetadataListener;
+import org.apache.flink.connector.pulsar.sink.writer.topic.MetadataListener;
import org.apache.pulsar.client.api.ProducerCryptoFailureAction;
import org.apache.pulsar.client.api.Schema;
@@ -110,7 +110,7 @@ public class PulsarSinkBuilder<IN> {
private final PulsarConfigBuilder configBuilder;
private PulsarSerializationSchema<IN> serializationSchema;
- private TopicMetadataListener metadataListener;
+ private MetadataListener metadataListener;
private TopicRoutingMode topicRoutingMode;
private TopicRouter<IN> topicRouter;
private MessageDelayer<IN> messageDelayer;
@@ -174,7 +174,7 @@ public class PulsarSinkBuilder<IN> {
checkState(metadataListener == null, "setTopics couldn't be set twice.");
// Making sure the topic should be distinct.
List<String> topicSet = distinctTopics(topics);
- this.metadataListener = new TopicMetadataListener(topicSet);
+ this.metadataListener = new MetadataListener(topicSet);
return this;
}
@@ -447,7 +447,7 @@ public class PulsarSinkBuilder<IN> {
} else {
LOG.warn(
"No topic set has been provided, make sure your custom topic router support empty topic set.");
- this.metadataListener = new TopicMetadataListener();
+ this.metadataListener = new MetadataListener();
}
}
diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriter.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriter.java
index f5a4bb7..0cd51f7 100644
--- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriter.java
+++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriter.java
@@ -34,12 +34,15 @@ import org.apache.flink.connector.pulsar.sink.writer.delayer.MessageDelayer;
import org.apache.flink.connector.pulsar.sink.writer.message.PulsarMessage;
import org.apache.flink.connector.pulsar.sink.writer.router.TopicRouter;
import org.apache.flink.connector.pulsar.sink.writer.serializer.PulsarSerializationSchema;
-import org.apache.flink.connector.pulsar.sink.writer.topic.TopicMetadataListener;
-import org.apache.flink.connector.pulsar.sink.writer.topic.TopicProducerRegister;
+import org.apache.flink.connector.pulsar.sink.writer.topic.MetadataListener;
+import org.apache.flink.connector.pulsar.sink.writer.topic.ProducerRegister;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
+import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.shaded.guava30.com.google.common.base.Strings;
+import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.TypedMessageBuilder;
import org.slf4j.Logger;
@@ -49,6 +52,7 @@ import java.io.IOException;
import java.util.Collection;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicLong;
import static java.util.Collections.emptyList;
@@ -66,12 +70,12 @@ public class PulsarWriter<IN> implements PrecommittingSinkWriter<IN, PulsarCommi
private static final Logger LOG = LoggerFactory.getLogger(PulsarWriter.class);
private final PulsarSerializationSchema<IN> serializationSchema;
- private final TopicMetadataListener metadataListener;
+ private final MetadataListener metadataListener;
private final TopicRouter<IN> topicRouter;
private final MessageDelayer<IN> messageDelayer;
private final DeliveryGuarantee deliveryGuarantee;
private final PulsarSinkContext sinkContext;
- private final TopicProducerRegister producerRegister;
+ private final ProducerRegister producerRegister;
private final MailboxExecutor mailboxExecutor;
private final AtomicLong pendingMessages;
@@ -92,7 +96,7 @@ public class PulsarWriter<IN> implements PrecommittingSinkWriter<IN, PulsarCommi
public PulsarWriter(
SinkConfiguration sinkConfiguration,
PulsarSerializationSchema<IN> serializationSchema,
- TopicMetadataListener metadataListener,
+ MetadataListener metadataListener,
TopicRouter<IN> topicRouter,
MessageDelayer<IN> messageDelayer,
PulsarCrypto pulsarCrypto,
@@ -105,7 +109,8 @@ public class PulsarWriter<IN> implements PrecommittingSinkWriter<IN, PulsarCommi
checkNotNull(initContext);
this.deliveryGuarantee = sinkConfiguration.getDeliveryGuarantee();
- this.sinkContext = new PulsarSinkContextImpl(initContext, sinkConfiguration);
+ this.sinkContext =
+ new PulsarSinkContextImpl(initContext, sinkConfiguration, metadataListener);
// Initialize topic metadata listener.
LOG.debug("Initialize topic metadata after creating Pulsar writer.");
@@ -125,9 +130,8 @@ public class PulsarWriter<IN> implements PrecommittingSinkWriter<IN, PulsarCommi
}
// Create this producer register after opening serialization schema!
- this.producerRegister =
- new TopicProducerRegister(
- sinkConfiguration, pulsarCrypto, initContext.metricGroup());
+ SinkWriterMetricGroup metricGroup = initContext.metricGroup();
+ this.producerRegister = new ProducerRegister(sinkConfiguration, pulsarCrypto, metricGroup);
this.mailboxExecutor = initContext.getMailboxExecutor();
this.pendingMessages = new AtomicLong(0);
}
@@ -138,8 +142,9 @@ public class PulsarWriter<IN> implements PrecommittingSinkWriter<IN, PulsarCommi
// Choose the right topic to send.
String key = message.getKey();
- List<String> availableTopics = metadataListener.availableTopics();
- String topic = topicRouter.route(element, key, availableTopics, sinkContext);
+ List<TopicPartition> partitions = metadataListener.availablePartitions();
+ TopicPartition partition = topicRouter.route(element, key, partitions, sinkContext);
+ String topic = partition.getFullTopicName();
// Create message builder for sending messages.
TypedMessageBuilder<?> builder = createMessageBuilder(topic, context, message);
@@ -157,28 +162,25 @@ public class PulsarWriter<IN> implements PrecommittingSinkWriter<IN, PulsarCommi
} else {
// Increase the pending message count.
pendingMessages.incrementAndGet();
- builder.sendAsync()
- .whenComplete(
- (id, ex) -> {
- pendingMessages.decrementAndGet();
- if (ex != null) {
- mailboxExecutor.execute(
- () -> {
- throw new FlinkRuntimeException(
- "Failed to send data to Pulsar " + topic,
- ex);
- },
- "Failed to send data to Pulsar");
- } else {
- LOG.debug(
- "Sent message to Pulsar {} with message id {}",
- topic,
- id);
- }
- });
+ CompletableFuture<MessageId> future = builder.sendAsync();
+ future.whenComplete(
+ (id, ex) -> {
+ pendingMessages.decrementAndGet();
+ if (ex != null) {
+ mailboxExecutor.execute(
+ () -> throwSendingException(topic, ex),
+ "Failed to send data to Pulsar");
+ } else {
+ LOG.debug("Sent message to Pulsar {} with message id {}", topic, id);
+ }
+ });
}
}
+ private void throwSendingException(String topic, Throwable ex) {
+ throw new FlinkRuntimeException("Failed to send data to Pulsar: " + topic, ex);
+ }
+
@SuppressWarnings({"rawtypes", "unchecked"})
private TypedMessageBuilder<?> createMessageBuilder(
String topic, Context context, PulsarMessage<?> message) {
@@ -209,7 +211,11 @@ public class PulsarWriter<IN> implements PrecommittingSinkWriter<IN, PulsarCommi
// Schema evolution would serialize the message by Pulsar Schema in TypedMessageBuilder.
// The type has been checked in PulsarMessageBuilder#value.
- ((TypedMessageBuilder) builder).value(message.getValue());
+ Object value = message.getValue();
+ if (value == null) {
+ LOG.warn("Send a message with empty payloads, this is a tombstone message in Pulsar.");
+ }
+ ((TypedMessageBuilder) builder).value(value);
Map<String, String> properties = message.getProperties();
if (properties != null && !properties.isEmpty()) {
diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/context/PulsarSinkContext.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/context/PulsarSinkContext.java
index 5c93339..6799677 100644
--- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/context/PulsarSinkContext.java
+++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/context/PulsarSinkContext.java
@@ -19,6 +19,12 @@
package org.apache.flink.connector.pulsar.sink.writer.context;
import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.connector.pulsar.sink.PulsarSinkOptions;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicMetadata;
+
+import org.apache.pulsar.client.admin.PulsarAdminException;
+
+import java.util.Optional;
/** This context provides information on the pulsar record target location. */
@PublicEvolving
@@ -28,7 +34,7 @@ public interface PulsarSinkContext {
* Get the number of the subtask that PulsarSink is running on. The numbering starts from 0 and
* goes up to parallelism-1. (parallelism as returned by {@link #getNumberOfParallelInstances()}
*
- * @return number of subtask
+ * @return index of subtask
*/
int getParallelInstanceId();
@@ -43,4 +49,12 @@ public interface PulsarSinkContext {
/** Returns the current process time in flink. */
long processTime();
+
+ /**
+ * Query the topic metadata from Pulsar. The query result will be cached in {@link
+ * PulsarSinkOptions#PULSAR_TOPIC_METADATA_REFRESH_INTERVAL} interval.
+ *
+ * @return Return {@link Optional#empty()} if the topic doesn't exist.
+ */
+ Optional<TopicMetadata> topicMetadata(String topic) throws PulsarAdminException;
}
diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/context/PulsarSinkContextImpl.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/context/PulsarSinkContextImpl.java
index 681b25a..5df0026 100644
--- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/context/PulsarSinkContextImpl.java
+++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/context/PulsarSinkContextImpl.java
@@ -22,6 +22,12 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.operators.ProcessingTimeService;
import org.apache.flink.api.connector.sink2.Sink.InitContext;
import org.apache.flink.connector.pulsar.sink.config.SinkConfiguration;
+import org.apache.flink.connector.pulsar.sink.writer.topic.MetadataListener;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicMetadata;
+
+import org.apache.pulsar.client.admin.PulsarAdminException;
+
+import java.util.Optional;
/** An implementation that would contain all the required context. */
@Internal
@@ -31,12 +37,17 @@ public class PulsarSinkContextImpl implements PulsarSinkContext {
private final int parallelInstanceId;
private final ProcessingTimeService processingTimeService;
private final boolean enableSchemaEvolution;
+ private final MetadataListener metadataListener;
- public PulsarSinkContextImpl(InitContext initContext, SinkConfiguration sinkConfiguration) {
+ public PulsarSinkContextImpl(
+ InitContext initContext,
+ SinkConfiguration sinkConfiguration,
+ MetadataListener metadataListener) {
this.parallelInstanceId = initContext.getSubtaskId();
this.numberOfParallelSubtasks = initContext.getNumberOfParallelSubtasks();
this.processingTimeService = initContext.getProcessingTimeService();
this.enableSchemaEvolution = sinkConfiguration.isEnableSchemaEvolution();
+ this.metadataListener = metadataListener;
}
@Override
@@ -58,4 +69,9 @@ public class PulsarSinkContextImpl implements PulsarSinkContext {
public long processTime() {
return processingTimeService.getCurrentProcessingTime();
}
+
+ @Override
+ public Optional<TopicMetadata> topicMetadata(String topic) throws PulsarAdminException {
+ return metadataListener.queryTopicMetadata(topic);
+ }
}
diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/message/PulsarMessageBuilder.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/message/PulsarMessageBuilder.java
index 95fb5cb..ac48918 100644
--- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/message/PulsarMessageBuilder.java
+++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/message/PulsarMessageBuilder.java
@@ -40,8 +40,8 @@ public class PulsarMessageBuilder<T> {
private byte[] orderingKey;
private String key;
private long eventTime;
- private final Schema<T> schema;
- private final T value;
+ @Nullable private final Schema<T> schema;
+ @Nullable private final T value;
private final Map<String, String> properties = new HashMap<>();
private Long sequenceId;
private List<String> replicationClusters;
@@ -51,7 +51,7 @@ public class PulsarMessageBuilder<T> {
* Make this constructor package private for in favor of the {@link PulsarMessage#builder()}
* method in {@link PulsarMessage}.
*/
- PulsarMessageBuilder(Schema<T> schema, @Nullable T value) {
+ PulsarMessageBuilder(@Nullable Schema<T> schema, @Nullable T value) {
this.schema = schema;
this.value = value;
}
diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/router/KeyHashTopicRouter.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/router/KeyHashTopicRouter.java
index 433d79c..0405bb3 100644
--- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/router/KeyHashTopicRouter.java
+++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/router/KeyHashTopicRouter.java
@@ -21,6 +21,7 @@ package org.apache.flink.connector.pulsar.sink.writer.router;
import org.apache.flink.annotation.Internal;
import org.apache.flink.connector.pulsar.sink.config.SinkConfiguration;
import org.apache.flink.connector.pulsar.sink.writer.context.PulsarSinkContext;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
import org.apache.flink.shaded.guava30.com.google.common.base.Strings;
@@ -50,7 +51,8 @@ public class KeyHashTopicRouter<IN> implements TopicRouter<IN> {
}
@Override
- public String route(IN in, String key, List<String> partitions, PulsarSinkContext context) {
+ public TopicPartition route(
+ IN in, String key, List<TopicPartition> partitions, PulsarSinkContext context) {
checkArgument(
!partitions.isEmpty(),
"You should provide topics for routing topic by message key hash.");
diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/router/RoundRobinTopicRouter.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/router/RoundRobinTopicRouter.java
index 65e5154..4bb27e9 100644
--- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/router/RoundRobinTopicRouter.java
+++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/router/RoundRobinTopicRouter.java
@@ -21,6 +21,7 @@ package org.apache.flink.connector.pulsar.sink.writer.router;
import org.apache.flink.annotation.Internal;
import org.apache.flink.connector.pulsar.sink.config.SinkConfiguration;
import org.apache.flink.connector.pulsar.sink.writer.context.PulsarSinkContext;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
import org.apache.flink.util.Preconditions;
import java.util.List;
@@ -50,7 +51,8 @@ public class RoundRobinTopicRouter<IN> implements TopicRouter<IN> {
}
@Override
- public String route(IN in, String key, List<String> partitions, PulsarSinkContext context) {
+ public TopicPartition route(
+ IN in, String key, List<TopicPartition> partitions, PulsarSinkContext context) {
checkArgument(
!partitions.isEmpty(),
"You should provide topics for routing topic by message key hash.");
diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/router/TopicRouter.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/router/TopicRouter.java
index a2c0589..0b6672c 100644
--- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/router/TopicRouter.java
+++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/router/TopicRouter.java
@@ -24,6 +24,7 @@ import org.apache.flink.connector.pulsar.sink.config.SinkConfiguration;
import org.apache.flink.connector.pulsar.sink.writer.context.PulsarSinkContext;
import org.apache.flink.connector.pulsar.sink.writer.message.PulsarMessageBuilder;
import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
import java.io.Serializable;
import java.util.List;
@@ -55,7 +56,8 @@ public interface TopicRouter<IN> extends Serializable {
* @param context The context contains useful information for determining the topic.
* @return The topic name to use.
*/
- String route(IN in, String key, List<String> partitions, PulsarSinkContext context);
+ TopicPartition route(
+ IN in, String key, List<TopicPartition> partitions, PulsarSinkContext context);
/** Implement this method if you have some non-serializable field. */
default void open(SinkConfiguration sinkConfiguration) {
diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/topic/MetadataListener.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/topic/MetadataListener.java
new file mode 100644
index 0000000..bbf909c
--- /dev/null
+++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/topic/MetadataListener.java
@@ -0,0 +1,235 @@
+/*
+ * 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.sink.writer.topic;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.operators.ProcessingTimeService;
+import org.apache.flink.connector.pulsar.sink.PulsarSinkOptions;
+import org.apache.flink.connector.pulsar.sink.config.SinkConfiguration;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicMetadata;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import org.apache.flink.shaded.guava30.com.google.common.cache.CacheBuilder;
+import org.apache.flink.shaded.guava30.com.google.common.cache.CacheLoader;
+import org.apache.flink.shaded.guava30.com.google.common.cache.LoadingCache;
+import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList;
+
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.admin.PulsarAdminException.NotFoundException;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.ParametersAreNonnullByDefault;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+
+import static java.util.Collections.emptyList;
+import static org.apache.flink.connector.pulsar.common.config.PulsarClientFactory.createAdmin;
+import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils.isPartition;
+import static org.apache.pulsar.common.partition.PartitionedTopicMetadata.NON_PARTITIONED;
+
+/**
+ * We need the latest topic metadata for making sure the newly created topic partitions would be
+ * used by the Pulsar sink. This routing policy would be different compared with Pulsar Client
+ * built-in logic. We use Flink's ProcessingTimer as the executor.
+ */
+@Internal
+public class MetadataListener implements Serializable, Closeable {
+ private static final long serialVersionUID = 6186948471557507522L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(MetadataListener.class);
+
+ private final ImmutableList<String> partitions;
+ private final ImmutableList<String> topics;
+ private ImmutableList<TopicPartition> availablePartitions;
+
+ // Dynamic fields.
+ private transient PulsarAdmin pulsarAdmin;
+ private transient Long topicMetadataRefreshInterval;
+ private transient ProcessingTimeService timeService;
+ private transient LoadingCache<String, Optional<Integer>> topicPartitionCache;
+
+ public MetadataListener() {
+ this(emptyList());
+ }
+
+ public MetadataListener(List<String> topics) {
+ ImmutableList.Builder<String> partitionsBuilder = ImmutableList.builder();
+ ImmutableList.Builder<String> topicsBuilder = ImmutableList.builder();
+
+ for (String topic : topics) {
+ if (isPartition(topic)) {
+ partitionsBuilder.add(topic);
+ } else {
+ topicsBuilder.add(topic);
+ }
+ }
+
+ this.partitions = partitionsBuilder.build();
+ this.topics = topicsBuilder.build();
+ this.availablePartitions = ImmutableList.of();
+ }
+
+ /** Register the topic metadata update action in process time service. */
+ public void open(SinkConfiguration sinkConfiguration, ProcessingTimeService timeService) {
+ // Initialize listener properties.
+ this.pulsarAdmin = createAdmin(sinkConfiguration);
+ this.topicMetadataRefreshInterval = sinkConfiguration.getTopicMetadataRefreshInterval();
+ this.timeService = timeService;
+ this.topicPartitionCache =
+ CacheBuilder.newBuilder()
+ .expireAfterWrite(topicMetadataRefreshInterval, TimeUnit.MILLISECONDS)
+ .build(
+ new CacheLoader<String, Optional<Integer>>() {
+ @Override
+ @ParametersAreNonnullByDefault
+ public Optional<Integer> load(String topic)
+ throws PulsarAdminException {
+ try {
+ PartitionedTopicMetadata metadata =
+ pulsarAdmin
+ .topics()
+ .getPartitionedTopicMetadata(topic);
+ return Optional.of(metadata.partitions);
+ } catch (NotFoundException e) {
+ return Optional.empty();
+ }
+ }
+ });
+
+ // Initialize the topic metadata. Quit if fail to connect to Pulsar.
+ try {
+ updateTopicMetadata();
+ } catch (PulsarAdminException e) {
+ throw new FlinkRuntimeException(e);
+ }
+
+ // Register time service for update the topic metadata.
+ if (topics.isEmpty()) {
+ LOG.info("No topics have been provided, skip metadata update timer.");
+ } else {
+ registerNextTopicMetadataUpdateTimer();
+ }
+ }
+
+ /**
+ * Return all the available topic partitions. We would recalculate the partitions if the topic
+ * metadata has been changed. Otherwise, we would return the cached result for better
+ * performance.
+ */
+ public List<TopicPartition> availablePartitions() {
+ return availablePartitions;
+ }
+
+ /**
+ * Query the topic metadata from Pulsar. The query result will be cached in {@link
+ * PulsarSinkOptions#PULSAR_TOPIC_METADATA_REFRESH_INTERVAL} interval.
+ *
+ * @return Return {@link Optional#empty()} if the topic doesn't exist.
+ */
+ public Optional<TopicMetadata> queryTopicMetadata(String topic) throws PulsarAdminException {
+ if (isPartition(topic)) {
+ return Optional.of(new TopicMetadata(topic, NON_PARTITIONED));
+ }
+
+ try {
+ return topicPartitionCache.get(topic).map(size -> new TopicMetadata(topic, size));
+ } catch (ExecutionException e) {
+ Optional<PulsarAdminException> optional =
+ ExceptionUtils.findThrowable(e, PulsarAdminException.class);
+ if (optional.isPresent()) {
+ throw optional.get();
+ } else {
+ throw new FlinkRuntimeException(e);
+ }
+ }
+ }
+
+ @VisibleForTesting
+ void refreshTopicMetadata(String topic) {
+ topicPartitionCache.refresh(topic);
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (pulsarAdmin != null) {
+ pulsarAdmin.close();
+ }
+ }
+
+ private void registerNextTopicMetadataUpdateTimer() {
+ long currentProcessingTime = timeService.getCurrentProcessingTime();
+ long triggerTime = currentProcessingTime + topicMetadataRefreshInterval;
+
+ timeService.registerTimer(triggerTime, time -> triggerNextTopicMetadataUpdate());
+ }
+
+ private void triggerNextTopicMetadataUpdate() {
+ // Try to update the topic metadata, ignore the pulsar admin exception.
+ try {
+ updateTopicMetadata();
+ } catch (PulsarAdminException e) {
+ LOG.warn("", e);
+ }
+
+ // Register next timer.
+ registerNextTopicMetadataUpdateTimer();
+ }
+
+ private void updateTopicMetadata() throws PulsarAdminException {
+ ImmutableList.Builder<TopicPartition> parititonsBuilder = ImmutableList.builder();
+
+ for (String topic : topics) {
+ Optional<TopicMetadata> optionalMetadata = queryTopicMetadata(topic);
+ if (optionalMetadata.isPresent()) {
+ TopicMetadata metadata = optionalMetadata.get();
+ int partitionSize = metadata.getPartitionSize();
+ if (metadata.isPartitioned()) {
+ for (int i = 0; i < partitionSize; i++) {
+ parititonsBuilder.add(new TopicPartition(topic, i));
+ }
+ } else {
+ parititonsBuilder.add(new TopicPartition(topic));
+ }
+ }
+ }
+
+ for (String partition : partitions) {
+ TopicName topicName = TopicName.get(partition);
+ String name = topicName.getPartitionedTopicName();
+ int index = topicName.getPartitionIndex();
+
+ parititonsBuilder.add(new TopicPartition(name, index));
+ }
+
+ this.availablePartitions = parititonsBuilder.build();
+ }
+}
diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/topic/TopicProducerRegister.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/topic/ProducerRegister.java
similarity index 93%
rename from flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/topic/TopicProducerRegister.java
rename to flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/topic/ProducerRegister.java
index 3b32467..f57faca 100644
--- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/topic/TopicProducerRegister.java
+++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/topic/ProducerRegister.java
@@ -44,10 +44,12 @@ import org.apache.pulsar.client.api.transaction.TransactionCoordinatorClient;
import org.apache.pulsar.client.api.transaction.TxnID;
import org.apache.pulsar.client.impl.ProducerBase;
import org.apache.pulsar.client.impl.ProducerBuilderImpl;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.pulsar.client.impl.TypedMessageBuilderImpl;
import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
import org.apache.pulsar.client.impl.transaction.TransactionImpl;
import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.protocol.schema.SchemaHash;
import org.apache.pulsar.common.schema.SchemaType;
import org.apache.pulsar.shade.com.google.common.base.Strings;
@@ -62,6 +64,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.ExecutionException;
import static org.apache.flink.connector.pulsar.common.config.PulsarClientFactory.createClient;
import static org.apache.flink.connector.pulsar.common.metrics.MetricNames.NUM_ACKS_RECEIVED;
@@ -92,7 +95,10 @@ import static org.apache.flink.connector.pulsar.sink.config.PulsarSinkConfigUtil
* we have to create different instances for different topics.
*/
@Internal
-public class TopicProducerRegister implements Closeable {
+public class ProducerRegister implements Closeable {
+
+ private static final String FAIL_TO_CREATE_TOPIC =
+ "Fail to create the non-exist topic, make sure you have enable the topic auto creation in Pulsar.";
private final PulsarClient pulsarClient;
@Nullable private final TransactionCoordinatorClient coordinatorClient;
@@ -103,7 +109,7 @@ public class TopicProducerRegister implements Closeable {
private final Map<String, Map<SchemaHash, Producer<?>>> producers;
private final Map<String, Transaction> transactions;
- public TopicProducerRegister(
+ public ProducerRegister(
SinkConfiguration sinkConfiguration,
PulsarCrypto pulsarCrypto,
SinkWriterMetricGroup metricGroup) {
@@ -210,6 +216,21 @@ public class TopicProducerRegister implements Closeable {
return (Producer<T>) set.get(hash);
}
+ try {
+ // Use this method for auto creating the non-exist topics. Otherwise, it will throw an
+ // exception.
+ TopicName topicName = TopicName.get(topic);
+ ((PulsarClientImpl) pulsarClient)
+ .getLookup()
+ .getPartitionedTopicMetadata(topicName)
+ .get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new FlinkRuntimeException(FAIL_TO_CREATE_TOPIC, e);
+ } catch (ExecutionException e) {
+ throw new FlinkRuntimeException(FAIL_TO_CREATE_TOPIC, e);
+ }
+
ProducerBuilder<T> builder = createProducerBuilder(pulsarClient, schema, sinkConfiguration);
// Enable end-to-end encryption if provided.
diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/topic/TopicMetadataListener.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/topic/TopicMetadataListener.java
deleted file mode 100644
index 7db42e6..0000000
--- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/topic/TopicMetadataListener.java
+++ /dev/null
@@ -1,181 +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.sink.writer.topic;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.operators.ProcessingTimeService;
-import org.apache.flink.connector.pulsar.sink.config.SinkConfiguration;
-
-import org.apache.flink.shaded.guava30.com.google.common.base.Objects;
-import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList;
-
-import org.apache.pulsar.client.admin.PulsarAdmin;
-import org.apache.pulsar.client.admin.PulsarAdminException;
-import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.Closeable;
-import java.io.IOException;
-import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import static java.util.Collections.emptyList;
-import static org.apache.flink.connector.pulsar.common.config.PulsarClientFactory.createAdmin;
-import static org.apache.flink.connector.pulsar.common.utils.PulsarExceptionUtils.sneakyAdmin;
-import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils.isPartition;
-import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils.topicName;
-import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils.topicNameWithPartition;
-import static org.apache.pulsar.common.partition.PartitionedTopicMetadata.NON_PARTITIONED;
-
-/**
- * We need the latest topic metadata for making sure the newly created topic partitions would be
- * used by the Pulsar sink. This routing policy would be different compared with Pulsar Client
- * built-in logic. We use Flink's ProcessingTimer as the executor.
- */
-@Internal
-public class TopicMetadataListener implements Serializable, Closeable {
- private static final long serialVersionUID = 6186948471557507522L;
-
- private static final Logger LOG = LoggerFactory.getLogger(TopicMetadataListener.class);
-
- private final ImmutableList<String> partitionedTopics;
- private final Map<String, Integer> topicMetadata;
- private volatile ImmutableList<String> availableTopics;
-
- // Dynamic fields.
- private transient PulsarAdmin pulsarAdmin;
- private transient Long topicMetadataRefreshInterval;
- private transient ProcessingTimeService timeService;
-
- public TopicMetadataListener() {
- this(emptyList());
- }
-
- public TopicMetadataListener(List<String> topics) {
- List<String> partitions = new ArrayList<>(topics.size());
- Map<String, Integer> metadata = new HashMap<>(topics.size());
- for (String topic : topics) {
- if (isPartition(topic)) {
- partitions.add(topic);
- } else {
- // This would be updated when open writing.
- metadata.put(topic, -1);
- }
- }
-
- this.partitionedTopics = ImmutableList.copyOf(partitions);
- this.topicMetadata = metadata;
- this.availableTopics = ImmutableList.of();
- }
-
- /** Register the topic metadata update in process time service. */
- public void open(SinkConfiguration sinkConfiguration, ProcessingTimeService timeService) {
- if (topicMetadata.isEmpty()) {
- LOG.info("No topics have been provided, skip listener initialize.");
- return;
- }
-
- // Initialize listener properties.
- this.pulsarAdmin = createAdmin(sinkConfiguration);
- this.topicMetadataRefreshInterval = sinkConfiguration.getTopicMetadataRefreshInterval();
- this.timeService = timeService;
-
- // Initialize the topic metadata. Quit if fail to connect to Pulsar.
- sneakyAdmin(this::updateTopicMetadata);
-
- // Register time service.
- triggerNextTopicMetadataUpdate(true);
- }
-
- /**
- * Return all the available topic partitions. We would recalculate the partitions if the topic
- * metadata has been changed. Otherwise, we would return the cached result for better
- * performance.
- */
- public List<String> availableTopics() {
- if (availableTopics.isEmpty()
- && (!partitionedTopics.isEmpty() || !topicMetadata.isEmpty())) {
- List<String> results = new ArrayList<>();
- for (Map.Entry<String, Integer> entry : topicMetadata.entrySet()) {
- int partitionNums = entry.getValue();
- // Get all topics from partitioned and non-partitioned topic names
- if (partitionNums == NON_PARTITIONED) {
- results.add(topicName(entry.getKey()));
- } else {
- for (int i = 0; i < partitionNums; i++) {
- results.add(topicNameWithPartition(entry.getKey(), i));
- }
- }
- }
-
- results.addAll(partitionedTopics);
- this.availableTopics = ImmutableList.copyOf(results);
- }
-
- return availableTopics;
- }
-
- @Override
- public void close() throws IOException {
- if (pulsarAdmin != null) {
- pulsarAdmin.close();
- }
- }
-
- private void triggerNextTopicMetadataUpdate(boolean initial) {
- if (!initial) {
- // We should update the topic metadata, ignore the pulsar admin exception.
- try {
- updateTopicMetadata();
- } catch (PulsarAdminException e) {
- LOG.warn("", e);
- }
- }
-
- // Register next timer.
- long currentProcessingTime = timeService.getCurrentProcessingTime();
- long triggerTime = currentProcessingTime + topicMetadataRefreshInterval;
- timeService.registerTimer(triggerTime, time -> triggerNextTopicMetadataUpdate(false));
- }
-
- private void updateTopicMetadata() throws PulsarAdminException {
- boolean shouldUpdate = false;
-
- for (Map.Entry<String, Integer> entry : topicMetadata.entrySet()) {
- String topic = entry.getKey();
- PartitionedTopicMetadata metadata =
- pulsarAdmin.topics().getPartitionedTopicMetadata(topic);
-
- // Update topic metadata if it has been changed.
- if (!Objects.equal(entry.getValue(), metadata.partitions)) {
- entry.setValue(metadata.partitions);
- shouldUpdate = true;
- }
- }
-
- // Clear available topics if the topic metadata has been changed.
- if (shouldUpdate) {
- this.availableTopics = ImmutableList.of();
- }
- }
-}
diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java
index fa42ccf..0f2e8f0 100644
--- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java
+++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java
@@ -76,7 +76,7 @@ public final class PulsarSourceOptions {
public static final ConfigOption<Long> PULSAR_PARTITION_DISCOVERY_INTERVAL_MS =
ConfigOptions.key(SOURCE_CONFIG_PREFIX + "partitionDiscoveryIntervalMs")
.longType()
- .defaultValue(Duration.ofSeconds(30).toMillis())
+ .defaultValue(Duration.ofMinutes(5).toMillis())
.withDescription(
Description.builder()
.text(
diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/PulsarSourceEnumStateSerializer.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/PulsarSourceEnumStateSerializer.java
index f7a1edc..5995d27 100644
--- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/PulsarSourceEnumStateSerializer.java
+++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/PulsarSourceEnumStateSerializer.java
@@ -85,8 +85,8 @@ public class PulsarSourceEnumStateSerializer
// Only deserialize these fields for backward compatibility.
if (version == 0) {
- deserializeSet(in, deserializeSplit(0));
- deserializeMap(in, DataInput::readInt, i -> deserializeSet(i, deserializeSplit(0)));
+ deserializeSet(in, deserializeSplit());
+ deserializeMap(in, DataInput::readInt, i -> deserializeSet(i, deserializeSplit()));
deserializeMap(in, DataInput::readInt, i -> deserializeSet(i, DataInput::readUTF));
in.readBoolean();
}
@@ -103,7 +103,7 @@ public class PulsarSourceEnumStateSerializer
}
private FunctionWithException<DataInputStream, PulsarPartitionSplit, IOException>
- deserializeSplit(int version) {
- return in -> SPLIT_SERIALIZER.deserializePulsarPartitionSplit(version, in);
+ deserializeSplit() {
+ return in -> SPLIT_SERIALIZER.deserializePulsarPartitionSplit(0, in);
}
}
diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/impl/BasePulsarSubscriber.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/impl/BasePulsarSubscriber.java
index 152657b..a206a25 100644
--- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/impl/BasePulsarSubscriber.java
+++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/impl/BasePulsarSubscriber.java
@@ -59,7 +59,7 @@ public abstract class BasePulsarSubscriber implements PulsarSubscriber {
TopicMetadata metadata, List<TopicRange> ranges) {
if (!metadata.isPartitioned()) {
// For non-partitioned topic.
- return singletonList(new TopicPartition(metadata.getName(), -1, ranges));
+ return singletonList(new TopicPartition(metadata.getName(), ranges));
} else {
// For partitioned topic.
List<TopicPartition> partitions = new ArrayList<>();
diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicPartition.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicPartition.java
index b851cb2..d6b98bd 100644
--- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicPartition.java
+++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicPartition.java
@@ -21,6 +21,7 @@ package org.apache.flink.connector.pulsar.source.enumerator.topic;
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRouter;
import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList;
@@ -35,16 +36,23 @@ import static java.util.stream.Collectors.toList;
import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils.topicName;
import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils.topicNameWithPartition;
import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange.createFullRange;
+import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
/**
- * Topic partition is the basic topic information used by {@link SplitReader}, we create this topic
- * metas for a specified topic by subscription type and convert it into a partition split.
+ * Topic partition is the basic topic information used by {@link SplitReader} and {@link
+ * TopicRouter}.
*/
@PublicEvolving
public class TopicPartition implements Serializable {
private static final long serialVersionUID = -1474354741550810953L;
+ /**
+ * If {@link TopicPartition#getPartitionId()} is equal to this. This topic partition wouldn't be
+ * a partition instance. It would be a non-partitioned topic name.
+ */
+ public static final int NON_PARTITION_ID = -1;
+
private static final List<TopicRange> FULL_RANGES = ImmutableList.of(createFullRange());
/**
@@ -66,11 +74,28 @@ public class TopicPartition implements Serializable {
*/
private final List<TopicRange> ranges;
+ /** Create a non-partition topic without partition information. */
+ @PublicEvolving
+ public TopicPartition(String topic) {
+ this(topic, NON_PARTITION_ID, FULL_RANGES);
+ }
+
+ /** Create a topic partition without key hash range. */
+ @PublicEvolving
public TopicPartition(String topic, int partitionId) {
this(topic, partitionId, FULL_RANGES);
}
+ /** Create a non-partition topic with key hash range. */
+ @Internal
+ public TopicPartition(String topic, List<TopicRange> ranges) {
+ this(topic, NON_PARTITION_ID, ranges);
+ }
+
+ @Internal
public TopicPartition(String topic, int partitionId, List<TopicRange> ranges) {
+ checkArgument(partitionId >= NON_PARTITION_ID, "Invalid partition id.");
+
this.topic = topicName(checkNotNull(topic));
this.partitionId = partitionId;
this.ranges = checkNotNull(ranges);
@@ -84,12 +109,16 @@ public class TopicPartition implements Serializable {
return partitionId;
}
+ public boolean isPartition() {
+ return partitionId != NON_PARTITION_ID;
+ }
+
/**
* Pulsar split the topic partition into a bunch of small topics, we would get the real topic
* name by using this method.
*/
public String getFullTopicName() {
- if (partitionId >= 0) {
+ if (isPartition()) {
return topicNameWithPartition(topic, partitionId);
} else {
return topic;
diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicRange.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicRange.java
index 1508b87..cef4a08 100644
--- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicRange.java
+++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicRange.java
@@ -48,6 +48,9 @@ public class TopicRange implements Serializable {
/** The end position for hash range, it's 65535. */
public static final int MAX_RANGE = RANGE_SIZE - 1;
+ /** A full-range instance for avoiding multiple instance creation. */
+ private static final TopicRange FULL_RANGE = new TopicRange(MIN_RANGE, MAX_RANGE);
+
/** The start of the range, default is zero. */
private final int start;
@@ -63,14 +66,14 @@ public class TopicRange implements Serializable {
this.end = end;
}
- /** Convert to pulsar's {@link Range} API for consuming in client. */
+ /** Convert to pulsar's {@link Range} API for consuming in the client. */
public Range toPulsarRange() {
return new Range(start, end);
}
/** Create a topic range which contains the full hash range. */
public static TopicRange createFullRange() {
- return new TopicRange(MIN_RANGE, MAX_RANGE);
+ return FULL_RANGE;
}
public int getStart() {
diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/split/PulsarPartitionSplitSerializer.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/split/PulsarPartitionSplitSerializer.java
index 2903bf9..1c32242 100644
--- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/split/PulsarPartitionSplitSerializer.java
+++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/split/PulsarPartitionSplitSerializer.java
@@ -62,7 +62,6 @@ public class PulsarPartitionSplitSerializer
@Override
public byte[] serialize(PulsarPartitionSplit obj) throws IOException {
- // VERSION 0 serialization
try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
DataOutputStream out = new DataOutputStream(baos)) {
serializePulsarPartitionSplit(out, obj);
@@ -73,7 +72,6 @@ public class PulsarPartitionSplitSerializer
@Override
public PulsarPartitionSplit deserialize(int version, byte[] serialized) throws IOException {
- // VERSION 0 deserialization
try (ByteArrayInputStream bais = new ByteArrayInputStream(serialized);
DataInputStream in = new DataInputStream(bais)) {
return deserializePulsarPartitionSplit(version, in);
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/PulsarSinkITCase.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/PulsarSinkITCase.java
index 6942034..6ce020e 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/PulsarSinkITCase.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/PulsarSinkITCase.java
@@ -25,8 +25,10 @@ import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
import org.apache.flink.connector.pulsar.testutils.PulsarTestSuiteBase;
import org.apache.flink.connector.pulsar.testutils.function.ControlSource;
import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntime;
-import org.apache.flink.connector.pulsar.testutils.sink.cases.PulsarEncryptSinkContext;
-import org.apache.flink.connector.pulsar.testutils.sink.cases.PulsarSinkTestContext;
+import org.apache.flink.connector.pulsar.testutils.sink.cases.AutoCreateTopicProducingContext;
+import org.apache.flink.connector.pulsar.testutils.sink.cases.EncryptedMessageProducingContext;
+import org.apache.flink.connector.pulsar.testutils.sink.cases.MultipleTopicsProducingContext;
+import org.apache.flink.connector.pulsar.testutils.sink.cases.SingleTopicProducingContext;
import org.apache.flink.connector.testframe.environment.MiniClusterTestEnvironment;
import org.apache.flink.connector.testframe.junit.annotations.TestContext;
import org.apache.flink.connector.testframe.junit.annotations.TestEnv;
@@ -51,6 +53,8 @@ import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
+import static org.apache.flink.streaming.api.CheckpointingMode.AT_LEAST_ONCE;
+import static org.apache.flink.streaming.api.CheckpointingMode.EXACTLY_ONCE;
import static org.apache.pulsar.client.api.Schema.STRING;
import static org.assertj.core.api.Assertions.assertThat;
@@ -68,18 +72,23 @@ class PulsarSinkITCase {
PulsarTestEnvironment pulsar = new PulsarTestEnvironment(PulsarRuntime.container());
@TestSemantics
- CheckpointingMode[] semantics =
- new CheckpointingMode[] {
- CheckpointingMode.EXACTLY_ONCE, CheckpointingMode.AT_LEAST_ONCE
- };
+ CheckpointingMode[] semantics = new CheckpointingMode[] {AT_LEAST_ONCE, EXACTLY_ONCE};
@TestContext
- PulsarTestContextFactory<String, PulsarSinkTestContext> sinkContext =
- new PulsarTestContextFactory<>(pulsar, PulsarSinkTestContext::new);
+ PulsarTestContextFactory<String, SingleTopicProducingContext> singleTopic =
+ new PulsarTestContextFactory<>(pulsar, SingleTopicProducingContext::new);
@TestContext
- PulsarTestContextFactory<String, PulsarEncryptSinkContext> encryptMessages =
- new PulsarTestContextFactory<>(pulsar, PulsarEncryptSinkContext::new);
+ PulsarTestContextFactory<String, MultipleTopicsProducingContext> multipleTopics =
+ new PulsarTestContextFactory<>(pulsar, MultipleTopicsProducingContext::new);
+
+ @TestContext
+ PulsarTestContextFactory<String, AutoCreateTopicProducingContext> topicAutoCreation =
+ new PulsarTestContextFactory<>(pulsar, AutoCreateTopicProducingContext::new);
+
+ @TestContext
+ PulsarTestContextFactory<String, EncryptedMessageProducingContext> encryptMessages =
+ new PulsarTestContextFactory<>(pulsar, EncryptedMessageProducingContext::new);
}
/** Tests for using PulsarSink writing to a Pulsar cluster. */
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriterTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriterTest.java
index d7a64fa..422d5e3 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriterTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/PulsarWriterTest.java
@@ -28,12 +28,15 @@ import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.pulsar.common.crypto.PulsarCrypto;
import org.apache.flink.connector.pulsar.sink.committer.PulsarCommittable;
import org.apache.flink.connector.pulsar.sink.config.SinkConfiguration;
+import org.apache.flink.connector.pulsar.sink.writer.context.PulsarSinkContext;
import org.apache.flink.connector.pulsar.sink.writer.delayer.FixedMessageDelayer;
import org.apache.flink.connector.pulsar.sink.writer.delayer.MessageDelayer;
import org.apache.flink.connector.pulsar.sink.writer.router.RoundRobinTopicRouter;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRouter;
import org.apache.flink.connector.pulsar.sink.writer.serializer.PulsarSchemaWrapper;
import org.apache.flink.connector.pulsar.sink.writer.serializer.PulsarSerializationSchema;
-import org.apache.flink.connector.pulsar.sink.writer.topic.TopicMetadataListener;
+import org.apache.flink.connector.pulsar.sink.writer.topic.MetadataListener;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
import org.apache.flink.connector.pulsar.testutils.PulsarTestSuiteBase;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.metrics.groups.OperatorIOMetricGroup;
@@ -46,10 +49,12 @@ import org.apache.flink.streaming.runtime.tasks.TestProcessingTimeService;
import org.apache.flink.util.UserCodeClassLoader;
import org.apache.pulsar.client.api.transaction.TransactionCoordinatorClient;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;
import java.util.Collection;
+import java.util.List;
import java.util.OptionalLong;
import static java.util.Collections.singletonList;
@@ -67,14 +72,27 @@ class PulsarWriterTest extends PulsarTestSuiteBase {
@ParameterizedTest
@EnumSource(DeliveryGuarantee.class)
- void writeMessages(DeliveryGuarantee guarantee) throws Exception {
- String topic = randomAlphabetic(10);
+ void writeMessagesToPulsar(DeliveryGuarantee guarantee) throws Exception {
+ String topic = "writer-" + randomAlphabetic(10);
operator().createTopic(topic, 8);
+ MetadataListener listener = new MetadataListener(singletonList(topic));
+ writeMessageAndVerify(guarantee, listener, topic);
+ }
+
+ @Test
+ void writeMessagesToPulsarWithTopicAutoCreation() throws Exception {
+ String topic = "non-existed-topic-" + randomAlphabetic(10);
+ MetadataListener listener = new MetadataListener();
+
+ writeMessageAndVerify(DeliveryGuarantee.AT_LEAST_ONCE, listener, topic);
+ }
+
+ private void writeMessageAndVerify(
+ DeliveryGuarantee guarantee, MetadataListener listener, String topic) throws Exception {
SinkConfiguration configuration = sinkConfiguration(guarantee);
+ TopicRouter<String> router = new DynamicTopicRouter<>(configuration, topic);
PulsarSerializationSchema<String> schema = new PulsarSchemaWrapper<>(STRING);
- TopicMetadataListener listener = new TopicMetadataListener(singletonList(topic));
- RoundRobinTopicRouter<String> router = new RoundRobinTopicRouter<>(configuration);
FixedMessageDelayer<String> delayer = MessageDelayer.never();
MockInitContext initContext = new MockInitContext();
@@ -119,6 +137,28 @@ class PulsarWriterTest extends PulsarTestSuiteBase {
return new SinkConfiguration(configuration);
}
+ private static class DynamicTopicRouter<IN> implements TopicRouter<IN> {
+ private static final long serialVersionUID = -2855742316603280628L;
+
+ private final RoundRobinTopicRouter<IN> robinTopicRouter;
+ private final String topic;
+
+ public DynamicTopicRouter(SinkConfiguration configuration, String topic) {
+ this.robinTopicRouter = new RoundRobinTopicRouter<>(configuration);
+ this.topic = topic;
+ }
+
+ @Override
+ public TopicPartition route(
+ IN in, String key, List<TopicPartition> partitions, PulsarSinkContext context) {
+ if (partitions.isEmpty()) {
+ return new TopicPartition(topic);
+ } else {
+ return robinTopicRouter.route(in, key, partitions, context);
+ }
+ }
+ }
+
private static class MockInitContext implements InitContext {
private final MetricListener metricListener;
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/router/KeyHashTopicRouterTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/router/KeyHashTopicRouterTest.java
index a136aa5..dac073e 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/router/KeyHashTopicRouterTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/router/KeyHashTopicRouterTest.java
@@ -21,6 +21,7 @@ package org.apache.flink.connector.pulsar.sink.writer.router;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.pulsar.sink.config.SinkConfiguration;
import org.apache.flink.connector.pulsar.sink.writer.context.PulsarSinkContext;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
import org.apache.pulsar.client.impl.Hash;
import org.junit.jupiter.params.ParameterizedTest;
@@ -50,33 +51,34 @@ class KeyHashTopicRouterTest {
String message = randomAlphanumeric(10);
String key = randomAlphanumeric(10);
- List<String> emptyTopics = emptyList();
+ List<TopicPartition> emptyPartitions = emptyList();
PulsarSinkContext sinkContext = mock(PulsarSinkContext.class);
assertThrows(
IllegalArgumentException.class,
- () -> router.route(message, key, emptyTopics, sinkContext));
+ () -> router.route(message, key, emptyPartitions, sinkContext));
}
@ParameterizedTest
@EnumSource(MessageKeyHash.class)
void routeOnlyOnePartition(MessageKeyHash keyHash) {
SinkConfiguration configuration = sinkConfiguration(keyHash);
- List<String> topics = singletonList(randomAlphanumeric(10));
+ List<TopicPartition> partitions = singletonList(new TopicPartition(randomAlphanumeric(10)));
KeyHashTopicRouter<String> router1 = new KeyHashTopicRouter<>(configuration);
- String topic1 =
+ TopicPartition partition1 =
router1.route(
randomAlphanumeric(10),
randomAlphanumeric(10),
- topics,
+ partitions,
mock(PulsarSinkContext.class));
- assertEquals(topic1, topics.get(0));
+ assertEquals(partition1, partitions.get(0));
KeyHashTopicRouter<String> router2 = new KeyHashTopicRouter<>(configuration);
- String topic2 =
- router2.route(randomAlphanumeric(10), null, topics, mock(PulsarSinkContext.class));
- assertEquals(topic2, topics.get(0));
+ TopicPartition partition2 =
+ router2.route(
+ randomAlphanumeric(10), null, partitions, mock(PulsarSinkContext.class));
+ assertEquals(partition2, partitions.get(0));
}
@ParameterizedTest
@@ -86,20 +88,21 @@ class KeyHashTopicRouterTest {
String messageKey = randomAlphanumeric(10);
KeyHashTopicRouter<String> router = new KeyHashTopicRouter<>(configuration);
- List<String> topics =
- Stream.generate(() -> randomAlphanumeric(10))
+ List<TopicPartition> partitions =
+ Stream.generate(() -> new TopicPartition(randomAlphanumeric(10)))
.distinct()
.limit(10)
.collect(toList());
Hash hash = keyHash.getHash();
- int index = signSafeMod(hash.makeHash(messageKey), topics.size());
- String desiredTopic = topics.get(index);
+ int index = signSafeMod(hash.makeHash(messageKey), partitions.size());
+ TopicPartition desiredPartitions = partitions.get(index);
String message = randomAlphanumeric(10);
- String topic = router.route(message, messageKey, topics, mock(PulsarSinkContext.class));
+ TopicPartition topic =
+ router.route(message, messageKey, partitions, mock(PulsarSinkContext.class));
- assertEquals(topic, desiredTopic);
+ assertEquals(topic, desiredPartitions);
}
private SinkConfiguration sinkConfiguration(MessageKeyHash hash) {
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/router/RoundRobinTopicRouterTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/router/RoundRobinTopicRouterTest.java
index 57e61e5..e64401a 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/router/RoundRobinTopicRouterTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/router/RoundRobinTopicRouterTest.java
@@ -21,6 +21,7 @@ package org.apache.flink.connector.pulsar.sink.writer.router;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.pulsar.sink.config.SinkConfiguration;
import org.apache.flink.connector.pulsar.sink.writer.context.PulsarSinkContext;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList;
@@ -45,7 +46,7 @@ class RoundRobinTopicRouterTest {
RoundRobinTopicRouter<String> router = new RoundRobinTopicRouter<>(configuration);
String message = randomAlphabetic(10);
- List<String> partitions = emptyList();
+ List<TopicPartition> partitions = emptyList();
PulsarSinkContext context = mock(PulsarSinkContext.class);
assertThrows(
@@ -59,24 +60,25 @@ class RoundRobinTopicRouterTest {
SinkConfiguration configuration = sinkConfiguration(batchSize);
RoundRobinTopicRouter<String> router = new RoundRobinTopicRouter<>(configuration);
- List<String> topics = ImmutableList.of("topic1", "topic2");
+ List<TopicPartition> topics =
+ ImmutableList.of(new TopicPartition("topic1"), new TopicPartition("topic2"));
PulsarSinkContext context = mock(PulsarSinkContext.class);
for (int i = 0; i < batchSize; i++) {
String message = randomAlphabetic(10);
- String topic = router.route(message, null, topics, context);
- assertEquals(topic, topics.get(0));
+ TopicPartition partition = router.route(message, null, topics, context);
+ assertEquals(partition, topics.get(0));
}
for (int i = 0; i < batchSize; i++) {
String message = randomAlphabetic(10);
- String topic = router.route(message, null, topics, context);
- assertEquals(topic, topics.get(1));
+ TopicPartition partition = router.route(message, null, topics, context);
+ assertEquals(partition, topics.get(1));
}
String message = randomAlphabetic(10);
- String topic = router.route(message, null, topics, context);
- assertEquals(topic, topics.get(0));
+ TopicPartition partition = router.route(message, null, topics, context);
+ assertEquals(partition, topics.get(0));
}
private SinkConfiguration sinkConfiguration(int switchSize) {
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/TopicMetadataListenerTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/MetadataListenerTest.java
similarity index 64%
rename from flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/TopicMetadataListenerTest.java
rename to flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/MetadataListenerTest.java
index 8af27e8..998a115 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/TopicMetadataListenerTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/MetadataListenerTest.java
@@ -20,6 +20,7 @@ package org.apache.flink.connector.pulsar.sink.writer.topic;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.pulsar.sink.config.SinkConfiguration;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
import org.apache.flink.connector.pulsar.testutils.PulsarTestSuiteBase;
import org.apache.flink.streaming.runtime.tasks.TestProcessingTimeService;
@@ -34,49 +35,48 @@ import static java.util.stream.Collectors.toList;
import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
import static org.apache.flink.connector.pulsar.sink.PulsarSinkOptions.PULSAR_TOPIC_METADATA_REFRESH_INTERVAL;
import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils.topicName;
-import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils.topicNameWithPartition;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
-/** Unit tests for {@link TopicMetadataListener}. */
-class TopicMetadataListenerTest extends PulsarTestSuiteBase {
+/** Unit tests for {@link MetadataListener}. */
+class MetadataListenerTest extends PulsarTestSuiteBase {
@Test
void listenEmptyTopics() {
- TopicMetadataListener listener = new TopicMetadataListener();
+ MetadataListener listener = new MetadataListener();
SinkConfiguration configuration = sinkConfiguration(Duration.ofMinutes(5).toMillis());
TestProcessingTimeService timeService = new TestProcessingTimeService();
- List<String> topics = listener.availableTopics();
- assertThat(topics).isEmpty();
+ List<TopicPartition> partitions = listener.availablePartitions();
+ assertThat(partitions).isEmpty();
listener.open(configuration, timeService);
- topics = listener.availableTopics();
- assertThat(topics).isEmpty();
+ partitions = listener.availablePartitions();
+ assertThat(partitions).isEmpty();
}
@Test
void listenOnPartitions() throws Exception {
String topic = randomAlphabetic(10);
operator().createTopic(topic, 6);
- List<String> partitions = topicPartitions(topic, 6);
+ List<TopicPartition> desiredPartitions = topicPartitions(topic, 6);
+ List<String> partitionNames =
+ desiredPartitions.stream().map(TopicPartition::getFullTopicName).collect(toList());
- TopicMetadataListener listener = new TopicMetadataListener(partitions);
+ MetadataListener listener = new MetadataListener(partitionNames);
long interval = Duration.ofMinutes(15).toMillis();
SinkConfiguration configuration = sinkConfiguration(interval);
TestProcessingTimeService timeService = new TestProcessingTimeService();
- List<String> topics = listener.availableTopics();
- assertEquals(topics, partitions);
-
listener.open(configuration, timeService);
- topics = listener.availableTopics();
- assertEquals(topics, partitions);
+ List<TopicPartition> partitions = listener.availablePartitions();
+ assertEquals(desiredPartitions, partitions);
operator().increaseTopicPartitions(topic, 12);
+ listener.refreshTopicMetadata(topic);
timeService.advance(interval);
- topics = listener.availableTopics();
- assertEquals(topics, partitions);
+ partitions = listener.availablePartitions();
+ assertEquals(desiredPartitions, partitions);
}
@Test
@@ -84,18 +84,19 @@ class TopicMetadataListenerTest extends PulsarTestSuiteBase {
String topic = randomAlphabetic(10);
operator().createTopic(topic, 8);
- TopicMetadataListener listener = new TopicMetadataListener(singletonList(topic));
+ MetadataListener listener = new MetadataListener(singletonList(topic));
SinkConfiguration configuration = sinkConfiguration(Duration.ofMinutes(10).toMillis());
TestProcessingTimeService timeService = new TestProcessingTimeService();
- List<String> topics = listener.availableTopics();
- assertThat(topics).isEmpty();
+ List<TopicPartition> partitions = listener.availablePartitions();
+ assertThat(partitions).isEmpty();
listener.open(configuration, timeService);
- topics = listener.availableTopics();
- List<String> desiredTopics = topicPartitions(topic, 8);
+ partitions = listener.availablePartitions();
+
+ List<TopicPartition> desiredPartitions = topicPartitions(topic, 8);
- assertThat(topics).hasSize(8).isEqualTo(desiredTopics);
+ assertThat(partitions).hasSize(8).isEqualTo(desiredPartitions);
}
@Test
@@ -105,46 +106,47 @@ class TopicMetadataListenerTest extends PulsarTestSuiteBase {
long interval = Duration.ofMinutes(20).toMillis();
- TopicMetadataListener listener = new TopicMetadataListener(singletonList(topic));
+ MetadataListener listener = new MetadataListener(singletonList(topic));
SinkConfiguration configuration = sinkConfiguration(interval);
TestProcessingTimeService timeService = new TestProcessingTimeService();
timeService.setCurrentTime(System.currentTimeMillis());
listener.open(configuration, timeService);
- List<String> topics = listener.availableTopics();
- List<String> desiredTopics = topicPartitions(topic, 8);
+ List<TopicPartition> partitions = listener.availablePartitions();
+ List<TopicPartition> desiredPartitions = topicPartitions(topic, 8);
- assertThat(topics).isEqualTo(desiredTopics);
+ assertThat(partitions).isEqualTo(desiredPartitions);
// Increase topic partitions and trigger the metadata update logic.
operator().increaseTopicPartitions(topic, 16);
+ listener.refreshTopicMetadata(topic);
timeService.advance(interval);
- topics = listener.availableTopics();
- desiredTopics = topicPartitions(topic, 16);
- assertThat(topics).isEqualTo(desiredTopics);
+ partitions = listener.availablePartitions();
+ desiredPartitions = topicPartitions(topic, 16);
+ assertThat(partitions).isEqualTo(desiredPartitions);
}
@Test
void fetchNonPartitionTopic() {
String topic = randomAlphabetic(10);
operator().createTopic(topic, 0);
- List<String> nonPartitionTopic = singletonList(topicName(topic));
+ List<TopicPartition> nonPartitionTopic = singletonList(new TopicPartition(topic));
- TopicMetadataListener listener = new TopicMetadataListener(nonPartitionTopic);
+ MetadataListener listener = new MetadataListener(singletonList(topicName(topic)));
long interval = Duration.ofMinutes(15).toMillis();
SinkConfiguration configuration = sinkConfiguration(interval);
TestProcessingTimeService timeService = new TestProcessingTimeService();
listener.open(configuration, timeService);
- List<String> topics = listener.availableTopics();
- assertEquals(topics, nonPartitionTopic);
+ List<TopicPartition> partitions = listener.availablePartitions();
+ assertEquals(partitions, nonPartitionTopic);
}
- private List<String> topicPartitions(String topic, int partitionSize) {
+ private List<TopicPartition> topicPartitions(String topic, int partitionSize) {
return IntStream.range(0, partitionSize)
.boxed()
- .map(i -> topicNameWithPartition(topic, i))
+ .map(i -> new TopicPartition(topic, i))
.collect(toList());
}
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/TopicProducerRegisterTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/ProducerRegisterTest.java
similarity index 93%
rename from flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/TopicProducerRegisterTest.java
rename to flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/ProducerRegisterTest.java
index 00deda7..75b5703 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/TopicProducerRegisterTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/writer/topic/ProducerRegisterTest.java
@@ -47,8 +47,8 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
-/** Unit tests for {@link TopicProducerRegister}. */
-class TopicProducerRegisterTest extends PulsarTestSuiteBase {
+/** Unit tests for {@link ProducerRegister}. */
+class ProducerRegisterTest extends PulsarTestSuiteBase {
@ParameterizedTest
@EnumSource(DeliveryGuarantee.class)
@@ -59,8 +59,8 @@ class TopicProducerRegisterTest extends PulsarTestSuiteBase {
SinkConfiguration configuration =
new SinkConfiguration(operator().sinkConfig(deliveryGuarantee));
- TopicProducerRegister register =
- new TopicProducerRegister(
+ ProducerRegister register =
+ new ProducerRegister(
configuration, PulsarCrypto.disabled(), createSinkWriterMetricGroup());
String message = randomAlphabetic(10);
@@ -89,8 +89,8 @@ class TopicProducerRegisterTest extends PulsarTestSuiteBase {
SinkConfiguration configuration =
new SinkConfiguration(operator().sinkConfig(deliveryGuarantee));
- TopicProducerRegister register =
- new TopicProducerRegister(
+ ProducerRegister register =
+ new ProducerRegister(
configuration, PulsarCrypto.disabled(), createSinkWriterMetricGroup());
String message = randomAlphabetic(10);
@@ -109,8 +109,8 @@ class TopicProducerRegisterTest extends PulsarTestSuiteBase {
Configuration configuration = operator().sinkConfig(DeliveryGuarantee.AT_LEAST_ONCE);
configuration.set(PULSAR_VALIDATE_SINK_MESSAGE_BYTES, true);
SinkConfiguration sinkConfiguration = new SinkConfiguration(configuration);
- TopicProducerRegister register =
- new TopicProducerRegister(
+ ProducerRegister register =
+ new ProducerRegister(
sinkConfiguration, PulsarCrypto.disabled(), createSinkWriterMetricGroup());
long message = ThreadLocalRandom.current().nextLong();
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarSourceITCase.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarSourceITCase.java
index 0d8041a..268e2cb 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarSourceITCase.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/PulsarSourceITCase.java
@@ -21,8 +21,8 @@ package org.apache.flink.connector.pulsar.source;
import org.apache.flink.connector.pulsar.testutils.PulsarTestContextFactory;
import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntime;
-import org.apache.flink.connector.pulsar.testutils.source.cases.ConsumeEncryptMessagesContext;
-import org.apache.flink.connector.pulsar.testutils.source.cases.MultipleTopicConsumingContext;
+import org.apache.flink.connector.pulsar.testutils.source.cases.EncryptedMessagesConsumingContext;
+import org.apache.flink.connector.pulsar.testutils.source.cases.MultipleTopicsConsumingContext;
import org.apache.flink.connector.pulsar.testutils.source.cases.PartialKeysConsumingContext;
import org.apache.flink.connector.pulsar.testutils.source.cases.SingleTopicConsumingContext;
import org.apache.flink.connector.testframe.environment.MiniClusterTestEnvironment;
@@ -36,6 +36,8 @@ import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.pulsar.client.api.SubscriptionType;
import org.junit.jupiter.api.Tag;
+import static org.apache.flink.streaming.api.CheckpointingMode.EXACTLY_ONCE;
+
/**
* Unit test class for {@link PulsarSource}. Used for {@link SubscriptionType#Exclusive}
* subscription.
@@ -51,8 +53,7 @@ class PulsarSourceITCase extends SourceTestSuiteBase<String> {
PulsarTestEnvironment pulsar = new PulsarTestEnvironment(PulsarRuntime.container());
// This field is preserved, we don't support the semantics in source currently.
- @TestSemantics
- CheckpointingMode[] semantics = new CheckpointingMode[] {CheckpointingMode.EXACTLY_ONCE};
+ @TestSemantics CheckpointingMode[] semantics = new CheckpointingMode[] {EXACTLY_ONCE};
// Defines an external context Factories,
// so test cases will be invoked using these external contexts.
@@ -61,14 +62,14 @@ class PulsarSourceITCase extends SourceTestSuiteBase<String> {
new PulsarTestContextFactory<>(pulsar, SingleTopicConsumingContext::new);
@TestContext
- PulsarTestContextFactory<String, MultipleTopicConsumingContext> multipleTopic =
- new PulsarTestContextFactory<>(pulsar, MultipleTopicConsumingContext::new);
+ PulsarTestContextFactory<String, MultipleTopicsConsumingContext> multipleTopic =
+ new PulsarTestContextFactory<>(pulsar, MultipleTopicsConsumingContext::new);
@TestContext
PulsarTestContextFactory<String, PartialKeysConsumingContext> partialKeys =
new PulsarTestContextFactory<>(pulsar, PartialKeysConsumingContext::new);
@TestContext
- PulsarTestContextFactory<String, ConsumeEncryptMessagesContext> encryptMessages =
- new PulsarTestContextFactory<>(pulsar, ConsumeEncryptMessagesContext::new);
+ PulsarTestContextFactory<String, EncryptedMessagesConsumingContext> encryptMessages =
+ new PulsarTestContextFactory<>(pulsar, EncryptedMessagesConsumingContext::new);
}
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/SplitAssignerImplTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/SplitAssignerImplTest.java
index b081ecb..3f11c41 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/SplitAssignerImplTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/assigner/SplitAssignerImplTest.java
@@ -25,7 +25,6 @@ import org.apache.flink.connector.pulsar.source.enumerator.PulsarSourceEnumState
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.source.split.PulsarPartitionSplit;
-import org.apache.flink.util.TestLogger;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Test;
@@ -49,7 +48,7 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
/** Unit tests for {@link SplitAssignerImpl}. */
-class SplitAssignerImplTest extends TestLogger {
+class SplitAssignerImplTest {
private static final List<MockSplitEnumeratorContext<PulsarPartitionSplit>> enumeratorContexts =
new ArrayList<>();
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/StopCursorTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/StopCursorTest.java
index 9e13b15..e09180f 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/StopCursorTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/StopCursorTest.java
@@ -52,7 +52,7 @@ class StopCursorTest extends PulsarTestSuiteBase {
@Test
void publishTimeStopCursor() throws IOException {
- String topicName = randomAlphanumeric(5);
+ String topicName = "stop-cursor-" + randomAlphanumeric(5);
operator().createTopic(topicName, 2);
PulsarPartitionSplitReader splitReader =
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/PulsarSubscriberTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/PulsarSubscriberTest.java
index cb8819d..bedc3d8 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/PulsarSubscriberTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/PulsarSubscriberTest.java
@@ -38,17 +38,21 @@ import static org.apache.flink.connector.pulsar.source.enumerator.subscriber.Pul
import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils.topicName;
import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils.topicNameWithPartition;
import static org.apache.pulsar.client.api.RegexSubscriptionMode.AllTopics;
+import static org.apache.pulsar.common.partition.PartitionedTopicMetadata.NON_PARTITIONED;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
/** Unit tests for {@link PulsarSubscriber}. */
class PulsarSubscriberTest extends PulsarTestSuiteBase {
- private final String topic1 = topicName("topic-" + randomAlphanumeric(4));
- private final String topic2 = topicName("pattern-topic-" + randomAlphanumeric(4));
- private final String topic3 = topicName("topic2-" + randomAlphanumeric(4));
- private final String topic4 = topicName("non-partitioned-topic-" + randomAlphanumeric(4));
- private final String topic5 = topicName("non-partitioned-topic2-" + randomAlphanumeric(4));
+ private final String topic1 = topicName("pulsar-subscriber-topic-" + randomAlphanumeric(4));
+ private final String topic2 =
+ topicName("pulsar-subscriber-pattern-topic-" + randomAlphanumeric(4));
+ private final String topic3 = topicName("pulsar-subscriber-topic-2-" + randomAlphanumeric(4));
+ private final String topic4 =
+ topicName("pulsar-subscriber-non-partitioned-topic-" + randomAlphanumeric(4));
+ private final String topic5 =
+ topicName("pulsar-subscriber-non-partitioned-topic-2-" + randomAlphanumeric(4));
private static final int NUM_PARTITIONS_PER_TOPIC = 5;
private static final int NUM_PARALLELISM = 10;
@@ -58,8 +62,8 @@ class PulsarSubscriberTest extends PulsarTestSuiteBase {
operator().createTopic(topic1, NUM_PARTITIONS_PER_TOPIC);
operator().createTopic(topic2, NUM_PARTITIONS_PER_TOPIC);
operator().createTopic(topic3, NUM_PARTITIONS_PER_TOPIC);
- operator().createTopic(topic4, 0);
- operator().createTopic(topic5, 0);
+ operator().createTopic(topic4, NON_PARTITIONED);
+ operator().createTopic(topic5, NON_PARTITIONED);
}
@AfterAll
@@ -107,7 +111,7 @@ class PulsarSubscriberTest extends PulsarTestSuiteBase {
subscriber.getSubscribedTopicPartitions(
operator().admin(), new FullRangeGenerator(), NUM_PARALLELISM);
- TopicPartition desiredPartition = new TopicPartition(topic4, -1);
+ TopicPartition desiredPartition = new TopicPartition(topic4);
assertThat(partitions).hasSize(1).containsExactly(desiredPartition);
}
@@ -115,7 +119,8 @@ class PulsarSubscriberTest extends PulsarTestSuiteBase {
void subscribeNonPartitionedTopicPattern() {
PulsarSubscriber subscriber =
getTopicPatternSubscriber(
- Pattern.compile("persistent://public/default/non-partitioned-topic*?"),
+ Pattern.compile(
+ "persistent://public/default/pulsar-subscriber-non-partitioned-topic*?"),
AllTopics);
Set<TopicPartition> topicPartitions =
@@ -134,7 +139,8 @@ class PulsarSubscriberTest extends PulsarTestSuiteBase {
void topicPatternSubscriber() {
PulsarSubscriber subscriber =
getTopicPatternSubscriber(
- Pattern.compile("persistent://public/default/topic*?"), AllTopics);
+ Pattern.compile("persistent://public/default/pulsar-subscriber-topic*?"),
+ AllTopics);
Set<TopicPartition> topicPartitions =
subscriber.getSubscribedTopicPartitions(
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReaderTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReaderTest.java
index 8a69992..7a9e842 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReaderTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReaderTest.java
@@ -284,7 +284,7 @@ class PulsarSourceReaderTest extends PulsarTestSuiteBase {
private void verifyAllMessageAcknowledged(int expectedMessages, String partitionName)
throws PulsarClientException {
- Consumer<byte[]> consumer =
+ try (Consumer<byte[]> consumer =
operator()
.client()
.newConsumer()
@@ -292,10 +292,11 @@ class PulsarSourceReaderTest extends PulsarTestSuiteBase {
.subscriptionInitialPosition(SubscriptionInitialPosition.Latest)
.subscriptionName("verify-message")
.topic(partitionName)
- .subscribe();
-
- assertEquals(
- expectedMessages - 1, ((MessageIdImpl) consumer.getLastMessageId()).getEntryId());
+ .subscribe()) {
+ assertEquals(
+ expectedMessages - 1,
+ ((MessageIdImpl) consumer.getLastMessageId()).getEntryId());
+ }
}
private void verifyNoSubscriptionCreated(String partitionName) throws PulsarAdminException {
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/deserializer/PulsarDeserializationSchemaTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/deserializer/PulsarDeserializationSchemaTest.java
index f833b16..d247427 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/deserializer/PulsarDeserializationSchemaTest.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/deserializer/PulsarDeserializationSchemaTest.java
@@ -291,8 +291,8 @@ class PulsarDeserializationSchemaTest extends PulsarTestSuiteBase {
assertThatCode(() -> runPipeline(source, expectedMessage)).doesNotThrowAnyException();
}
- private PulsarSource createSource(
- String topicName, PulsarDeserializationSchema<?> deserializationSchema) {
+ private <T> PulsarSource<T> createSource(
+ String topicName, PulsarDeserializationSchema<T> deserializationSchema) {
return PulsarSource.builder()
.setDeserializationSchema(deserializationSchema)
.setServiceUrl(operator().serviceUrl())
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContext.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContext.java
index 1408656..d283df8 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContext.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContext.java
@@ -52,10 +52,15 @@ public abstract class PulsarTestContext<T> implements ExternalContext {
@Override
public List<URL> getConnectorJarPaths() {
- // We don't need any test jars definition. They are provided in docker-related environments.
+ // We don't need any test jars' definition.
+ // They are provided in docker-related environments.
return Collections.emptyList();
}
+ /**
+ * This method is called after finishing executing a test case. Check the {@code
+ * TestResourceProvidingInvocationContext.getAdditionalExtensions()} for detailed information.
+ */
@Override
public void close() throws Exception {
// All the topics would be deleted in the PulsarRuntime. No need to manually close them.
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestSuiteBase.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestSuiteBase.java
index 567ccd3..44ca3c7 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestSuiteBase.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestSuiteBase.java
@@ -51,11 +51,11 @@ public abstract class PulsarTestSuiteBase {
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.
+ * Choose the desired pulsar runtime as the test backend. The default test backend is a
+ * singleton pulsar instance. Override this method when needed.
*/
protected PulsarRuntime runtime() {
- return PulsarRuntime.container();
+ return PulsarRuntime.singletonContainer();
}
/** Operate pulsar by acquiring a runtime operator. */
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntime.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntime.java
index 641b924..5f2557e 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntime.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntime.java
@@ -20,6 +20,7 @@ package org.apache.flink.connector.pulsar.testutils.runtime;
import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
import org.apache.flink.connector.pulsar.testutils.runtime.container.PulsarContainerRuntime;
+import org.apache.flink.connector.pulsar.testutils.runtime.singleton.PulsarSingletonRuntime;
import org.testcontainers.containers.GenericContainer;
@@ -43,6 +44,15 @@ public interface PulsarRuntime {
*/
PulsarRuntimeOperator operator();
+ /**
+ * A singleton instance of the Pulsar docker instance in the whole test lifecycle. We will start
+ * the instance only once and didn't stop it after the tests finished. The instance will be
+ * shared among all the tests.
+ */
+ static PulsarRuntime singletonContainer() {
+ return new PulsarSingletonRuntime();
+ }
+
/**
* Create a Pulsar instance in docker. We would start a standalone Pulsar in TestContainers.
* This runtime is often used in end-to-end tests. The stream storage for bookkeeper is
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntimeOperator.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntimeOperator.java
index f975cbe..310462f 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntimeOperator.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntimeOperator.java
@@ -22,7 +22,6 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.pulsar.common.config.PulsarConfiguration;
import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
-import org.apache.flink.connector.testframe.external.ExternalContext;
import org.apache.flink.shaded.guava30.com.google.common.base.Strings;
@@ -80,10 +79,7 @@ import static org.apache.pulsar.client.api.SubscriptionMode.Durable;
import static org.apache.pulsar.client.api.SubscriptionType.Exclusive;
import static org.apache.pulsar.common.partition.PartitionedTopicMetadata.NON_PARTITIONED;
-/**
- * A pulsar cluster operator used for operating pulsar instance. It's serializable for using in
- * {@link ExternalContext}.
- */
+/** A pulsar cluster operator used for operating pulsar instance. */
public class PulsarRuntimeOperator implements Closeable {
public static final int DEFAULT_PARTITIONS = 10;
@@ -447,7 +443,10 @@ public class PulsarRuntimeOperator implements Closeable {
return configuration;
}
- /** This method is used for test framework. You can't close this operator manually. */
+ /**
+ * This method is used for the connector test framework. You don't need to close this operator
+ * manually.
+ */
@Override
public void close() throws PulsarClientException {
if (admin != null) {
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/container/PulsarContainerRuntime.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/container/PulsarContainerRuntime.java
index 4032d0b..59b8759 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/container/PulsarContainerRuntime.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/container/PulsarContainerRuntime.java
@@ -78,8 +78,7 @@ public class PulsarContainerRuntime implements PulsarRuntime {
@Override
public void startUp() {
- boolean havenStartedBefore = started.compareAndSet(false, true);
- if (!havenStartedBefore) {
+ if (!started.compareAndSet(false, true)) {
LOG.warn("You have started the Pulsar Container. We will skip this execution.");
return;
}
@@ -90,6 +89,9 @@ public class PulsarContainerRuntime implements PulsarRuntime {
container.withEnv("PULSAR_PREFIX_systemTopicEnabled", "true");
container.withEnv("PULSAR_PREFIX_brokerDeduplicationEnabled", "true");
container.withEnv("PULSAR_PREFIX_defaultNumberOfNamespaceBundles", "1");
+ container.withEnv("PULSAR_PREFIX_allowAutoTopicCreation", "true");
+ container.withEnv("PULSAR_PREFIX_allowAutoTopicCreationType", "partitioned");
+ container.withEnv("PULSAR_PREFIX_defaultNumPartitions", "4");
// Change the default bootstrap script, it will override the default configuration
// and start a standalone Pulsar without streaming storage and function worker.
container.withCommand(
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/singleton/PulsarSingletonRuntime.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/singleton/PulsarSingletonRuntime.java
new file mode 100644
index 0000000..be6113d
--- /dev/null
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/singleton/PulsarSingletonRuntime.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.singleton;
+
+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.container.PulsarContainerRuntime;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import org.apache.flink.shaded.guava30.com.google.common.util.concurrent.Uninterruptibles;
+
+import java.time.Duration;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * A singleton instance of the Pulsar docker instance in the whole test lifecycle. We will start the
+ * instance only once and didn't stop it after the tests finished. The instance will be shared among
+ * all the tests.
+ *
+ * <p>This runtime can't be used in end-to-end tests.
+ */
+public class PulsarSingletonRuntime implements PulsarRuntime {
+
+ private static final PulsarContainerRuntime container = new PulsarContainerRuntime();
+ private static final AtomicReference<RuntimeStatus> status =
+ new AtomicReference<>(RuntimeStatus.STOPPED);
+
+ @Override
+ public void startUp() {
+ if (status.compareAndSet(RuntimeStatus.STOPPED, RuntimeStatus.STARTING)) {
+ // Entering the start operation.
+ try {
+ container.startUp();
+ status.set(RuntimeStatus.STARTED);
+ } catch (Exception e) {
+ status.set(RuntimeStatus.FATAL);
+ // Crash the test thread.
+ throw new FlinkRuntimeException(e);
+ }
+ } else {
+ // Other thread may have started the runtime. Waiting for the final status.
+ while (status.get() != RuntimeStatus.STARTED && status.get() != RuntimeStatus.FATAL) {
+ Uninterruptibles.sleepUninterruptibly(Duration.ofMillis(500));
+ }
+ if (status.get() == RuntimeStatus.FATAL) {
+ // Crash the test thread.
+ throw new FlinkRuntimeException(
+ "Fall to start this singleton container. Some error occurs.");
+ }
+ }
+ }
+
+ @Override
+ public void tearDown() {
+ // Nothing to do here. TestContainers will take care of stopping the Docker instance.
+ // See https://www.testcontainers.org/test_framework_integration/manual_lifecycle_control/
+ }
+
+ @Override
+ public PulsarRuntimeOperator operator() {
+ return container.operator();
+ }
+
+ /** The runtime signal which is used to share the status among different test threads. */
+ private enum RuntimeStatus {
+ STARTED,
+ STARTING,
+ STOPPED,
+ FATAL,
+ }
+}
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/cases/PulsarSinkTestContext.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/PulsarSinkTestContext.java
similarity index 54%
rename from flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/cases/PulsarSinkTestContext.java
rename to flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/PulsarSinkTestContext.java
index e563b56..8c6877c 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/cases/PulsarSinkTestContext.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/PulsarSinkTestContext.java
@@ -1,22 +1,4 @@
-/*
- * 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.sink.cases;
+package org.apache.flink.connector.pulsar.testutils.sink;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
@@ -24,6 +6,9 @@ import org.apache.flink.api.connector.sink2.Sink;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.pulsar.sink.PulsarSink;
import org.apache.flink.connector.pulsar.sink.PulsarSinkBuilder;
+import org.apache.flink.connector.pulsar.sink.writer.context.PulsarSinkContext;
+import org.apache.flink.connector.pulsar.sink.writer.router.TopicRouter;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
import org.apache.flink.connector.pulsar.testutils.PulsarTestContext;
import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
import org.apache.flink.connector.pulsar.testutils.sink.reader.PulsarPartitionDataReader;
@@ -38,46 +23,59 @@ import org.apache.pulsar.client.api.Schema;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
+import java.util.concurrent.atomic.AtomicInteger;
+import static java.util.Collections.singletonList;
import static org.apache.commons.lang3.RandomStringUtils.randomAlphanumeric;
import static org.apache.flink.connector.pulsar.sink.PulsarSinkOptions.PULSAR_BATCHING_MAX_MESSAGES;
+import static org.apache.flink.connector.pulsar.sink.writer.router.TopicRoutingMode.ROUND_ROBIN;
import static org.apache.flink.connector.pulsar.testutils.PulsarTestCommonUtils.toDeliveryGuarantee;
-/** Common sink test context for pulsar based test. */
-public class PulsarSinkTestContext extends PulsarTestContext<String>
+/**
+ * Common sink test context for the pulsar based tests. We use the string text as the basic send
+ * content.
+ */
+public abstract class PulsarSinkTestContext extends PulsarTestContext<String>
implements DataStreamSinkV2ExternalContext<String> {
- private static final String TOPIC_NAME_PREFIX = "flink-sink-topic-";
+ private static final String TOPIC_NAME_PREFIX = "persistent://public/default/flink-sink-topic-";
private static final int RECORD_SIZE_UPPER_BOUND = 300;
private static final int RECORD_SIZE_LOWER_BOUND = 100;
private static final int RECORD_STRING_SIZE = 20;
- protected String topicName = topicName();
- protected final Closer closer = Closer.create();
+ private List<String> topics = generateTopics();
+ private final Closer closer = Closer.create();
public PulsarSinkTestContext(PulsarTestEnvironment environment) {
super(environment, Schema.STRING);
}
- @Override
- protected String displayName() {
- return "write messages into one topic in Pulsar";
- }
-
@Override
public Sink<String> createSink(TestingSinkSettings sinkSettings) {
- operator.createTopic(topicName, 4);
+ // Create the topic if it needs.
+ if (creatTopic()) {
+ for (String topic : topics) {
+ operator.createTopic(topic, 4);
+ }
+ }
+
+ // We don't have NONE delivery guarantee and no need to test it by default.
DeliveryGuarantee guarantee = toDeliveryGuarantee(sinkSettings.getCheckpointingMode());
PulsarSinkBuilder<String> builder =
PulsarSink.builder()
.setServiceUrl(operator.serviceUrl())
.setAdminUrl(operator.adminUrl())
- .setTopics(topicName)
.setDeliveryGuarantee(guarantee)
.setSerializationSchema(schema)
.enableSchemaEvolution()
.setConfig(PULSAR_BATCHING_MAX_MESSAGES, 4);
+ if (creatTopic()) {
+ builder.setTopics(topics).setTopicRoutingMode(ROUND_ROBIN);
+ } else {
+ builder.setTopicRouter(new CustomTopicRouter(topics));
+ }
+
setSinkBuilder(builder);
return builder.build();
@@ -85,8 +83,7 @@ public class PulsarSinkTestContext extends PulsarTestContext<String>
@Override
public ExternalSystemDataReader<String> createSinkDataReader(TestingSinkSettings sinkSettings) {
- PulsarPartitionDataReader<String> reader =
- new PulsarPartitionDataReader<>(operator, topicName, Schema.STRING);
+ PulsarPartitionDataReader<String> reader = createSinkDataReader(topics);
closer.register(reader);
return reader;
@@ -115,16 +112,53 @@ public class PulsarSinkTestContext extends PulsarTestContext<String>
@Override
public void close() throws Exception {
- // Change the topic name after finishing a test case.
- closer.register(() -> topicName = topicName());
+ // Switch to another topic info after finishing a test case.
+ this.topics = generateTopics();
+
closer.close();
}
+ /** Return topic info. We would use only one topic by default. */
+ protected List<String> generateTopics() {
+ String topicName = TOPIC_NAME_PREFIX + randomAlphanumeric(8);
+ return singletonList(topicName);
+ }
+
+ /** Override the default sink behavior. */
protected void setSinkBuilder(PulsarSinkBuilder<String> builder) {
// Nothing to do by default.
}
- private String topicName() {
- return TOPIC_NAME_PREFIX + randomAlphanumeric(8);
+ /** Set it to false for testing topic auto creation feature. */
+ protected boolean creatTopic() {
+ return true;
+ }
+
+ protected PulsarPartitionDataReader<String> createSinkDataReader(List<String> topics) {
+ return new PulsarPartitionDataReader<>(operator, topics, Schema.STRING);
+ }
+
+ /**
+ * Custom topic router for auto-generated topics. We will switch to another topic every 10
+ * messages.
+ */
+ public static class CustomTopicRouter implements TopicRouter<String> {
+ private static final long serialVersionUID = 1698701183626468094L;
+
+ private final List<String> topics;
+ private final AtomicInteger counter;
+
+ public CustomTopicRouter(List<String> topics) {
+ this.topics = topics;
+ this.counter = new AtomicInteger(0);
+ }
+
+ @Override
+ public TopicPartition route(
+ String s, String key, List<TopicPartition> partitions, PulsarSinkContext context) {
+ int index = counter.incrementAndGet() / 10 % topics.size();
+ String topic = topics.get(index);
+ return new TopicPartition(topic);
+ }
}
}
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/reader/PulsarEncryptDataReader.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/cases/AutoCreateTopicProducingContext.java
similarity index 50%
copy from flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/reader/PulsarEncryptDataReader.java
copy to flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/cases/AutoCreateTopicProducingContext.java
index 14ece03..37d1ffb 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/reader/PulsarEncryptDataReader.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/cases/AutoCreateTopicProducingContext.java
@@ -16,21 +16,28 @@
* limitations under the License.
*/
-package org.apache.flink.connector.pulsar.testutils.sink.reader;
+package org.apache.flink.connector.pulsar.testutils.sink.cases;
-import org.apache.flink.connector.pulsar.common.crypto.PulsarCrypto;
-import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntimeOperator;
+import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
+import org.apache.flink.connector.pulsar.testutils.sink.PulsarSinkTestContext;
-import org.apache.pulsar.client.api.Schema;
+/**
+ * Sink the messages into a non-existed topic and test the connector could auto create it. This test
+ * case only gets passed when the {@code allowAutoTopicCreation} is enabled on the Pulsar runtime.
+ */
+public class AutoCreateTopicProducingContext extends PulsarSinkTestContext {
+
+ public AutoCreateTopicProducingContext(PulsarTestEnvironment environment) {
+ super(environment);
+ }
-/** The data reader for reading encrypted messages from Pulsar. */
-public class PulsarEncryptDataReader<T> extends PulsarPartitionDataReader<T> {
+ @Override
+ protected boolean creatTopic() {
+ return false;
+ }
- public PulsarEncryptDataReader(
- PulsarRuntimeOperator operator,
- String fullTopicName,
- Schema<T> schema,
- PulsarCrypto pulsarCrypto) {
- super(operator, fullTopicName, schema, pulsarCrypto);
+ @Override
+ protected String displayName() {
+ return "write messages into a non-existed topic in Pulsar";
}
}
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/cases/PulsarEncryptSinkContext.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/cases/EncryptedMessageProducingContext.java
similarity index 75%
rename from flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/cases/PulsarEncryptSinkContext.java
rename to flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/cases/EncryptedMessageProducingContext.java
index 42857a1..0710dd1 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/cases/PulsarEncryptSinkContext.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/cases/EncryptedMessageProducingContext.java
@@ -23,25 +23,25 @@ import org.apache.flink.connector.pulsar.sink.PulsarSinkBuilder;
import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
import org.apache.flink.connector.pulsar.testutils.PulsarTestKeyReader;
import org.apache.flink.connector.pulsar.testutils.PulsarTestKeyReader.MessageCryptoBcSupplier;
+import org.apache.flink.connector.pulsar.testutils.sink.PulsarSinkTestContext;
import org.apache.flink.connector.pulsar.testutils.sink.reader.PulsarEncryptDataReader;
-import org.apache.flink.connector.testframe.external.ExternalSystemDataReader;
-import org.apache.flink.connector.testframe.external.sink.TestingSinkSettings;
+import org.apache.flink.connector.pulsar.testutils.sink.reader.PulsarPartitionDataReader;
+
+import java.util.List;
import static org.apache.flink.connector.pulsar.testutils.PulsarTestKeyReader.ENCRYPT_KEY;
import static org.apache.pulsar.client.api.ProducerCryptoFailureAction.FAIL;
import static org.apache.pulsar.client.api.Schema.STRING;
-/** The sink context for supporting producing messages which are encrypted. */
-public class PulsarEncryptSinkContext extends PulsarSinkTestContext {
+/** Sink the encrypted messages into a topic and try to decrypt them. */
+public class EncryptedMessageProducingContext extends PulsarSinkTestContext {
- public PulsarEncryptSinkContext(PulsarTestEnvironment environment) {
+ public EncryptedMessageProducingContext(PulsarTestEnvironment environment) {
super(environment);
}
@Override
protected void setSinkBuilder(PulsarSinkBuilder<String> builder) {
- super.setSinkBuilder(builder);
-
PulsarCrypto pulsarCrypto =
PulsarCrypto.builder()
.cryptoKeyReader(new PulsarTestKeyReader())
@@ -52,22 +52,18 @@ public class PulsarEncryptSinkContext extends PulsarSinkTestContext {
}
@Override
- public ExternalSystemDataReader<String> createSinkDataReader(TestingSinkSettings sinkSettings) {
+ protected PulsarPartitionDataReader<String> createSinkDataReader(List<String> topics) {
PulsarCrypto pulsarCrypto =
PulsarCrypto.builder()
.cryptoKeyReader(new PulsarTestKeyReader())
.addEncryptKeys(ENCRYPT_KEY)
.messageCrypto(new MessageCryptoBcSupplier(false))
.build();
- PulsarEncryptDataReader<String> reader =
- new PulsarEncryptDataReader<>(operator, topicName, STRING, pulsarCrypto);
- closer.register(reader);
-
- return reader;
+ return new PulsarEncryptDataReader<>(operator, topics, STRING, pulsarCrypto);
}
@Override
protected String displayName() {
- return "write messages into one topic by End-to-end encryption";
+ return "write messages into one topic by end-to-end encryption";
}
}
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/cases/MultipleTopicsProducingContext.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/cases/MultipleTopicsProducingContext.java
new file mode 100644
index 0000000..ddd1e2a
--- /dev/null
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/cases/MultipleTopicsProducingContext.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.sink.cases;
+
+import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
+import org.apache.flink.connector.pulsar.testutils.sink.PulsarSinkTestContext;
+
+import java.util.List;
+import java.util.stream.Stream;
+
+import static java.util.stream.Collectors.toList;
+import static org.apache.commons.lang3.RandomStringUtils.randomAlphanumeric;
+
+/** Sink the message into multiple topics and test all the messages have been written. */
+public class MultipleTopicsProducingContext extends PulsarSinkTestContext {
+
+ private static final String TOPIC_NAME_PREFIX = "sink-multiple-topic-";
+
+ public MultipleTopicsProducingContext(PulsarTestEnvironment environment) {
+ super(environment);
+ }
+
+ @Override
+ protected String displayName() {
+ return "write messages into multiple topics in Pulsar";
+ }
+
+ @Override
+ protected List<String> generateTopics() {
+ return Stream.generate(() -> TOPIC_NAME_PREFIX + randomAlphanumeric(10))
+ .limit(4)
+ .collect(toList());
+ }
+}
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/reader/PulsarEncryptDataReader.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/cases/SingleTopicProducingContext.java
similarity index 55%
copy from flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/reader/PulsarEncryptDataReader.java
copy to flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/cases/SingleTopicProducingContext.java
index 14ece03..6417856 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/reader/PulsarEncryptDataReader.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/cases/SingleTopicProducingContext.java
@@ -16,21 +16,20 @@
* limitations under the License.
*/
-package org.apache.flink.connector.pulsar.testutils.sink.reader;
+package org.apache.flink.connector.pulsar.testutils.sink.cases;
-import org.apache.flink.connector.pulsar.common.crypto.PulsarCrypto;
-import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntimeOperator;
+import org.apache.flink.connector.pulsar.testutils.PulsarTestEnvironment;
+import org.apache.flink.connector.pulsar.testutils.sink.PulsarSinkTestContext;
-import org.apache.pulsar.client.api.Schema;
+/** Sink the message into a topic and test all the messages have been written. */
+public class SingleTopicProducingContext extends PulsarSinkTestContext {
-/** The data reader for reading encrypted messages from Pulsar. */
-public class PulsarEncryptDataReader<T> extends PulsarPartitionDataReader<T> {
+ public SingleTopicProducingContext(PulsarTestEnvironment environment) {
+ super(environment);
+ }
- public PulsarEncryptDataReader(
- PulsarRuntimeOperator operator,
- String fullTopicName,
- Schema<T> schema,
- PulsarCrypto pulsarCrypto) {
- super(operator, fullTopicName, schema, pulsarCrypto);
+ @Override
+ protected String displayName() {
+ return "write messages into one topic in Pulsar";
}
}
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/reader/PulsarEncryptDataReader.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/reader/PulsarEncryptDataReader.java
index 14ece03..ac6be2e 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/reader/PulsarEncryptDataReader.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/reader/PulsarEncryptDataReader.java
@@ -23,14 +23,16 @@ import org.apache.flink.connector.pulsar.testutils.runtime.PulsarRuntimeOperator
import org.apache.pulsar.client.api.Schema;
+import java.util.List;
+
/** The data reader for reading encrypted messages from Pulsar. */
public class PulsarEncryptDataReader<T> extends PulsarPartitionDataReader<T> {
public PulsarEncryptDataReader(
PulsarRuntimeOperator operator,
- String fullTopicName,
+ List<String> topics,
Schema<T> schema,
PulsarCrypto pulsarCrypto) {
- super(operator, fullTopicName, schema, pulsarCrypto);
+ super(operator, topics, schema, pulsarCrypto);
}
}
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/reader/PulsarPartitionDataReader.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/reader/PulsarPartitionDataReader.java
index 17a6523..55c37b0 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/reader/PulsarPartitionDataReader.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/sink/reader/PulsarPartitionDataReader.java
@@ -52,24 +52,24 @@ public class PulsarPartitionDataReader<T> implements ExternalSystemDataReader<T>
private static final Logger LOG = LoggerFactory.getLogger(PulsarPartitionDataReader.class);
private final Consumer<T> consumer;
+ private final PulsarRuntimeOperator operator;
public PulsarPartitionDataReader(
- PulsarRuntimeOperator operator, String fullTopicName, Schema<T> schema) {
- this(operator, fullTopicName, schema, PulsarCrypto.disabled());
+ PulsarRuntimeOperator operator, List<String> topics, Schema<T> schema) {
+ this(operator, topics, schema, PulsarCrypto.disabled());
}
protected PulsarPartitionDataReader(
PulsarRuntimeOperator operator,
- String fullTopicName,
+ List<String> topics,
Schema<T> schema,
PulsarCrypto pulsarCrypto) {
// Create the consumer for supporting the E2E tests in the meantime.
- String subscriptionName = randomAlphanumeric(12);
ConsumerBuilder<T> builder =
operator.client()
.newConsumer(schema)
- .topic(fullTopicName)
- .subscriptionName(subscriptionName)
+ .topics(topics)
+ .subscriptionName(randomAlphanumeric(12))
.subscriptionType(SubscriptionType.Exclusive)
.subscriptionMode(SubscriptionMode.Durable)
.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
@@ -88,6 +88,7 @@ public class PulsarPartitionDataReader<T> implements ExternalSystemDataReader<T>
}
this.consumer = sneakyClient(builder::subscribe);
+ this.operator = operator;
}
@Override
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/PulsarSourceTestContext.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/PulsarSourceTestContext.java
index a3aa9e9..5a65a33 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/PulsarSourceTestContext.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/PulsarSourceTestContext.java
@@ -44,7 +44,7 @@ import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSA
import static org.apache.pulsar.client.api.RegexSubscriptionMode.AllTopics;
/**
- * Common source test context for pulsar based test. We use the string text as the basic send
+ * Common source test context for the pulsar based tests. We use the string text as the basic send
* content.
*/
public abstract class PulsarSourceTestContext extends PulsarTestContext<String>
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/ConsumeEncryptMessagesContext.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/EncryptedMessagesConsumingContext.java
similarity index 94%
rename from flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/ConsumeEncryptMessagesContext.java
rename to flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/EncryptedMessagesConsumingContext.java
index 367102a..fea857a 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/ConsumeEncryptMessagesContext.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/EncryptedMessagesConsumingContext.java
@@ -31,9 +31,9 @@ import static org.apache.flink.connector.pulsar.testutils.PulsarTestKeyReader.EN
import static org.apache.pulsar.client.api.ConsumerCryptoFailureAction.FAIL;
/** We will use this context for producing messages with encryption support. */
-public class ConsumeEncryptMessagesContext extends MultipleTopicConsumingContext {
+public class EncryptedMessagesConsumingContext extends MultipleTopicsConsumingContext {
- public ConsumeEncryptMessagesContext(PulsarTestEnvironment environment) {
+ public EncryptedMessagesConsumingContext(PulsarTestEnvironment environment) {
super(environment);
}
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/MultipleTopicConsumingContext.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/MultipleTopicsConsumingContext.java
similarity index 92%
rename from flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/MultipleTopicConsumingContext.java
rename to flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/MultipleTopicsConsumingContext.java
index d5dc3fb..009a98c 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/MultipleTopicConsumingContext.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/MultipleTopicsConsumingContext.java
@@ -28,13 +28,13 @@ import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNam
* Pulsar external context that will create multiple topics with only one partitions as source
* splits.
*/
-public class MultipleTopicConsumingContext extends PulsarSourceTestContext {
+public class MultipleTopicsConsumingContext extends PulsarSourceTestContext {
private final String topicPrefix = "flink-multiple-topic-" + randomAlphabetic(8) + "-";
private int index = 0;
- public MultipleTopicConsumingContext(PulsarTestEnvironment environment) {
+ public MultipleTopicsConsumingContext(PulsarTestEnvironment environment) {
super(environment);
}
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/PartialKeysConsumingContext.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/PartialKeysConsumingContext.java
index 7ebb499..6dd7735 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/PartialKeysConsumingContext.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/PartialKeysConsumingContext.java
@@ -30,7 +30,7 @@ import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSA
import static org.apache.flink.connector.pulsar.source.enumerator.topic.range.TopicRangeUtils.keyHash;
/** We would consume from test splits by using partial keys consuming. */
-public class PartialKeysConsumingContext extends MultipleTopicConsumingContext {
+public class PartialKeysConsumingContext extends MultipleTopicsConsumingContext {
private final String keyToRead;
private final String keyToExclude;
diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/SingleTopicConsumingContext.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/SingleTopicConsumingContext.java
index 469eeeb..9755e58 100644
--- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/SingleTopicConsumingContext.java
+++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/source/cases/SingleTopicConsumingContext.java
@@ -25,7 +25,7 @@ import static org.apache.commons.lang3.RandomStringUtils.randomAlphanumeric;
import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils.topicNameWithPartition;
/**
- * A Pulsar external context that will create only one topic and use partitions in that topic as
+ * A Pulsar external context that will create only one topic and use partitions in that topic as the
* source splits.
*/
public class SingleTopicConsumingContext extends PulsarSourceTestContext {