You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ar...@apache.org on 2021/09/10 12:53:31 UTC
[flink] branch release-1.14 updated (df3b723 -> b32e6a6)
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a change to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git.
from df3b723 [FLINK-24161] Fix interplay of stop-with-savepoint w/o drain with final checkpoints
new 77012db [FLINK-23864][docs] Add flink-connector-pulsar module to flink-docs, auto generate the config document.
new d40cfc6 [FLINK-23864][connector/pulsar] Release Pulsar Message if user enable poolMessage option.
new 982db57 [FLINK-23864][connector/pulsar] Remove PULSAR_AUTO_UPDATE_PARTITIONS option.
new b32e6a6 [FLINK-23864][docs] Add pulsar connector document (Chinese & English).
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../docs/connectors/datastream/pulsar.md | 356 +++++++++++++++++
docs/content/docs/connectors/datastream/pulsar.md | 420 +++++++++++++++++++++
.../generated/pulsar_admin_configuration.html | 42 +++
.../generated/pulsar_client_configuration.html | 222 +++++++++++
.../generated/pulsar_consumer_configuration.html | 174 +++++++++
.../generated/pulsar_source_configuration.html | 54 +++
.../pulsar/common/config/PulsarOptions.java | 153 ++++----
.../pulsar/source/PulsarSourceOptions.java | 233 ++++++------
.../source/config/PulsarSourceConfigUtils.java | 2 -
.../split/PulsarOrderedPartitionSplitReader.java | 3 +
.../split/PulsarUnorderedPartitionSplitReader.java | 3 +
flink-docs/pom.xml | 6 +
.../configuration/ConfigOptionsDocGenerator.java | 8 +-
13 files changed, 1480 insertions(+), 196 deletions(-)
create mode 100644 docs/content.zh/docs/connectors/datastream/pulsar.md
create mode 100644 docs/content/docs/connectors/datastream/pulsar.md
create mode 100644 docs/layouts/shortcodes/generated/pulsar_admin_configuration.html
create mode 100644 docs/layouts/shortcodes/generated/pulsar_client_configuration.html
create mode 100644 docs/layouts/shortcodes/generated/pulsar_consumer_configuration.html
create mode 100644 docs/layouts/shortcodes/generated/pulsar_source_configuration.html
[flink] 04/04: [FLINK-23864][docs] Add pulsar connector document
(Chinese & English).
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
commit b32e6a68359b3e9b705044aaf1ccc6dad125cf21
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Fri Sep 10 14:01:34 2021 +0800
[FLINK-23864][docs] Add pulsar connector document (Chinese & English).
---
.../docs/connectors/datastream/pulsar.md | 356 +++++++++++++++++
docs/content/docs/connectors/datastream/pulsar.md | 420 +++++++++++++++++++++
2 files changed, 776 insertions(+)
diff --git a/docs/content.zh/docs/connectors/datastream/pulsar.md b/docs/content.zh/docs/connectors/datastream/pulsar.md
new file mode 100644
index 0000000..da6380a
--- /dev/null
+++ b/docs/content.zh/docs/connectors/datastream/pulsar.md
@@ -0,0 +1,356 @@
+---
+title: Pulsar
+weight: 9
+type: docs
+---
+<!--
+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.
+-->
+
+# Apache Pulsar 连接器
+
+Flink 当前只提供 [Apache Pulsar](https://pulsar.apache.org) 数据源，用户可以使用它从 Pulsar 读取数据，并保证每条数据只被处理一次。
+
+## 添加依赖
+
+连接器当前支持 Pulsar 2.7.0 之后的版本，但是连接器使用到了 Pulsar 的[事务机制](https://pulsar.apache.org/docs/en/txn-what/),建议在 Pulsar 2.8.0
+及其之后的版本上使用连接器进行数据读取。
+
+如果想要了解更多关于 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 withScalaVersion >}}
+
+使用本连接器的同时，记得把 `flink-connector-base` 也加到项目的依赖里面：
+
+{{< artifact flink-connector-base >}}
+
+Flink 的流连接器并不会放到发行文件里面一同发布，阅读[此文档]({{< ref "docs/dev/datastream/project-configuration" >}})，了解如何将连接器添加到集群实例内。
+
+## Pulsar 数据源
+
+{{< hint info >}}
+Pulsar 数据源基于 Flink 最新的[批流一体 API]({{< ref "docs/dev/datastream/sources.md" >}}) 进行开发。
+
+如果要想使用基于旧版的 `SourceFunction` 实现的 Pulsar 数据源，或者是项目的 Flink 版本低于 1.14，可以使用 StreamNative 单独维护的 [pulsar-flink](https://github.com/streamnative/pulsar-flink)。
+{{< /hint >}}
+
+### 使用示例
+
+Pulsar 数据源提供了 builder 类来构造数据源实例。下面的代码实例使用 builder 类创建的数据源会从 topic "persistent://public/default/my-topic" 的数据开始端进行消费。
+连接器使用了 **Exclusive**（独占）的订阅方式消费消息，订阅名称为 `my-subscription`，并把消息体的二进制字节流以 UTF-8 的方式编码为字符串。
+
+```java
+PulsarSource<String> pulsarSource = PulsarSource.builder()
+ .setServiceUrl(serviceUrl)
+ .setAdminUrl(adminUrl)
+ .setStartCursor(StartCursor.earliest())
+ .setTopics("my-topic")
+ .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(new SimpleStringSchema()))
+ .setSubscriptionName("my-subscription")
+ .setSubscriptionType(SubscriptionType.Exclusive)
+ .build();
+
+env.fromSource(source, WatermarkStrategy.noWatermarks(), "Pulsar Source");
+```
+
+如果使用构造类构造 Pulsar 数据源，一定要提供下面几个属性：
+
+- Pulsar 数据消费的地址，使用 `setServiceUrl(String)` 方法提供
+- Pulsar HTTP 管理地址，使用 `setAdminUrl(String)` 方法提供
+- Pulsar 订阅名称，使用 `setSubscriptionName(String)` 方法提供
+- 需要消费的 topic 或者是 topic 下面的分区，详见[指定消费的 Topic 或者 Topic 分区](#指定消费的-topic-或者-topic-分区)
+- 解码 Pulsar 消息的反序列化器，详见[反序列化器](#反序列化器)
+
+### 指定消费的 Topic 或者 Topic 分区
+
+Pulsar 数据源提供了两种订阅 topic 或 topic 分区的方式。
+
+- Topic 列表，从这个 Topic 的所有分区上消费消息，例如：
+ ```java
+ PulsarSource.builder().setTopics("some-topic1", "some-topic2")
+
+ // 从 topic "topic-a" 的 0 和 1 分区上消费
+ PulsarSource.builder().setTopics("topic-a-partition-0", "topic-a-partition-2")
+ ```
+
+- Topic 正则，连接器使用给定的正则表达式匹配出所有合规的 topic，例如：
+ ```java
+ PulsarSource.builder().setTopicPattern("topic-*")
+ ```
+
+#### Topic 名称简写
+
+从 Pulsar 2.0 之后，完整的 topic 名称格式为 `{persistent|non-persistent}://租户/命名空间/topic`。
+但是连接器不需要提供 topic 名称的完整定义，因为 topic 类型、租户、命名空间都设置了默认值。
+
+Topic 属性 | 默认值
+:------------|:-------
+topic 类型 | `persistent`
+租户 | `public`
+命名空间 | `default`
+
+下面的表格提供了当前 Pulsar 支持的简写方式：
+
+topic 名称简写 | 翻译后的 topic 名称
+:----------------|:---------------------
+`my-topic` | `persistent://public/default/my-topic`
+`my-tenant/my-namespace/my-topic` | `persistent://my-tenant/my-namespace/my-topic`
+
+{{< hint warning >}}
+对于 non-persistent（非持久化） topic，连接器不支持简写名称。所以无法将 `non-persistent://public/default/my-topic` 简写成 `non-persistent://my-topic`。
+{{< /hint >}}
+
+#### Pulsar Topic 层次结构
+
+对于 Pulsar 而言，Topic 分区也是一种 Topic。Pulsar 会将一个有分区的 Topic 在内部按照分区的大小拆分成等量的无分区 Topic。
+
+例如，在 Pulsar 的 `sample` 租户下面的 `flink` 命名空间里面创建了一个有 3 个分区的 topic，给它起名为 `simple-string`。
+可以在 Pulsar 上看到如下的 topic 列表：
+
+Topic 名称 | 是否分区
+:--------- | :----------
+`persistent://sample/flink/simple-string` | 是
+`persistent://sample/flink/simple-string-partition-0` | 否
+`persistent://sample/flink/simple-string-partition-1` | 否
+`persistent://sample/flink/simple-string-partition-2` | 否
+
+这意味着，用户可以用上面的子 topic 去直接消费分区里面的数据，不需要再去基于上层的父 topic 去消费全部分区的数据。
+例如：使用 `PulsarSource.builder().setTopics("sample/flink/simple-string-partition-1", "sample/flink/simple-string-partition-2")` 将会只消费 topic `sample/flink/simple-string` 上面的分区 1 和 2 里面的消息。
+
+#### 配置 Topic 正则表达式
+
+前面提到了 Pulsar topic 有 `persistent`、`non-persistent` 两种类型，使用正则表达式消费数据的时候，连接器会尝试从正则表达式里面解析出消息的类型。
+例如：`PulsarSource.builder().setTopicPattern("non-persistent://my-topic*")` 会解析出 `non-persistent` 这个 topic 类型。
+如果用户使用 topic 名称简写的方式，连接器会使用默认的消息类型 `persistent`。
+
+如果想用正则去消费 `persistent` 和 `non-persistent` 类型的 topic，需要使用 `RegexSubscriptionMode` 定义 topic 类型，例如：`setTopicPattern("topic-*", RegexSubscriptionMode.AllTopics)`。
+
+### 反序列化器
+
+反序列化器用于解析 Pulsar 消息，连接器使用 `PulsarDeserializationSchema` 来定义反序列化器。
+用户可以在 builder 类中使用 `setDeserializationSchema(PulsarDeserializationSchema)` 方法配置反序列化器，它会解析 Pulsar 的 `Message<byte[]>` 实例。
+
+如果用户只关心消息体的二进制字节流，并不需要其他属性来解析数据。可以直接使用预定义的 `PulsarDeserializationSchema`。Pulsar 连接器里面提供了 3 种预定义好的反序列化器。
+
+- 使用 Pulsar 的 [Schema](https://pulsar.apache.org/docs/en/schema-understand/) 解析消息。
+ ```java
+ // 基础数据类型
+ PulsarDeserializationSchema.pulsarSchema(Schema)
+
+ // 结构类型 (JSON, Protobuf, Avro, etc.)
+ PulsarDeserializationSchema.pulsarSchema(Schema, Class)
+
+ // 键值对类型
+ PulsarDeserializationSchema.pulsarSchema(Schema, Class, Class)
+ ```
+- 使用 Flink 的 `DeserializationSchema` 解析消息。
+ ```java
+ PulsarDeserializationSchema.flinkSchema(DeserializationSchema)
+ ```
+- 使用 Flink 的 `TypeInformation` 解析消息。
+ ```java
+ PulsarDeserializationSchema.flinkTypeInfo(TypeInformation, ExecutionConfig)
+ ```
+
+Pulsar 的 `Message<byte[]>` 包含了很多 [额外的属性](https://pulsar.apache.org/docs/zh-CN/concepts-messaging/#%E6%B6%88%E6%81%AF)。
+例如,消息的 key，消息发送时间，消息生产时间，用户在消息上自定义的键值对属性等。可以使用 `Message<byte[]>` 接口来获取这些属性。
+
+如果用户需要基于这些额外的属性来解析一条消息，可以实现 `PulsarDeserializationSchema` 接口。
+并一定要确保 `PulsarDeserializationSchema.getProducedType()` 方法返回的 `TypeInformation` 是正确的结果。
+Flink 使用 `TypeInformation` 将解析出来的结果序列化传递到下游算子。
+
+### Pulsar 订阅
+
+订阅是命名好的配置规则，指导消息如何投递给消费者。连接器需要提供一个独立的订阅名称,支持 Pulsar 的四种订阅模式：
+
+- [exclusive（独占）](https://pulsar.apache.org/docs/zh-CN/concepts-messaging/#exclusive)
+- [shared（共享）](https://pulsar.apache.org/docs/zh-CN/concepts-messaging/#shared)
+- [failover（灾备）](https://pulsar.apache.org/docs/zh-CN/concepts-messaging/#failover)
+- [key_shared（key 共享）](https://pulsar.apache.org/docs/zh-CN/concepts-messaging/#key_shared)
+
+当前 Pulsar 连接器里面，`独占` 和 `灾备` 的实现没有区别，如果 Flink 的一个 reader 挂了，连接器会把所有未消费的数据交给其他的 reader 来消费数据。
+
+默认情况下，如果没有指定订阅类型，连接器使用共享订阅类型（`SubscriptionType.Shared`）。
+
+```java
+// 名为 "my-shared" 的共享订阅
+PulsarSource.builder().setSubscriptionName("my-shared")
+
+// 名为 "my-exclusive" 的独占订阅
+PulsarSource.builder().setSubscriptionName("my-exclusive").setSubscriptionType(SubscriptionType.Exclusive)
+```
+
+如果想在 Pulsar 连接器里面使用 `key 共享` 订阅，需要提供 `RangeGenerator` 实例。`RangeGenerator` 会生成一组消息 key 的 hash 范围，连接器会基于给定的范围来消费数据。
+
+Pulsar 连接器也提供了一个名为 `UniformRangeGenerator` 的默认实现，它会基于 flink 数据源的并行度将 hash 范围均分。
+
+### 起始消费位置
+
+连接器使用 `setStartCursor(StartCursor)` 方法给定开始消费的位置。内置的消费位置有：
+
+- 从 topic 里面最早的一条消息开始消费。
+ ```java
+ StartCursor.earliest()
+ ```
+- 从 topic 里面最新的一条消息开始消费。
+ ```java
+ StartCursor.latest()
+ ```
+- 从给定的消息开始消费。
+ ```java
+ StartCursor.fromMessageId(MessageId)
+ ```
+- 与前者不同的是，给定的消息可以跳过，再进行消费。
+ ```java
+ StartCursor.fromMessageId(MessageId, boolean)
+ ```
+- 从给定的消息时间开始消费。
+ ```java
+ StartCursor.fromMessageTime(long)
+ ```
+
+{{< hint info >}}
+每条消息都有一个固定的序列号，这个序列号在 Pulsar 上有序排列，其包含了 ledger、entry、partition 等原始信息，用于在 Pulsar 底层存储上查找到具体的消息。
+Pulsar 称这个序列号为 `MessageId`，用户可以使用 `DefaultImplementation.newMessageId(long ledgerId, long entryId, int partitionIndex)` 创建它。
+{{< /hint >}}
+
+### 边界
+
+Pulsar 连接器同时支持流式和批的消费方式，默认情况下，连接器使用流的方式消费数据。除非任务失败或者被取消，否则连接器将持续消费数据。
+用户可以使用 `setBoundedStopCursor(StopCursor)` 给定停止消费的位置，这种情况下连接器会使用批的方式进行消费。当所有 topic 分区都消费到了停止位置，Flink 任务就会结束。
+
+使用流的方式一样可以给定停止位置，使用 `setUnboundedStopCursor(StopCursor)` 方法即可。
+
+内置的停止位置如下:
+
+- 永不停止。
+ ```java
+ StopCursor.never()
+ ```
+- 停止于 Pulsar 启动时 topic 里面最新的那条数据。
+ ```java
+ StopCursor.latest()
+ ```
+- 停止于某条消息，结果里不包含此消息。
+ ```java
+ StopCursor.atMessageId(MessageId)
+ ```
+- 停止于某条消息之后，结果里包含此消息。
+ ```java
+ StopCursor.afterMessageId(MessageId)
+ ```
+- 停止于某个给定的消息时间戳。
+ ```java
+ StopCursor.atEventTime(long)
+ ```
+
+### 其他配置项
+
+除了前面提到的配置选项，连接器还提供了丰富的选项供 Pulsar 专家使用，在 builder 类里通过 `setConfig(ConfigOption<T>, T)` 和 `setConfig(Configuration)` 方法给定下述的全部配置。
+
+#### Pulsar Java 客户端配置项
+
+Pulsar 连接器使用[Java 客户端](https://pulsar.apache.org/docs/en/client-libraries-java/)来创建消费实例，相关的配置定义于 Pulsar 的 `ClientConfigurationData` 内。连接器在 `PulsarOptions` 选项中，定义大部分的可供用户定义的配置。
+
+{{< generated/pulsar_client_configuration >}}
+
+#### Pulsar 管理 API 配置项
+
+[管理 API](https://pulsar.apache.org/docs/en/admin-api-overview/) 用于查询 topic 的元数据和用正则订阅的时候的 topic 查找，它与
+Java 客户端共享大部分配置。下面列举的配置只供管理 API 使用，连接器也在 `PulsarOptions` 里予以定义。
+
+{{< generated/pulsar_admin_configuration >}}
+
+#### Pulsar 消费者 API 配置项
+
+Pulsar 提供了消费者 API 和读者 API 两套 API 来进行数据消费，它们可用于不同的业务场景。
+Flink 上的 Pulsar 连接器使用消费者 API 进行消费，它的配置定义于 Pulsar 的 `ConsumerConfigurationData` 内。连接器将其中大部分的可供用户定义的配置定义于 `PulsarSourceOptions` 内。
+
+{{< generated/pulsar_consumer_configuration >}}
+
+#### Pulsar 数据源配置项
+
+下述配置主要用于性能调优或者是控制消息确认的行为。如非必要，可以不用强制配置。
+
+{{< generated/pulsar_source_configuration >}}
+
+### 动态分区发现
+
+为了能在启动 Flink 任务之后还能发现在 Pulsar 上扩容的分区或者是新创建的 topic，连接器提供了动态分区发现机制。该机制不需要重启 Flink 任务。
+对选项 `PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS` 设置一个正整数即可启用。
+
+```java
+// 10 秒查询一次分区信息
+PulsarSource.builder()
+ .setConfig(PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS, 10000);
+```
+
+{{< hint warning >}}
+默认情况下，Pulsar 启用动态分区发现，查询间隔为 30 秒。用户可以给定一个负数，将该功能禁用。如果使用批的方式消费数据，将无法启用该功能。
+{{< /hint >}}
+
+### 事件时间和水位线
+
+默认情况下，连接器使用 Pulsar 的 `Message<byte[]>` 里面的时间作为解析结果的时间戳。用户可以使用 `WatermarkStrategy` 来自行解析出想要的消息时间，并向下游传递对应的水位线。
+
+```java
+env.fromSource(pulsarSource, new CustomWatermarkStrategy(), "Pulsar Source With Custom Watermark Strategy")
+```
+
+[这篇文档]({{< ref "docs/dev/datastream/event-time/generating_watermarks.md" >}}) 详细讲解了如何定义 `WatermarkStrategy`。
+
+### 消息确认
+
+一旦在 topic 上创建了订阅，消息便会[存储](https://pulsar.apache.org/docs/zh-CN/concepts-architecture-overview/#%E6%8C%81%E4%B9%85%E5%8C%96%E5%AD%98%E5%82%A8)在 Pulsar 里。即使没有消费者，消息也不会被丢弃。只有当连接器同 Pulsar 确认此条消息已经被消费，该消息才以某种机制会被移除。连接器支持四种订阅方式，它们的消息确认方式也大不相同。
+
+#### 独占和灾备订阅下的消息确认
+
+`独占` 和 `灾备` 订阅下,连接器使用累进式确认方式。确认某条消息已经被处理时，其前面被消费的消息会自动被置为已读。Pulsar 连接器会在 Flink 完成检查点时将对应时刻消费的消息置为已读，以此来保证 Pulsar 状态与 Flink 状态一致。
+
+如果用户没有在 Flink 上启用检查点，连接器可以使用周期性提交来将消费状态提交给 Pulsar，使用配置 `PulsarSourceOptions.PULSAR_AUTO_COMMIT_CURSOR_INTERVAL` 来进行定义。
+
+需要注意的是，此种场景下，Pulsar 连接器并不依赖于提交到 Pulsar 的状态来做容错。消息确认只是为了能在 Pulsar 端看到对应的消费处理情况。
+
+#### 共享和 key 共享订阅下的消息确认
+
+`共享` 和 `key 共享` 需要依次确认每一条消息，所以连接器在 Pulsar 事务里面进行消息确认，然后将事务提交到 Pulsar。
+
+首先需要在 Pulsar 的 `borker.conf` 文件里面启用事务：
+
+```text
+transactionCoordinatorEnabled=true
+```
+
+连接器创建的事务的默认超时时间为 3 小时，请确保这个时间大于 Flink 检查点的间隔。用户可以使用 `PulsarSourceOptions.PULSAR_TRANSACTION_TIMEOUT_MILLIS` 来设置事务的超时时间。
+
+如果用户无法启用 Pulsar 的事务，或者是因为项目禁用了检查点，需要将 `PulsarSourceOptions.PULSAR_ENABLE_AUTO_ACKNOWLEDGE_MESSAGE` 选项设置为 `true`，消息从 Pulsar 消费后会被立刻置为已读。连接器无法保证此种场景下的消息一致性。
+
+连接器在 Pulsar 上使用日志的形式记录某个事务下的消息确认，为了更好的性能，请缩短 Flink 做检查点的间隔。
+
+## 升级至最新的连接器
+
+常见的升级步骤，请参阅[升级应用程序和 Flink 版本]({{< ref "docs/ops/upgrading" >}})。Pulsar 连接器没有在 Flink 端存储消费的状态，所有的消费信息都推送到了 Pulsar。所以需要注意下面的事项：
+
+* 不要同时升级 Pulsar 连接器和 Pulsar 服务端的版本。
+* 使用最新版本的 Pulsar 客户端来消费消息。
+
+## 问题诊断
+
+使用 Flink 和 Pulsar 交互时如果遇到问题，一定要牢记 Flink 只使用了 Pulsar 的[Java 客户端](https://pulsar.apache.org/docs/zh-CN/client-libraries-java/) 和[管理 API](https://pulsar.apache.org/docs/zh-CN/admin-api-overview/)。用户遇到的问题很有可能与 Flink 无关，请先升级 Pulsar 的版本、Pulsar 客户端的版本、或者修改 Pulsar 的配置，Pulsar 连接器的配置来尝试解决问题。
+
+{{< top >}}
diff --git a/docs/content/docs/connectors/datastream/pulsar.md b/docs/content/docs/connectors/datastream/pulsar.md
new file mode 100644
index 0000000..511b39a
--- /dev/null
+++ b/docs/content/docs/connectors/datastream/pulsar.md
@@ -0,0 +1,420 @@
+---
+title: Pulsar
+weight: 9
+type: docs
+---
+<!--
+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.
+-->
+
+# Apache Pulsar Connector
+
+Flink provides an [Apache Pulsar](https://pulsar.apache.org) source connector for reading data from Pulsar topics with exactly-once guarantees.
+
+## Dependency
+
+You can use the connector with Pulsar 2.7.0 or higher. However, the Pulsar source connector supports
+Pulsar [transactions](https://pulsar.apache.org/docs/en/txn-what/),
+it is recommended to use Pulsar 2.8.0 or higher releases.
+For details on Pulsar compatibility, please refer to the [PIP-72](https://github.com/apache/pulsar/wiki/PIP-72%3A-Introduce-Pulsar-Interface-Taxonomy%3A-Audience-and-Stability-Classification).
+
+{{< artifact flink-connector-pulsar withScalaVersion >}}
+
+If you are using Pulsar source, `flink-connector-base` is also required as dependency:
+
+{{< artifact flink-connector-base >}}
+
+Flink's streaming connectors are not currently part of the binary distribution.
+See how to link with them for cluster execution [here]({{< ref "docs/dev/datastream/project-configuration" >}}).
+
+## Pulsar Source
+
+{{< hint info >}}
+This part describes the Pulsar source based on the new
+[data source]({{< ref "docs/dev/datastream/sources.md" >}}) API.
+
+If you want to use the legacy `SourceFunction` or on Flink 1.13 or lower releases, just use the StreamNative's [pulsar-flink](https://github.com/streamnative/pulsar-flink).
+{{< /hint >}}
+
+### Usage
+
+Pulsar source provides a builder class for constructing an instance of PulsarSource. The code snippet below shows
+how to build a PulsarSource to consume messages from the earliest cursor of topic "persistent://public/default/my-topic",
+with **Exclusive** subscription `my-subscription` and deserialize the raw payload of the messages as strings.
+
+```java
+PulsarSource<String> pulsarSource = PulsarSource.builder()
+ .setServiceUrl(serviceUrl)
+ .setAdminUrl(adminUrl)
+ .setStartCursor(StartCursor.earliest())
+ .setTopics("my-topic")
+ .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(new SimpleStringSchema()))
+ .setSubscriptionName("my-subscription")
+ .setSubscriptionType(SubscriptionType.Exclusive)
+ .build();
+
+env.fromSource(source, WatermarkStrategy.noWatermarks(), "Pulsar Source");
+```
+
+The following properties are **required** for building a PulsarSource:
+
+- Pulsar service url, configured by `setServiceUrl(String)`
+- Pulsar service http url (aka. admin url), configured by `setAdminUrl(String)`
+- Pulsar subscription name, configured by `setSubscriptionName(String)`
+- Topics / partitions to subscribe, see the following
+ [Topic-partition subscription](#topic-partition-subscription) for more details.
+- Deserializer to parse Pulsar messages, see the following
+ [Deserializer](#deserializer) for more details.
+
+### Topic-partition Subscription
+
+Pulsar source provide two ways of topic-partition subscription:
+
+- Topic list, subscribing messages from all partitions in a list of topics. For example:
+ ```java
+ PulsarSource.builder().setTopics("some-topic1", "some-topic2")
+
+ // Partition 0 and 2 of topic "topic-a"
+ PulsarSource.builder().setTopics("topic-a-partition-0", "topic-a-partition-2")
+ ```
+
+- Topic pattern, subscribing messages from all topics whose name matches the provided regular expression. For example:
+ ```java
+ PulsarSource.builder().setTopicPattern("topic-*")
+ ```
+
+#### Flexible Topic Naming
+
+Since Pulsar 2.0, all topic names internally have the form `{persistent|non-persistent}://tenant/namespace/topic`.
+Now, for partitioned topics, you can use short names in many cases (for the sake of simplicity).
+The flexible naming system stems from the fact that there is now a default topic type, tenant, and namespace in a Pulsar cluster.
+
+Topic property | Default
+:------------|:-------
+topic type | `persistent`
+tenant | `public`
+namespace | `default`
+
+This table lists a mapping relationship between your input topic name and translated topic name:
+
+Input topic name | Translated topic name
+:----------------|:---------------------
+`my-topic` | `persistent://public/default/my-topic`
+`my-tenant/my-namespace/my-topic` | `persistent://my-tenant/my-namespace/my-topic`
+
+{{< hint warning >}}
+For non-persistent topics, you need to continue to specify the entire topic name,
+as the default-based rules do not apply for non-partitioned topics.
+Thus, you cannot use a short name like `non-persistent://my-topic` and would need to use `non-persistent://public/default/my-topic` instead.
+{{< /hint >}}
+
+#### Subscribing Pulsar Topic Partition
+
+Internally, Pulsar divides a partitioned topic as a set of non-partitioned topics according to the partition size.
+
+For example, if a `simple-string` topic with 3 partitions is created under the `sample` tenant with `flink` namespace.
+The topics on Pulsar would be:
+
+Topic name | Partitioned
+:--------- | :----------
+`persistent://sample/flink/simple-string` | Y
+`persistent://sample/flink/simple-string-partition-0` | N
+`persistent://sample/flink/simple-string-partition-1` | N
+`persistent://sample/flink/simple-string-partition-2` | N
+
+You can directly consume messages from the topic partitions by using the non-partitioned topic names above.
+For example, use `PulsarSource.builder().setTopics("sample/flink/simple-string-partition-1", "sample/flink/simple-string-partition-2")` would consume the partitions 1 and 2 of the `sample/flink/simple-string` topic.
+
+#### RegexSubscriptionMode for Topic Pattern
+
+Pulsar connector extracts the topic type (`persistent` or `non-persistent`) from the given topic pattern.
+For example, `PulsarSource.builder().setTopicPattern("non-persistent://my-topic*")` would be `non-persistent`.
+The topic type would be `persistent` if you do not provide the topic type in the regular expression.
+
+To consume both `persistent` and `non-persistent` topics based on the topic pattern,
+you can use `setTopicPattern("topic-*", RegexSubscriptionMode.AllTopics)`.
+Pulsar connector would filter the available topics by the `RegexSubscriptionMode`.
+
+### Deserializer
+
+A deserializer (Deserialization schema) is required for parsing Pulsar messages. The deserializer is
+configured by `setDeserializationSchema(PulsarDeserializationSchema)`.
+The `PulsarDeserializationSchema` defines how to deserialize a Pulsar `Message<byte[]>`.
+
+If only the raw payload of a message (message data in bytes) is needed,
+you can use the predefined `PulsarDeserializationSchema`. Pulsar connector provides three types of implementation.
+
+- Decode the message by using Pulsar's [Schema](https://pulsar.apache.org/docs/en/schema-understand/).
+ ```java
+ // Primitive types
+ PulsarDeserializationSchema.pulsarSchema(Schema)
+
+ // Struct types (JSON, Protobuf, Avro, etc.)
+ PulsarDeserializationSchema.pulsarSchema(Schema, Class)
+
+ // KeyValue type
+ PulsarDeserializationSchema.pulsarSchema(Schema, Class, Class)
+ ```
+- Decode the message by using Flink's `DeserializationSchema`
+ ```java
+ PulsarDeserializationSchema.flinkSchema(DeserializationSchema)
+ ```
+- Decode the message by using Flink's `TypeInformation`
+ ```java
+ PulsarDeserializationSchema.flinkTypeInfo(TypeInformation, ExecutionConfig)
+ ```
+
+Pulsar `Message<byte[]>` contains some [extra properties](https://pulsar.apache.org/docs/en/concepts-messaging/#messages),
+such as message key, message publish time, message time, application defined key/value pairs that will be attached to the message, etc.
+These properties could be acquired by the `Message<byte[]>` interface.
+
+If you want to deserialize the Pulsar message by these properties, you need to implement `PulsarDeserializationSchema`.
+And ensure that the `TypeInformation` from the `PulsarDeserializationSchema.getProducedType()` must be correct.
+Flink would use this `TypeInformation` for passing the messages to downstream operators.
+
+### Pulsar Subscriptions
+
+A Pulsar subscription is a named configuration rule that determines how messages are delivered to Flink readers.
+The subscription name is required for consuming messages. Pulsar connector supports four subscription types:
+
+- [Exclusive](https://pulsar.apache.org/docs/en/concepts-messaging/#exclusive)
+- [Shared](https://pulsar.apache.org/docs/en/concepts-messaging/#shared)
+- [Failover](https://pulsar.apache.org/docs/en/concepts-messaging/#failover)
+- [Key_Shared](https://pulsar.apache.org/docs/en/concepts-messaging/#key_shared)
+
+There is no difference between `Exclusive` and `Failover` in the Pulsar connector.
+When a Flink reader crashes, all (non-acknowledged and subsequent) messages are redelivered to the available Flink readers.
+
+By default, if no subscription type is defined, Pulsar source uses `Shared` subscription.
+
+```java
+// Shared subscription with name "my-shared"
+PulsarSource.builder().setSubscriptionName("my-shared")
+
+// Exclusive subscription with name "my-exclusive"
+PulsarSource.builder().setSubscriptionName("my-exclusive").setSubscriptionType(SubscriptionType.Exclusive)
+```
+
+If you want to use `Key_Shared` subscription type on the Pulsar connector. Ensure that you provide a `RangeGenerator` implementation.
+The `RangeGenerator` generates a set of key hash ranges so that
+a respective reader subtask will only dispatch messages where the hash of the message key is contained in the specified range.
+
+Pulsar connector would use a `UniformRangeGenerator` which would divides the range by the Flink source parallelism
+if no `RangeGenerator` is provided in the `Key_Shared` subscription type.
+
+### Starting Position
+
+Pulsar source is able to consume messages starting from different positions by `setStartCursor(StartCursor)`.
+Built-in start cursors include:
+
+- Start from the earliest available message in the topic.
+ ```java
+ StartCursor.earliest()
+ ```
+- Start from the latest available message in the topic.
+ ```java
+ StartCursor.latest()
+ ```
+- Start from a specified message between the earliest and the latest.
+ Pulsar connector would consume from the latest available message if the message id doesn't exist.
+
+ The start message is included in consuming result.
+ ```java
+ StartCursor.fromMessageId(MessageId)
+ ```
+- Start from a specified message between the earliest and the latest.
+ Pulsar connector would consume from the latest available message if the message id doesn't exist.
+
+ Include or exclude the start message by using the second boolean parameter.
+ ```java
+ StartCursor.fromMessageId(MessageId, boolean)
+ ```
+
+{{< hint info >}}
+Each Pulsar message belongs to an ordered sequence on its topic.
+The sequence ID (`MessageId`) of the message is ordered in that sequence.
+`MessageId` contains some extra information (the ledger, entry, partition) on how the message is stored,
+you can create a `MessageId` by using `DefaultImplementation.newMessageId(long ledgerId, long entryId, int partitionIndex)`.
+{{< /hint >}}
+
+### Boundedness
+
+Pulsar source supports streaming and batch running modes.
+By default, the `PulsarSource` is set to run in the streaming mode.
+
+In streaming mode, Pulsar source never stops until a Flink job fails or is cancelled. However,
+you can set Pulsar source stopping at a stop position by using ```setUnboundedStopCursor(StopCursor)```.
+The Pulsar source will finish when all partitions reach their specified stop positions.
+
+You can use ```setBoundedStopCursor(StopCursor)``` to specify a stop position so that the Pulsar source can run in the batch mode.
+When all partitions have reached their stop positions, the source will finish.
+
+Built-in stop cursors include:
+
+- Connector will never stop consuming.
+ ```java
+ StopCursor.never()
+ ```
+- Stop at the latest available message in Pulsar when the connector starts consuming.
+ ```java
+ StopCursor.latest()
+ ```
+- Stop when connector meet a given message, or stop at a message which is produced after this given message.
+ ```java
+ StopCursor.atMessageId(MessageId)
+ ```
+- Stop but include the given message in consuming result.
+ ```java
+ StopCursor.afterMessageId(MessageId)
+ ```
+- Stop at the specified message time by `Message<byte[]>.getEventTime()`.
+ ```java
+ StopCursor.atEventTime(long)
+ ```
+
+### Configurable Options
+
+In addition to configuration options described above, you can set arbitrary options for `PulsarClient`,
+`PulsarAdmin`， Pulsar `Consumer` and `PulsarSource` by using `setConfig(ConfigOption<T>, T)` and `setConfig(Configuration)`.
+
+#### PulsarClient Options
+
+Pulsar connector use the [client API](https://pulsar.apache.org/docs/en/client-libraries-java/)
+to create the `Consumer` instance. Pulsar connector extracts most parts of Pulsar's `ClientConfigurationData`,
+which is required for creating a `PulsarClient`, as Flink configuration options in `PulsarOptions`.
+
+{{< generated/pulsar_client_configuration >}}
+
+#### PulsarAdmin Options
+
+The [admin API](https://pulsar.apache.org/docs/en/admin-api-overview/) is used for querying topic metadata
+and for discovering the desired topics when Pulsar connector uses topic pattern subscription. It would share most part of the
+configuration options with the client API. The configuration options listed here are only used in the admin API.
+They are also defined in `PulsarOptions`.
+
+{{< generated/pulsar_admin_configuration >}}
+
+#### Pulsar Consumer Options
+
+In general, Pulsar provides the Reader API and Consumer API for consuming messages in different scenarios.
+Flink's Pulsar connector uses the Consumer API. It extracts most parts of Pulsar's `ConsumerConfigurationData` as Flink configuration options in `PulsarSourceOptions`.
+
+{{< generated/pulsar_consumer_configuration >}}
+
+#### PulsarSource Options
+
+The configuration options below are mainly used for customizing the performance and message acknowledgement behavior.
+You can just leave them alone if you do not meet any performance issues.
+
+{{< generated/pulsar_source_configuration >}}
+
+### Dynamic Partition Discovery
+
+To handle scenarios like topic scaling-out or topic creation without restarting the Flink
+job, Pulsar source can be configured to periodically discover new partitions under provided
+topic-partition subscribing pattern. To enable partition discovery, set a non-negative value for
+the option `PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS`:
+
+```java
+// discover new partitions per 10 seconds
+PulsarSource.builder()
+ .setConfig(PulsarSourceOptions.PULSAR_PARTITION_DISCOVERY_INTERVAL_MS, 10000);
+```
+
+{{< hint warning >}}
+- Partition discovery is **enabled** by default. Pulsar connector would query the topic metadata every 30 seconds.
+- You need to set the partition discovery interval to a negative value to disable this feature.
+- The partition discovery would be disabled in batch mode even if you set this option with a non-negative value.
+{{< /hint >}}
+
+### Event Time and Watermarks
+
+By default, the message uses the timestamp embedded in Pulsar `Message<byte[]>` as the event time.
+You can define your own `WatermarkStrategy` to extract the event time from the message,
+and emit the watermark downstream:
+
+```java
+env.fromSource(pulsarSource, new CustomWatermarkStrategy(), "Pulsar Source With Custom Watermark Strategy")
+```
+
+[This documentation]({{< ref "docs/dev/datastream/event-time/generating_watermarks.md" >}}) describes
+details about how to define a `WatermarkStrategy`.
+
+### Message Acknowledgement
+
+When a subscription is created, Pulsar [retains](https://pulsar.apache.org/docs/en/concepts-architecture-overview/#persistent-storage) all messages, even if the consumer is disconnected.
+The retained messages are discarded only when the connector acknowledges that all these messages are processed successfully.
+Pulsar connector supports four subscription types, which makes the acknowledgement behaviors variety among different subscriptions.
+
+#### Acknowledgement on Exclusive and Failover Subscription Types
+
+`Exclusive` and `Failover` subscription types support cumulative acknowledgment. In these subscription types, Flink only needs to acknowledge
+the latest successfully consumed message. All the message before the given message are marked
+with a consumed status.
+
+Pulsar source acknowledges the current consuming message when checkpoints are **completed**,
+to ensure the consistency between Flink's checkpoint state and committed position on Pulsar brokers.
+
+If checkpointing is disabled, Pulsar source periodically acknowledges messages.
+You can set the acknowledgement period by using the `PulsarSourceOptions.PULSAR_AUTO_COMMIT_CURSOR_INTERVAL` option.
+
+Pulsar source does **NOT** rely on committed positions for fault tolerance.
+Acknowledging messages is only for exposing the progress of consumer and monitoring on these two subscription types.
+
+#### Acknowledgement on Shared and Key_Shared Subscription Types
+
+`Shared` and `Key_Shared` subscription types need to acknowledge messages one by one. You can acknowledge
+a message in a transaction and commit it to Pulsar.
+
+You should enable transaction in the Pulsar `borker.conf` file when using these two subscription types in connector:
+
+```text
+transactionCoordinatorEnabled=true
+```
+
+Pulsar transaction would be created with 3 hours as the timeout by default. Make sure that timeout > checkpoint interval + maximum recovery time.
+A shorter checkpoint interval would increase the consuming performance.
+You can change the transaction timeout by using the `PulsarSourceOptions.PULSAR_TRANSACTION_TIMEOUT_MILLIS` option.
+
+If checkpointing is disabled or you can not enable the transaction on Pulsar broker, you should set
+`PulsarSourceOptions.PULSAR_ENABLE_AUTO_ACKNOWLEDGE_MESSAGE` to `true`.
+The message would be immediately acknowledged after consuming.
+We can not promise consistency in this scenario.
+
+{{< hint info >}}
+All acknowledgements in a transaction are recorded in the Pulsar broker side.
+{{< /hint >}}
+
+## Upgrading to the Latest Connector Version
+
+The generic upgrade steps are outlined in [upgrading jobs and Flink versions guide]({{< ref "docs/ops/upgrading" >}}).
+The Pulsar connector does not store any state on the Flink side. The Pulsar connector pushes and stores all the states on the Pulsar side.
+For Pulsar, you additionally need to know these limitations:
+
+* Do not upgrade the Pulsar connector and Pulsar broker version at the same time.
+* Always use a newer Pulsar client with Pulsar connector for consuming message from Pulsar.
+
+## Troubleshooting
+
+If you have a problem with Pulsar when using Flink, keep in mind that Flink only wraps
+[PulsarClient](https://pulsar.apache.org/docs/en/client-libraries-java/) or
+[PulsarAdmin](https://pulsar.apache.org/docs/en/admin-api-overview/)
+and your problem might be independent of Flink and sometimes can be solved by upgrading Pulsar brokers,
+reconfiguring Pulsar brokers or reconfiguring Pulsar connector in Flink.
+
+{{< top >}}
[flink] 02/04: [FLINK-23864][connector/pulsar] Release Pulsar
Message if user enable poolMessage option.
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d40cfc6ce62873be6de9e6313fd6dac34688f089
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Fri Sep 10 12:10:12 2021 +0800
[FLINK-23864][connector/pulsar] Release Pulsar Message if user enable poolMessage option.
---
.../pulsar/source/reader/split/PulsarOrderedPartitionSplitReader.java | 3 +++
.../source/reader/split/PulsarUnorderedPartitionSplitReader.java | 3 +++
2 files changed, 6 insertions(+)
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarOrderedPartitionSplitReader.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarOrderedPartitionSplitReader.java
index ac1821b..af650ea 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarOrderedPartitionSplitReader.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarOrderedPartitionSplitReader.java
@@ -74,6 +74,9 @@ public class PulsarOrderedPartitionSplitReader<OUT> extends PulsarPartitionSplit
protected void finishedPollMessage(Message<byte[]> message) {
// Nothing to do here.
LOG.debug("Finished polling message {}", message);
+
+ // Release message
+ message.release();
}
@Override
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarUnorderedPartitionSplitReader.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarUnorderedPartitionSplitReader.java
index efa383f..9c8e3d7 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarUnorderedPartitionSplitReader.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarUnorderedPartitionSplitReader.java
@@ -118,6 +118,9 @@ public class PulsarUnorderedPartitionSplitReader<OUT> extends PulsarPartitionSpl
if (sourceConfiguration.isEnableAutoAcknowledgeMessage()) {
sneakyClient(() -> pulsarConsumer.acknowledge(message));
}
+
+ // Release message
+ message.release();
}
@Override
[flink] 01/04: [FLINK-23864][docs] Add flink-connector-pulsar
module to flink-docs, auto generate the config document.
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 77012db20dc4728732acff22b3d0da7a4b57c7f4
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Fri Sep 10 12:09:54 2021 +0800
[FLINK-23864][docs] Add flink-connector-pulsar module to flink-docs, auto generate the config document.
---
.../generated/pulsar_admin_configuration.html | 42 ++++
.../generated/pulsar_client_configuration.html | 222 ++++++++++++++++++++
.../generated/pulsar_consumer_configuration.html | 174 +++++++++++++++
.../generated/pulsar_source_configuration.html | 54 +++++
.../pulsar/common/config/PulsarOptions.java | 153 ++++++++------
.../pulsar/source/PulsarSourceOptions.java | 233 ++++++++++-----------
flink-docs/pom.xml | 6 +
.../configuration/ConfigOptionsDocGenerator.java | 8 +-
8 files changed, 698 insertions(+), 194 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/pulsar_admin_configuration.html b/docs/layouts/shortcodes/generated/pulsar_admin_configuration.html
new file mode 100644
index 0000000..31c7273
--- /dev/null
+++ b/docs/layouts/shortcodes/generated/pulsar_admin_configuration.html
@@ -0,0 +1,42 @@
+<table class="configuration table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Key</th>
+ <th class="text-left" style="width: 15%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 55%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>pulsar.admin.adminUrl</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The Pulsar service HTTP URL for the admin endpoint. For example, <code class="highlighter-rouge">http://my-broker.example.com:8080</code>, or <code class="highlighter-rouge">https://my-broker.example.com:8443</code> for TLS.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.admin.autoCertRefreshTime</h5></td>
+ <td style="word-wrap: break-word;">300000</td>
+ <td>Integer</td>
+ <td>The auto cert refresh time (in ms) if Pulsar admin supports TLS authentication.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.admin.connectTimeout</h5></td>
+ <td style="word-wrap: break-word;">60000</td>
+ <td>Integer</td>
+ <td>The connection time out (in ms) for the PulsarAdmin client.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.admin.readTimeout</h5></td>
+ <td style="word-wrap: break-word;">60000</td>
+ <td>Integer</td>
+ <td>The server response read timeout (in ms) for the PulsarAdmin client for any request.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.admin.requestTimeout</h5></td>
+ <td style="word-wrap: break-word;">300000</td>
+ <td>Integer</td>
+ <td>The server request timeout (in ms) for the PulsarAdmin client for any request.</td>
+ </tr>
+ </tbody>
+</table>
diff --git a/docs/layouts/shortcodes/generated/pulsar_client_configuration.html b/docs/layouts/shortcodes/generated/pulsar_client_configuration.html
new file mode 100644
index 0000000..d44a93c
--- /dev/null
+++ b/docs/layouts/shortcodes/generated/pulsar_client_configuration.html
@@ -0,0 +1,222 @@
+<table class="configuration table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Key</th>
+ <th class="text-left" style="width: 15%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 55%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>pulsar.client.authParamMap</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Map</td>
+ <td>Parameters for the authentication plugin.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.authParams</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Parameters for the authentication plugin.<br /><br />Example:<br /><code class="highlighter-rouge">key1:val1,key2:val2</code></td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.authPluginClassName</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Name of the authentication plugin.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.concurrentLookupRequest</h5></td>
+ <td style="word-wrap: break-word;">5000</td>
+ <td>Integer</td>
+ <td>The number of concurrent lookup requests allowed to send on each broker connection to prevent overload on the broker. It should be configured with a higher value only in case of it requires to produce or subscribe on thousands of topic using a created <code class="highlighter-rouge">PulsarClient</code></td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.connectionTimeoutMs</h5></td>
+ <td style="word-wrap: break-word;">10000</td>
+ <td>Integer</td>
+ <td>Duration (in ms) of waiting for a connection to a broker to be established.<br />If the duration passes without a response from a broker, the connection attempt is dropped.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.connectionsPerBroker</h5></td>
+ <td style="word-wrap: break-word;">1</td>
+ <td>Integer</td>
+ <td>The maximum number of connections that the client library will open to a single broker.<br /> By default, the connection pool will use a single connection for all the producers and consumers. Increasing this parameter may improve throughput when using many producers over a high latency connection.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.enableBusyWait</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Option to enable busy-wait settings.<br />This option will enable spin-waiting on executors and IO threads in order to reduce latency during context switches. The spinning will consume 100% CPU even when the broker is not doing any work. It is recommended to reduce the number of IO threads and BookKeeper client threads to only have fewer CPU cores busy.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.enableTransaction</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>If transaction is enabled, start the <code class="highlighter-rouge">transactionCoordinatorClient</code> with <code class="highlighter-rouge">PulsarClient</code>.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.initialBackoffIntervalNanos</h5></td>
+ <td style="word-wrap: break-word;">100000000</td>
+ <td>Long</td>
+ <td>Default duration (in nanoseconds) for a backoff interval.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.keepAliveIntervalSeconds</h5></td>
+ <td style="word-wrap: break-word;">30</td>
+ <td>Integer</td>
+ <td>Interval (in seconds) for keeping connection between the Pulsar client and broker alive.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.listenerName</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Configure the <code class="highlighter-rouge">listenerName</code> that the broker will return the corresponding <code class="highlighter-rouge">advertisedListener</code>.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.maxBackoffIntervalNanos</h5></td>
+ <td style="word-wrap: break-word;">60000000000</td>
+ <td>Long</td>
+ <td>The maximum duration (in nanoseconds) for a backoff interval.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.maxLookupRedirects</h5></td>
+ <td style="word-wrap: break-word;">20</td>
+ <td>Integer</td>
+ <td>The maximum number of times a lookup-request redirections to a broker.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.maxLookupRequest</h5></td>
+ <td style="word-wrap: break-word;">50000</td>
+ <td>Integer</td>
+ <td>The maximum number of lookup requests allowed on each broker connection to prevent overload on the broker. It should be greater than <code class="highlighter-rouge">maxConcurrentLookupRequests</code>. Requests that inside <code class="highlighter-rouge">maxConcurrentLookupRequests</code> are already sent to broker, and requests beyond <code class="highlighter-rouge">maxConcurrentLookupRequests</code> and under <code class="highlighter-rouge">maxLookupRequests</code> will [...]
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.maxNumberOfRejectedRequestPerConnection</h5></td>
+ <td style="word-wrap: break-word;">50</td>
+ <td>Integer</td>
+ <td>The maximum number of rejected requests of a broker in a certain period (30s) after the current connection is closed and the client creates a new connection to connect to a different broker.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.memoryLimitBytes</h5></td>
+ <td style="word-wrap: break-word;">0</td>
+ <td>Long</td>
+ <td>The limit (in bytes) on the amount of direct memory that will be allocated by this client instance.<br />Note: at this moment this is only limiting the memory for producers. Setting this to <code class="highlighter-rouge">0</code> will disable the limit.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.numIoThreads</h5></td>
+ <td style="word-wrap: break-word;">1</td>
+ <td>Integer</td>
+ <td>The number of threads used for handling connections to brokers.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.numListenerThreads</h5></td>
+ <td style="word-wrap: break-word;">1</td>
+ <td>Integer</td>
+ <td>The number of threads used for handling message listeners. The listener thread pool is shared across all the consumers and readers that are using a <code class="highlighter-rouge">listener</code> model to get messages. For a given consumer, the listener is always invoked from the same thread to ensure ordering.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.operationTimeoutMs</h5></td>
+ <td style="word-wrap: break-word;">30000</td>
+ <td>Integer</td>
+ <td>Operation timeout (in ms). Operations such as creating producers, subscribing or unsubscribing topics are retried during this interval. If the operation is not completed during this interval, the operation will be marked as failed.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.proxyProtocol</h5></td>
+ <td style="word-wrap: break-word;">SNI</td>
+ <td><p>Enum</p></td>
+ <td>Protocol type to determine the type of proxy routing when a client connects to the proxy using <code class="highlighter-rouge">pulsar.client.proxyServiceUrl</code>.<br /><br />Possible values:<ul><li>"SNI"</li></ul></td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.proxyServiceUrl</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Proxy-service URL when a client connects to the broker via the proxy. The client can choose the type of proxy-routing.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.requestTimeoutMs</h5></td>
+ <td style="word-wrap: break-word;">60000</td>
+ <td>Integer</td>
+ <td>Maximum duration (in ms) for completing a request.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.serviceUrl</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Service URL provider for Pulsar service.<br />To connect to Pulsar using client libraries, you need to specify a Pulsar protocol URL.<br />You can assign Pulsar protocol URLs to specific clusters and use the <code class="highlighter-rouge">pulsar</code> scheme.<br /><ul><li>This is an example of <code class="highlighter-rouge">localhost</code>: <code class="highlighter-rouge">pulsar://localhost:6650</code>.</li><li>If you have multiple brokers, the URL is as: <code class= [...]
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.sslProvider</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The name of the security provider used for SSL connections. The default value is the default security provider of the JVM.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.statsIntervalSeconds</h5></td>
+ <td style="word-wrap: break-word;">60</td>
+ <td>Long</td>
+ <td>Interval between each stats info.<br /><ul><li>Stats is activated with positive <code class="highlighter-rouge">statsInterval</code></li><li>Set <code class="highlighter-rouge">statsIntervalSeconds</code> to 1 second at least.</li></ul></td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.tlsAllowInsecureConnection</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether the Pulsar client accepts untrusted TLS certificate from the broker.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.tlsCiphers</h5></td>
+ <td style="word-wrap: break-word;"></td>
+ <td>List<String></td>
+ <td>A list of cipher suites. This is a named combination of authentication, encryption, MAC and the key exchange algorithm used to negotiate the security settings for a network connection using the TLS or SSL network protocol. By default all the available cipher suites are supported.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.tlsHostnameVerificationEnable</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to enable TLS hostname verification. It allows to validate hostname verification when a client connects to the broker over TLS. It validates incoming x509 certificate and matches provided hostname (CN/SAN) with the expected broker's host name. It follows RFC 2818, 3.1. Server Identity hostname verification.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.tlsProtocols</h5></td>
+ <td style="word-wrap: break-word;"></td>
+ <td>List<String></td>
+ <td>The SSL protocol used to generate the SSLContext. By default, it is set TLS, which is fine for most cases. Allowed values in recent JVMs are TLS, TLSv1.3, TLSv1.2 and TLSv1.1.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.tlsTrustCertsFilePath</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Path to the trusted TLS certificate file.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.tlsTrustStorePassword</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The store password for the key store file.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.tlsTrustStorePath</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The location of the trust store file.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.tlsTrustStoreType</h5></td>
+ <td style="word-wrap: break-word;">"JKS"</td>
+ <td>String</td>
+ <td>The file format of the trust store file.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.useKeyStoreTls</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>If TLS is enabled, whether use the KeyStore type as the TLS configuration parameter. If it is set to <code class="highlighter-rouge">false</code>, it means to use the default pem type configuration.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.client.useTcpNoDelay</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Whether to use the TCP no-delay flag on the connection to disable Nagle algorithm.<br />No-delay features ensures that packets are sent out on the network as soon as possible, and it is critical to achieve low latency publishes. On the other hand, sending out a huge number of small packets might limit the overall throughput. Therefore, if latency is not a concern, it is recommended to set the <code class="highlighter-rouge">useTcpNoDelay</code> flag to <code class="highli [...]
+ </tr>
+ </tbody>
+</table>
diff --git a/docs/layouts/shortcodes/generated/pulsar_consumer_configuration.html b/docs/layouts/shortcodes/generated/pulsar_consumer_configuration.html
new file mode 100644
index 0000000..bc8b6df
--- /dev/null
+++ b/docs/layouts/shortcodes/generated/pulsar_consumer_configuration.html
@@ -0,0 +1,174 @@
+<table class="configuration table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Key</th>
+ <th class="text-left" style="width: 15%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 55%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>pulsar.consumer.ackReceiptEnabled</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Acknowledgement will return a receipt but this does not mean that the message will not be resent after getting the receipt.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.consumer.ackTimeoutMillis</h5></td>
+ <td style="word-wrap: break-word;">0</td>
+ <td>Long</td>
+ <td>The timeout (in ms) for unacknowledged messages, truncated to the nearest millisecond. The timeout needs to be greater than 1 second.<br />By default, the acknowledge timeout is disabled and that means that messages delivered to a consumer will not be re-delivered unless the consumer crashes.<br />When acknowledgement timeout being enabled, if a message is not acknowledged within the specified timeout it will be re-delivered to the consumer (possibly to a different consum [...]
+ </tr>
+ <tr>
+ <td><h5>pulsar.consumer.acknowledgementsGroupTimeMicros</h5></td>
+ <td style="word-wrap: break-word;">100000</td>
+ <td>Long</td>
+ <td>Group a consumer acknowledgment for a specified time (in μs). By default, a consumer uses <code class="highlighter-rouge">100μs</code> grouping time to send out acknowledgments to a broker. If the group time is set to <code class="highlighter-rouge">0</code>, acknowledgments are sent out immediately. A longer ack group time is more efficient at the expense of a slight increase in message re-deliveries after a failure.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.consumer.autoAckOldestChunkedMessageOnQueueFull</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Buffering a large number of outstanding uncompleted chunked messages can bring memory pressure and it can be guarded by providing this <code class="highlighter-rouge">pulsar.consumer.maxPendingChunkedMessage</code> threshold. Once a consumer reaches this threshold, it drops the outstanding unchunked-messages by silently acknowledging if <code class="highlighter-rouge">pulsar.consumer.autoAckOldestChunkedMessageOnQueueFull</code> is true. Otherwise, it marks them for redel [...]
+ </tr>
+ <tr>
+ <td><h5>pulsar.consumer.autoUpdatePartitionsIntervalSeconds</h5></td>
+ <td style="word-wrap: break-word;">60</td>
+ <td>Integer</td>
+ <td>The interval (in seconds) of updating partitions. This only works if autoUpdatePartitions is enabled.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.consumer.consumerName</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The consumer name is informative and it can be used to identify a particular consumer instance from the topic stats.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.consumer.cryptoFailureAction</h5></td>
+ <td style="word-wrap: break-word;">FAIL</td>
+ <td><p>Enum</p></td>
+ <td>The consumer should take action when it receives a message that can not be decrypted.<ul><li><code class="highlighter-rouge">FAIL</code>: this is the default option to fail messages until crypto succeeds.</li><li><code class="highlighter-rouge">DISCARD</code>: silently acknowledge but do not deliver messages to an application.</li><li><code class="highlighter-rouge">CONSUME</code>: deliver encrypted messages to applications. It is the application's responsibility to decry [...]
+ </tr>
+ <tr>
+ <td><h5>pulsar.consumer.deadLetterPolicy.deadLetterTopic</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Name of the dead topic where the failed messages are sent.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.consumer.deadLetterPolicy.maxRedeliverCount</h5></td>
+ <td style="word-wrap: break-word;">0</td>
+ <td>Integer</td>
+ <td>The maximum number of times that a message are redelivered before being sent to the dead letter queue.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.consumer.deadLetterPolicy.retryLetterTopic</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Name of the retry topic where the failed messages are sent.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.consumer.expireTimeOfIncompleteChunkedMessageMillis</h5></td>
+ <td style="word-wrap: break-word;">60000</td>
+ <td>Long</td>
+ <td>If a producer fails to publish all the chunks of a message, the consumer can expire incomplete chunks if the consumer cannot receive all chunks in expire times (default 1 hour, in ms).</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.consumer.maxPendingChunkedMessage</h5></td>
+ <td style="word-wrap: break-word;">10</td>
+ <td>Integer</td>
+ <td>The consumer buffers chunk messages into memory until it receives all the chunks of the original message. While consuming chunk-messages, chunks from the same message might not be contiguous in the stream and they might be mixed with other messages' chunks. So, consumer has to maintain multiple buffers to manage chunks coming from different messages. This mainly happens when multiple publishers are publishing messages on the topic concurrently or publishers failed to publ [...]
+ </tr>
+ <tr>
+ <td><h5>pulsar.consumer.maxTotalReceiverQueueSizeAcrossPartitions</h5></td>
+ <td style="word-wrap: break-word;">50000</td>
+ <td>Integer</td>
+ <td>The maximum total receiver queue size across partitions.<br />This setting reduces the receiver queue size for individual partitions if the total receiver queue size exceeds this value.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.consumer.negativeAckRedeliveryDelayMicros</h5></td>
+ <td style="word-wrap: break-word;">60000000</td>
+ <td>Long</td>
+ <td>Delay (in μs) to wait before redelivering messages that failed to be processed.<br />When an application uses <code class="highlighter-rouge">Consumer.negativeAcknowledge(Message)</code>, failed messages are redelivered after a fixed timeout.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.consumer.poolMessages</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Enable pooling of messages and the underlying data buffers.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.consumer.priorityLevel</h5></td>
+ <td style="word-wrap: break-word;">0</td>
+ <td>Integer</td>
+ <td>Priority level for a consumer to which a broker gives more priorities while dispatching messages in the shared subscription type.<br />The broker follows descending priorities. For example, 0=max-priority, 1, 2,...<br />In shared subscription mode, the broker first dispatches messages to the consumers on the highest priority level if they have permits. Otherwise, the broker considers consumers on the next priority level.<br /><br />Example 1<br />If a subscription has con [...]
+C1, 0, 2
+C2, 0, 1
+C3, 0, 1
+C4, 1, 2
+C5, 1, 1
+<br />The order in which a broker dispatches messages to consumers is: C1, C2, C3, C1, C4, C5, C4.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.consumer.properties</h5></td>
+ <td style="word-wrap: break-word;"></td>
+ <td>Map</td>
+ <td>A name or value property of this consumer. <code class="highlighter-rouge">properties</code> is application defined metadata attached to a consumer. When getting a topic stats, associate this metadata with the consumer stats for easier identification.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.consumer.readCompacted</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>If enabling <code class="highlighter-rouge">readCompacted</code>, a consumer reads messages from a compacted topic rather than reading a full message backlog of a topic.<br />A consumer only sees the latest value for each key in the compacted topic, up until reaching the point in the topic message when compacting backlog. Beyond that point, send messages as normal.<br />Only enabling <code class="highlighter-rouge">readCompacted</code> on subscriptions to persistent topic [...]
+ </tr>
+ <tr>
+ <td><h5>pulsar.consumer.receiverQueueSize</h5></td>
+ <td style="word-wrap: break-word;">1000</td>
+ <td>Integer</td>
+ <td>Size of a consumer's receiver queue.<br />For example, the number of messages accumulated by a consumer before an application calls <code class="highlighter-rouge">Receive</code>.<br />A value higher than the default value increases consumer throughput, though at the expense of more memory utilization.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.consumer.replicateSubscriptionState</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>If <code class="highlighter-rouge">replicateSubscriptionState</code> is enabled, a subscription state is replicated to geo-replicated clusters.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.consumer.retryEnable</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>If enabled, the consumer will automatically retry messages.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.consumer.subscriptionInitialPosition</h5></td>
+ <td style="word-wrap: break-word;">Latest</td>
+ <td><p>Enum</p></td>
+ <td>Initial position at which to set cursor when subscribing to a topic at first time.<br /><br />Possible values:<ul><li>"Latest"</li><li>"Earliest"</li></ul></td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.consumer.subscriptionMode</h5></td>
+ <td style="word-wrap: break-word;">Durable</td>
+ <td><p>Enum</p></td>
+ <td>Select the subscription mode to be used when subscribing to the topic.<ul><li><code class="highlighter-rouge">Durable</code>: Make the subscription to be backed by a durable cursor that will retain messages and persist the current position.</li><li><code class="highlighter-rouge">NonDurable</code>: Lightweight subscription mode that doesn't have a durable cursor associated</li></ul><br /><br />Possible values:<ul><li>"Durable"</li><li>"NonDurable"</li></ul></td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.consumer.subscriptionName</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Specify the subscription name for this consumer. This argument is required when constructing the consumer.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.consumer.subscriptionType</h5></td>
+ <td style="word-wrap: break-word;">Shared</td>
+ <td><p>Enum</p></td>
+ <td>Subscription type.<br /><br />Four subscription types are available:<ul><li>Exclusive</li><li>Failover</li><li>Shared</li><li>Key_Shared</li></ul><br /><br />Possible values:<ul><li>"Exclusive"</li><li>"Shared"</li><li>"Failover"</li><li>"Key_Shared"</li></ul></td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.consumer.tickDurationMillis</h5></td>
+ <td style="word-wrap: break-word;">1000</td>
+ <td>Long</td>
+ <td>Granularity (in ms) of the ack-timeout redelivery.<br />A greater (for example, 1 hour) <code class="highlighter-rouge">tickDurationMillis</code> reduces the memory overhead to track messages.</td>
+ </tr>
+ </tbody>
+</table>
diff --git a/docs/layouts/shortcodes/generated/pulsar_source_configuration.html b/docs/layouts/shortcodes/generated/pulsar_source_configuration.html
new file mode 100644
index 0000000..02e512b
--- /dev/null
+++ b/docs/layouts/shortcodes/generated/pulsar_source_configuration.html
@@ -0,0 +1,54 @@
+<table class="configuration table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Key</th>
+ <th class="text-left" style="width: 15%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 55%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>pulsar.source.autoCommitCursorInterval</h5></td>
+ <td style="word-wrap: break-word;">5000</td>
+ <td>Long</td>
+ <td>This option is used only when the user disables the checkpoint and uses Exclusive or Failover subscription. We would automatically commit the cursor using the given period (in ms).</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.source.enableAutoAcknowledgeMessage</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Flink commits the consuming position with pulsar transactions on checkpoint. However, if you have disabled the Flink checkpoint or disabled transaction for your Pulsar cluster, ensure that you have set this option to <code class="highlighter-rouge">true</code>.<br />The source would use pulsar client's internal mechanism and commit cursor in two ways.<ul><li>For <code class="highlighter-rouge">Key_Shared</code> and <code class="highlighter-rouge">Shared</code> subscriptio [...]
+ </tr>
+ <tr>
+ <td><h5>pulsar.source.maxFetchRecords</h5></td>
+ <td style="word-wrap: break-word;">100</td>
+ <td>Integer</td>
+ <td>The maximum number of records to fetch to wait when polling. A longer time increases throughput but also latency. A fetch batch might be finished earlier because of <code class="highlighter-rouge">pulsar.source.maxFetchTime</code>.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.source.maxFetchTime</h5></td>
+ <td style="word-wrap: break-word;">10000</td>
+ <td>Long</td>
+ <td>The maximum time (in ms) to wait when fetching records. A longer time increases throughput but also latency. A fetch batch might be finished earlier because of <code class="highlighter-rouge">pulsar.source.maxFetchRecords</code>.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.source.partitionDiscoveryIntervalMs</h5></td>
+ <td style="word-wrap: break-word;">30000</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>
+ <tr>
+ <td><h5>pulsar.source.transactionTimeoutMillis</h5></td>
+ <td style="word-wrap: break-word;">10800000</td>
+ <td>Long</td>
+ <td>This option is used in <code class="highlighter-rouge">Shared</code> or <code class="highlighter-rouge">Key_Shared</code> subscription. You should configure this option when you do not enable the <code class="highlighter-rouge">pulsar.source.enableAutoAcknowledgeMessage</code> option.<br />The value (in ms) should be greater than the checkpoint interval.</td>
+ </tr>
+ <tr>
+ <td><h5>pulsar.source.verifyInitialOffsets</h5></td>
+ <td style="word-wrap: break-word;">WARN_ON_MISMATCH</td>
+ <td><p>Enum</p></td>
+ <td>Upon (re)starting the source, check whether the expected message can be read. If failure is enabled, the application fails. Otherwise, it logs a warning. A possible solution is to adjust the retention settings in Pulsar or ignoring the check result.<br /><br />Possible values:<ul><li>"FAIL_ON_MISMATCH"</li><li>"WARN_ON_MISMATCH"</li></ul></td>
+ </tr>
+ </tbody>
+</table>
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarOptions.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarOptions.java
index 4c33a70..db90bf1 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarOptions.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/config/PulsarOptions.java
@@ -31,7 +31,6 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
-import static java.util.Collections.emptyMap;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.flink.configuration.description.TextElement.code;
import static org.apache.flink.configuration.description.TextElement.text;
@@ -48,6 +47,7 @@ import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.CLIE
@ConfigGroup(name = "PulsarClient", keyPrefix = CLIENT_CONFIG_PREFIX),
@ConfigGroup(name = "PulsarAdmin", keyPrefix = ADMIN_CONFIG_PREFIX)
})
+@SuppressWarnings("java:S1192")
public final class PulsarOptions {
// Pulsar client API config prefix.
@@ -112,8 +112,7 @@ public final class PulsarOptions {
.noDefaultValue()
.withDescription(
Description.builder()
- .text(
- "String represents parameters for the authentication plugin.")
+ .text("Parameters for the authentication plugin.")
.linebreak()
.linebreak()
.text("Example:")
@@ -124,9 +123,8 @@ public final class PulsarOptions {
public static final ConfigOption<Map<String, String>> PULSAR_AUTH_PARAM_MAP =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "authParamMap")
.mapType()
- .defaultValue(emptyMap())
- .withDescription(
- "Map which represents parameters for the authentication plugin.");
+ .noDefaultValue()
+ .withDescription("Parameters for the authentication plugin.");
public static final ConfigOption<Integer> PULSAR_OPERATION_TIMEOUT_MS =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "operationTimeoutMs")
@@ -134,10 +132,11 @@ public final class PulsarOptions {
.defaultValue(30000)
.withDescription(
Description.builder()
- .text("Operation timeout (in millis).")
+ .text("Operation timeout (in ms).")
.text(
- "Producer-create, subscribe and unsubscribe operations will be retried until this interval,"
- + " after which the operation will be marked as failed.")
+ " Operations such as creating producers, subscribing or unsubscribing topics are retried during this interval.")
+ .text(
+ " If the operation is not completed during this interval, the operation will be marked as failed.")
.build());
public static final ConfigOption<Long> PULSAR_STATS_INTERVAL_SECONDS =
@@ -153,7 +152,7 @@ public final class PulsarOptions {
"Stats is activated with positive %s",
code("statsInterval")),
text(
- "Set %s to 1 second at least",
+ "Set %s to 1 second at least.",
code("statsIntervalSeconds")))
.build());
@@ -173,10 +172,10 @@ public final class PulsarOptions {
.text(
"The number of threads used for handling message listeners.")
.text(
- "The listener thread pool is shared across all the consumers and readers that are using a %s model to get messages.",
+ " The listener thread pool is shared across all the consumers and readers that are using a %s model to get messages.",
code("listener"))
.text(
- "For a given consumer, the listener will be always invoked from the same thread, to ensure ordering.")
+ " For a given consumer, the listener is always invoked from the same thread to ensure ordering.")
.build());
public static final ConfigOption<Integer> PULSAR_CONNECTIONS_PER_BROKER =
@@ -186,12 +185,12 @@ public final class PulsarOptions {
.withDescription(
Description.builder()
.text(
- "Sets the max number of connection that the client library will open to a single broker.")
+ "The maximum number of connections that the client library will open to a single broker.")
.linebreak()
.text(
- "By default, the connection pool will use a single connection for all the producers and consumers.")
+ " By default, the connection pool will use a single connection for all the producers and consumers.")
.text(
- "Increasing this parameter may improve throughput when using many producers over a high latency connection.")
+ " Increasing this parameter may improve throughput when using many producers over a high latency connection.")
.build());
public static final ConfigOption<Boolean> PULSAR_USE_TCP_NO_DELAY =
@@ -201,18 +200,18 @@ public final class PulsarOptions {
.withDescription(
Description.builder()
.text(
- "Whether to use TCP no-delay flag on the connection to disable Nagle algorithm.")
+ "Whether to use the TCP no-delay flag on the connection to disable Nagle algorithm.")
.linebreak()
.text(
- "No-delay features make sure packets are sent out on the network as soon as possible,")
- .text("and it's critical to achieve low latency publishes.")
+ "No-delay features ensures that packets are sent out on the network as soon as possible,")
+ .text(" and it is critical to achieve low latency publishes.")
.text(
- "On the other hand, sending out a huge number of small packets might limit the overall throughput,")
+ " On the other hand, sending out a huge number of small packets might limit the overall throughput.")
.text(
- "so if latency is not a concern, it's advisable to set the %s flag to false.",
- code("useTcpNoDelay"))
+ " Therefore, if latency is not a concern, it is recommended to set the %s flag to %s.",
+ code("useTcpNoDelay"), code("false"))
.linebreak()
- .text("Default value is true.")
+ .text("By default, it is set to %s.", code("true"))
.build());
public static final ConfigOption<String> PULSAR_TLS_TRUST_CERTS_FILE_PATH =
@@ -226,7 +225,7 @@ public final class PulsarOptions {
.booleanType()
.defaultValue(false)
.withDescription(
- "Whether the Pulsar client accepts untrusted TLS certificate from broker.");
+ "Whether the Pulsar client accepts untrusted TLS certificate from the broker.");
public static final ConfigOption<Boolean> PULSAR_TLS_HOSTNAME_VERIFICATION_ENABLE =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "tlsHostnameVerificationEnable")
@@ -236,11 +235,11 @@ public final class PulsarOptions {
Description.builder()
.text("Whether to enable TLS hostname verification.")
.text(
- "It allows to validate hostname verification when client connects to broker over tls.")
+ " It allows to validate hostname verification when a client connects to the broker over TLS.")
.text(
- "It validates incoming x509 certificate and matches provided hostname(CN/SAN) with expected broker's host name.")
+ " It validates incoming x509 certificate and matches provided hostname (CN/SAN) with the expected broker's host name.")
.text(
- "It follows RFC 2818, 3.1. Server Identity hostname verification.")
+ " It follows RFC 2818, 3.1. Server Identity hostname verification.")
.build());
public static final ConfigOption<Integer> PULSAR_CONCURRENT_LOOKUP_REQUEST =
@@ -250,10 +249,9 @@ public final class PulsarOptions {
.withDescription(
Description.builder()
.text(
- "The number of concurrent lookup requests allowed to send on each broker connection to prevent overload on broker.")
+ "The number of concurrent lookup requests allowed to send on each broker connection to prevent overload on the broker.")
.text(
- "It should be configured with higher value only in case of it requires"
- + " to produce/subscribe on thousands of topic using created %s",
+ " It should be configured with a higher value only in case of it requires to produce or subscribe on thousands of topic using a created %s",
code("PulsarClient"))
.build());
@@ -264,12 +262,17 @@ public final class PulsarOptions {
.withDescription(
Description.builder()
.text(
- "The maximum number of lookup requests allowed on each broker connection to prevent overload on broker.")
- .text("It should be bigger than maxConcurrentLookupRequests.")
+ "The maximum number of lookup requests allowed on each broker connection to prevent overload on the broker.")
+ .text(
+ " It should be greater than %s.",
+ code("maxConcurrentLookupRequests"))
.text(
- "Requests that inside maxConcurrentLookupRequests already send to broker,")
+ " Requests that inside %s are already sent to broker,",
+ code("maxConcurrentLookupRequests"))
.text(
- "and requests beyond maxConcurrentLookupRequests and under maxLookupRequests will wait in each client cnx.")
+ " and requests beyond %s and under %s will wait in each client cnx.",
+ code("maxConcurrentLookupRequests"),
+ code("maxLookupRequests"))
.build());
public static final ConfigOption<Integer> PULSAR_MAX_LOOKUP_REDIRECTS =
@@ -277,23 +280,26 @@ public final class PulsarOptions {
.intType()
.defaultValue(20)
.withDescription(
- "Set the maximum number of times a lookup-request to a broker will be redirected.");
+ "The maximum number of times a lookup-request redirections to a broker.");
public static final ConfigOption<Integer> PULSAR_MAX_NUMBER_OF_REJECTED_REQUEST_PER_CONNECTION =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "maxNumberOfRejectedRequestPerConnection")
.intType()
.defaultValue(50)
.withDescription(
- "The maximum number of rejected requests of a broker in a certain time"
- + " frame (30 seconds) after the current connection is closed and"
- + " the client creates a new connection to connect to a different broker.");
+ Description.builder()
+ .text(
+ "The maximum number of rejected requests of a broker in a certain period (30s) after the current connection is closed")
+ .text(
+ " and the client creates a new connection to connect to a different broker.")
+ .build());
public static final ConfigOption<Integer> PULSAR_KEEP_ALIVE_INTERVAL_SECONDS =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "keepAliveIntervalSeconds")
.intType()
.defaultValue(30)
.withDescription(
- "Seconds of keeping alive interval for each client broker connection.");
+ "Interval (in seconds) for keeping connection between the Pulsar client and broker alive.");
public static final ConfigOption<Integer> PULSAR_CONNECTION_TIMEOUT_MS =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "connectionTimeoutMs")
@@ -302,18 +308,18 @@ public final class PulsarOptions {
.withDescription(
Description.builder()
.text(
- "Duration (in millis) of waiting for a connection to a broker to be established.")
+ "Duration (in ms) of waiting for a connection to a broker to be established.")
.linebreak()
.text(
"If the duration passes without a response from a broker, the connection attempt is dropped.")
.build());
- // TODO This option would be exposed by Pulsar's ClientBuilder in next release.
+ // TODO This option would be exposed by Pulsar's ClientBuilder in the next Pulsar release.
public static final ConfigOption<Integer> PULSAR_REQUEST_TIMEOUT_MS =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "requestTimeoutMs")
.intType()
.defaultValue(60000)
- .withDescription("Maximum duration (in millis) for completing a request.");
+ .withDescription("Maximum duration (in ms) for completing a request.");
public static final ConfigOption<Long> PULSAR_INITIAL_BACKOFF_INTERVAL_NANOS =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "initialBackoffIntervalNanos")
@@ -325,7 +331,8 @@ public final class PulsarOptions {
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "maxBackoffIntervalNanos")
.longType()
.defaultValue(SECONDS.toNanos(60))
- .withDescription("Maximum duration (in nanoseconds) for a backoff interval.");
+ .withDescription(
+ "The maximum duration (in nanoseconds) for a backoff interval.");
public static final ConfigOption<Boolean> PULSAR_ENABLE_BUSY_WAIT =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "enableBusyWait")
@@ -338,9 +345,9 @@ public final class PulsarOptions {
.text(
"This option will enable spin-waiting on executors and IO threads in order to reduce latency during context switches.")
.text(
- "The spinning will consume 100% CPU even when the broker is not doing any work.")
+ " The spinning will consume 100% CPU even when the broker is not doing any work.")
.text(
- "It is recommended to reduce the number of IO threads and BK client threads to only have few CPU cores busy.")
+ " It is recommended to reduce the number of IO threads and BookKeeper client threads to only have fewer CPU cores busy.")
.build());
public static final ConfigOption<String> PULSAR_LISTENER_NAME =
@@ -350,8 +357,8 @@ public final class PulsarOptions {
.withDescription(
Description.builder()
.text(
- "Configure the listenerName that the broker will return the corresponding %s.",
- code("advertisedListener"))
+ "Configure the %s that the broker will return the corresponding %s.",
+ code("listenerName"), code("advertisedListener"))
.build());
public static final ConfigOption<Boolean> PULSAR_USE_KEY_STORE_TLS =
@@ -361,8 +368,10 @@ public final class PulsarOptions {
.withDescription(
Description.builder()
.text(
- "If Tls is enabled, whether use KeyStore type as tls configuration parameter.")
- .text("False means use default pem type configuration.")
+ "If TLS is enabled, whether use the KeyStore type as the TLS configuration parameter.")
+ .text(
+ " If it is set to %s, it means to use the default pem type configuration.",
+ code("false"))
.build());
public static final ConfigOption<String> PULSAR_SSL_PROVIDER =
@@ -374,7 +383,7 @@ public final class PulsarOptions {
.text(
"The name of the security provider used for SSL connections.")
.text(
- "Default value is the default security provider of the JVM.")
+ " The default value is the default security provider of the JVM.")
.build());
public static final ConfigOption<String> PULSAR_TLS_TRUST_STORE_TYPE =
@@ -405,11 +414,11 @@ public final class PulsarOptions {
Description.builder()
.text("A list of cipher suites.")
.text(
- "This is a named combination of authentication, encryption,"
- + " MAC and key exchange algorithm used to negotiate the security"
- + " settings for a network connection using TLS or SSL network protocol.")
+ " This is a named combination of authentication, encryption,")
.text(
- "By default all the available cipher suites are supported.")
+ " MAC and the key exchange algorithm used to negotiate the security settings for a network connection using the TLS or SSL network protocol.")
+ .text(
+ " By default all the available cipher suites are supported.")
.build());
public static final ConfigOption<List<String>> PULSAR_TLS_PROTOCOLS =
@@ -420,9 +429,10 @@ public final class PulsarOptions {
.withDescription(
Description.builder()
.text("The SSL protocol used to generate the SSLContext.")
- .text("Default setting is TLS, which is fine for most cases.")
.text(
- "Allowed values in recent JVMs are TLS, TLSv1.3, TLSv1.2 and TLSv1.1.")
+ " By default, it is set TLS, which is fine for most cases.")
+ .text(
+ " Allowed values in recent JVMs are TLS, TLSv1.3, TLSv1.2 and TLSv1.1.")
.build());
public static final ConfigOption<Long> PULSAR_MEMORY_LIMIT_BYTES =
@@ -432,11 +442,11 @@ public final class PulsarOptions {
.withDescription(
Description.builder()
.text(
- "Configure a limit on the amount of direct memory that will be allocated by this client instance. Its unit is byte.")
+ "The limit (in bytes) on the amount of direct memory that will be allocated by this client instance.")
.linebreak()
.text(
"Note: at this moment this is only limiting the memory for producers.")
- .text("Setting this to 0 will disable the limit.")
+ .text(" Setting this to %s will disable the limit.", code("0"))
.build());
public static final ConfigOption<String> PULSAR_PROXY_SERVICE_URL =
@@ -444,8 +454,11 @@ public final class PulsarOptions {
.stringType()
.noDefaultValue()
.withDescription(
- "Proxy-service url when client would like to connect to broker via proxy."
- + " Client can choose type of proxy-routing.");
+ Description.builder()
+ .text(
+ "Proxy-service URL when a client connects to the broker via the proxy.")
+ .text(" The client can choose the type of proxy-routing.")
+ .build());
public static final ConfigOption<ProxyProtocol> PULSAR_PROXY_PROTOCOL =
ConfigOptions.key(CLIENT_CONFIG_PREFIX + "proxyProtocol")
@@ -454,8 +467,8 @@ public final class PulsarOptions {
.withDescription(
Description.builder()
.text(
- "Protocol type to determine type of proxy routing when client connects to proxy using %s.",
- code(CLIENT_CONFIG_PREFIX + "proxyServiceUrl"))
+ "Protocol type to determine the type of proxy routing when a client connects to the proxy using %s.",
+ code("pulsar.client.proxyServiceUrl"))
.build());
public static final ConfigOption<Boolean> PULSAR_ENABLE_TRANSACTION =
@@ -463,7 +476,12 @@ public final class PulsarOptions {
.booleanType()
.defaultValue(false)
.withDescription(
- "If enable transaction, start the transactionCoordinatorClient with pulsar client.");
+ Description.builder()
+ .text(
+ "If transaction is enabled, start the %s with %s.",
+ code("transactionCoordinatorClient"),
+ code("PulsarClient"))
+ .build());
///////////////////////////////////////////////////////////////////////////////
//
@@ -479,7 +497,7 @@ public final class PulsarOptions {
.withDescription(
Description.builder()
.text(
- "Set the Pulsar service HTTP URL for the admin endpoint. eg. %s, or %s for TLS.",
+ "The Pulsar service HTTP URL for the admin endpoint. For example, %s, or %s for TLS.",
code("http://my-broker.example.com:8080"),
code("https://my-broker.example.com:8443"))
.build());
@@ -488,27 +506,26 @@ public final class PulsarOptions {
ConfigOptions.key(ADMIN_CONFIG_PREFIX + "connectTimeout")
.intType()
.defaultValue(60000)
- .withDescription(
- "This sets the connection time out (in millis) for the pulsar admin client.");
+ .withDescription("The connection time out (in ms) for the PulsarAdmin client.");
public static final ConfigOption<Integer> PULSAR_READ_TIMEOUT =
ConfigOptions.key(ADMIN_CONFIG_PREFIX + "readTimeout")
.intType()
.defaultValue(60000)
.withDescription(
- "This sets the server response read time out (in millis) for the pulsar admin client for any request.");
+ "The server response read timeout (in ms) for the PulsarAdmin client for any request.");
public static final ConfigOption<Integer> PULSAR_REQUEST_TIMEOUT =
ConfigOptions.key(ADMIN_CONFIG_PREFIX + "requestTimeout")
.intType()
.defaultValue(300000)
.withDescription(
- "This sets the server request time out (in millis) for the pulsar admin client for any request.");
+ "The server request timeout (in ms) for the PulsarAdmin client for any request.");
public static final ConfigOption<Integer> PULSAR_AUTO_CERT_REFRESH_TIME =
ConfigOptions.key(ADMIN_CONFIG_PREFIX + "autoCertRefreshTime")
.intType()
.defaultValue(300000)
.withDescription(
- "This sets auto cert refresh time (in millis) if Pulsar admin uses tls authentication.");
+ "The auto cert refresh time (in ms) if Pulsar admin supports TLS authentication.");
}
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java
index 2884f5e..c319915 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java
@@ -55,6 +55,7 @@ import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.SOURC
@ConfigGroup(name = "PulsarSource", keyPrefix = SOURCE_CONFIG_PREFIX),
@ConfigGroup(name = "PulsarConsumer", keyPrefix = CONSUMER_CONFIG_PREFIX)
})
+@SuppressWarnings("java:S1192")
public final class PulsarSourceOptions {
// Pulsar source connector config prefix.
@@ -78,8 +79,11 @@ public final class PulsarSourceOptions {
.longType()
.defaultValue(Duration.ofSeconds(30).toMillis())
.withDescription(
- "The interval in milliseconds for the Pulsar source to discover "
- + "the new partitions. A non-positive value disables the partition discovery.");
+ Description.builder()
+ .text(
+ "The interval (in ms) for the Pulsar source to discover the new partitions.")
+ .text(" A non-positive value disables the partition discovery.")
+ .build());
public static final ConfigOption<Boolean> PULSAR_ENABLE_AUTO_ACKNOWLEDGE_MESSAGE =
ConfigOptions.key(SOURCE_CONFIG_PREFIX + "enableAutoAcknowledgeMessage")
@@ -89,18 +93,21 @@ public final class PulsarSourceOptions {
Description.builder()
.text(
"Flink commits the consuming position with pulsar transactions on checkpoint.")
- .linebreak()
.text(
- "However, if you have disabled the flink checkpoint or your pulsar cluster disabled the transaction,"
- + " make sure you have set this option to %s.",
+ " However, if you have disabled the Flink checkpoint or disabled transaction for your Pulsar cluster,")
+ .text(
+ " ensure that you have set this option to %s.",
code("true"))
+ .linebreak()
.text(
"The source would use pulsar client's internal mechanism and commit cursor in two ways.")
.list(
text(
- "For Key_Shared and Shared subscription: the cursor would be committed once the message is consumed."),
+ "For %s and %s subscription, the cursor would be committed once the message is consumed.",
+ code("Key_Shared"), code("Shared")),
text(
- "For Exclusive and Failover subscription: the cursor would be committed in a fixed interval."))
+ "For %s and %s subscription, the cursor would be committed in a given interval.",
+ code("Exclusive"), code("Failover")))
.build());
public static final ConfigOption<Long> PULSAR_AUTO_COMMIT_CURSOR_INTERVAL =
@@ -110,9 +117,9 @@ public final class PulsarSourceOptions {
.withDescription(
Description.builder()
.text(
- "This option is used only when user disabled checkpoint and using Exclusive or Failover subscription.")
+ "This option is used only when the user disables the checkpoint and uses Exclusive or Failover subscription.")
.text(
- "We would automatically commit the cursor using the given period (in millis).")
+ " We would automatically commit the cursor using the given period (in ms).")
.build());
public static final ConfigOption<Long> PULSAR_TRANSACTION_TIMEOUT_MILLIS =
@@ -122,13 +129,14 @@ public final class PulsarSourceOptions {
.withDescription(
Description.builder()
.text(
- "This option is used for when using Shared or Key_Shared subscription."
- + " You should set this option when you didn't enable the %s option.",
+ "This option is used in %s or %s subscription.",
+ code("Shared"), code("Key_Shared"))
+ .text(
+ " You should configure this option when you do not enable the %s option.",
code("pulsar.source.enableAutoAcknowledgeMessage"))
.linebreak()
.text(
- "This value should be greater than the checkpoint interval.")
- .text("It uses milliseconds as the unit of time.")
+ "The value (in ms) should be greater than the checkpoint interval.")
.build());
public static final ConfigOption<Long> PULSAR_MAX_FETCH_TIME =
@@ -137,10 +145,10 @@ public final class PulsarSourceOptions {
.defaultValue(Duration.ofSeconds(10).toMillis())
.withDescription(
Description.builder()
+ .text("The maximum time (in ms) to wait when fetching records.")
+ .text(" A longer time increases throughput but also latency.")
.text(
- "The max time (in millis) to wait when fetching records. "
- + "A longer time increases throughput but also latency. "
- + "A fetch batch might be finished earlier because of %s.",
+ " A fetch batch might be finished earlier because of %s.",
code("pulsar.source.maxFetchRecords"))
.build());
@@ -151,9 +159,10 @@ public final class PulsarSourceOptions {
.withDescription(
Description.builder()
.text(
- "The max number of records to fetch to wait when polling. "
- + "A longer time increases throughput but also latency."
- + "A fetch batch might be finished earlier because of %s.",
+ "The maximum number of records to fetch to wait when polling.")
+ .text(" A longer time increases throughput but also latency.")
+ .text(
+ " A fetch batch might be finished earlier because of %s.",
code("pulsar.source.maxFetchTime"))
.build());
@@ -162,9 +171,14 @@ public final class PulsarSourceOptions {
.enumType(CursorVerification.class)
.defaultValue(CursorVerification.WARN_ON_MISMATCH)
.withDescription(
- "Upon (re)starting the source checks whether the expected message can be read. "
- + "If failure is enabled the application fails, else it logs a warning. "
- + "A possible solution is to adjust the retention settings in pulsar or ignoring the check result.");
+ Description.builder()
+ .text(
+ "Upon (re)starting the source, check whether the expected message can be read.")
+ .text(
+ " If failure is enabled, the application fails. Otherwise, it logs a warning.")
+ .text(
+ " A possible solution is to adjust the retention settings in Pulsar or ignoring the check result.")
+ .build());
///////////////////////////////////////////////////////////////////////////////
//
@@ -178,8 +192,11 @@ public final class PulsarSourceOptions {
.stringType()
.noDefaultValue()
.withDescription(
- "Specify the subscription name for this consumer."
- + " This argument is required when constructing the consumer.");
+ Description.builder()
+ .text("Specify the subscription name for this consumer.")
+ .text(
+ " This argument is required when constructing the consumer.")
+ .build());
public static final ConfigOption<SubscriptionType> PULSAR_SUBSCRIPTION_TYPE =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "subscriptionType")
@@ -238,16 +255,15 @@ public final class PulsarSourceOptions {
.withDescription(
Description.builder()
.text(
- "Group a consumer acknowledgment for a specified time (in microseconds).")
- .linebreak()
+ "Group a consumer acknowledgment for a specified time (in μs).")
.text(
- "By default, a consumer uses 100ms grouping time to send out acknowledgments to a broker.")
- .linebreak()
+ " By default, a consumer uses %s grouping time to send out acknowledgments to a broker.",
+ code("100μs"))
.text(
- "Setting a group time of 0 sends out acknowledgments immediately.")
- .linebreak()
+ " If the group time is set to %s, acknowledgments are sent out immediately.",
+ code("0"))
.text(
- "A longer ack group time is more efficient at the expense of a slight increase in message re-deliveries after a failure.")
+ " A longer ack group time is more efficient at the expense of a slight increase in message re-deliveries after a failure.")
.build());
public static final ConfigOption<Long> PULSAR_NEGATIVE_ACK_REDELIVERY_DELAY_MICROS =
@@ -257,11 +273,11 @@ public final class PulsarSourceOptions {
.withDescription(
Description.builder()
.text(
- "Delay (in microseconds) to wait before redelivering messages that failed to be processed.")
+ "Delay (in μs) to wait before redelivering messages that failed to be processed.")
.linebreak()
.text(
"When an application uses %s, failed messages are redelivered after a fixed timeout.",
- code("Consumer#negativeAcknowledge(Message)"))
+ code("Consumer.negativeAcknowledge(Message)"))
.build());
public static final ConfigOption<Integer>
@@ -274,7 +290,7 @@ public final class PulsarSourceOptions {
.withDescription(
Description.builder()
.text(
- "The max total receiver queue size across partitions.")
+ "The maximum total receiver queue size across partitions.")
.linebreak()
.text(
"This setting reduces the receiver queue size for individual partitions if the total receiver queue size exceeds this value.")
@@ -285,7 +301,7 @@ public final class PulsarSourceOptions {
.stringType()
.noDefaultValue()
.withDescription(
- "Consumer name is informative and it can be used to identify a particular consumer instance from the topic stats.");
+ "The consumer name is informative and it can be used to identify a particular consumer instance from the topic stats.");
public static final ConfigOption<Long> PULSAR_ACK_TIMEOUT_MILLIS =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "ackTimeoutMillis")
@@ -294,17 +310,13 @@ public final class PulsarSourceOptions {
.withDescription(
Description.builder()
.text(
- "Set the timeout (in millis) for unacknowledged messages, truncated to the nearest millisecond."
- + " The timeout needs to be greater than 1 second.")
+ "The timeout (in ms) for unacknowledged messages, truncated to the nearest millisecond. The timeout needs to be greater than 1 second.")
.linebreak()
.text(
- "By default, the acknowledge timeout is disabled and that means that messages delivered to a"
- + " consumer will not be re-delivered unless the consumer crashes.")
+ "By default, the acknowledge timeout is disabled and that means that messages delivered to a consumer will not be re-delivered unless the consumer crashes.")
.linebreak()
.text(
- "When enabling ack timeout, if a message is not acknowledged within the specified timeout"
- + " it will be re-delivered to the consumer"
- + " (possibly to a different consumer in case of a shared subscription).")
+ "When acknowledgement timeout being enabled, if a message is not acknowledged within the specified timeout it will be re-delivered to the consumer (possibly to a different consumer in case of a shared subscription).")
.build());
public static final ConfigOption<Long> PULSAR_TICK_DURATION_MILLIS =
@@ -313,10 +325,10 @@ public final class PulsarSourceOptions {
.defaultValue(1000L)
.withDescription(
Description.builder()
- .text("Granularity (in millis) of the ack-timeout redelivery.")
+ .text("Granularity (in ms) of the ack-timeout redelivery.")
.linebreak()
.text(
- "Using an higher %s reduces the memory overhead to track messages when setting ack-timeout to a bigger value (for example, 1 hour).",
+ "A greater (for example, 1 hour) %s reduces the memory overhead to track messages.",
code("tickDurationMillis"))
.build());
@@ -327,19 +339,21 @@ public final class PulsarSourceOptions {
.withDescription(
Description.builder()
.text(
- "Priority level for a consumer to which a broker gives more priority while dispatching messages in the shared subscription mode.")
+ "Priority level for a consumer to which a broker gives more priorities while dispatching messages in the shared subscription type.")
.linebreak()
.text(
"The broker follows descending priorities. For example, 0=max-priority, 1, 2,...")
.linebreak()
.text(
- "In shared subscription mode, the broker first dispatches messages to the max priority level consumers if they have permits. Otherwise, the broker considers next priority level consumers.")
+ "In shared subscription mode, the broker first dispatches messages to the consumers on the highest priority level if they have permits.")
+ .text(
+ " Otherwise, the broker considers consumers on the next priority level.")
.linebreak()
.linebreak()
.text("Example 1")
.linebreak()
.text(
- "If a subscription has consumerA with %s 0 and consumerB with %s 1, then the broker only dispatches messages to consumerA until it runs out permits and then starts dispatching messages to consumerB.",
+ "If a subscription has consumer A with %s 0 and consumer B with %s 1, then the broker only dispatches messages to consumer A until it runs out permits and then starts dispatching messages to consumer B.",
code("priorityLevel"), code("priorityLevel"))
.linebreak()
.text("Example 2")
@@ -353,7 +367,7 @@ public final class PulsarSourceOptions {
+ "C5, 1, 1\n")
.linebreak()
.text(
- "Order in which a broker dispatches messages to consumers is: C1, C2, C3, C1, C4, C5, C4.")
+ "The order in which a broker dispatches messages to consumers is: C1, C2, C3, C1, C4, C5, C4.")
.build());
public static final ConfigOption<Integer> PULSAR_MAX_PENDING_CHUNKED_MESSAGE =
@@ -363,32 +377,28 @@ public final class PulsarSourceOptions {
.withDescription(
Description.builder()
.text(
- "Consumer buffers chunk messages into memory until it receives all the chunks of the original message.")
+ "The consumer buffers chunk messages into memory until it receives all the chunks of the original message.")
.text(
- "While consuming chunk-messages, chunks from same message might not be contiguous"
- + " in the stream and they might be mixed with other messages' chunks.")
+ " While consuming chunk-messages, chunks from the same message might not be contiguous in the stream and they might be mixed with other messages' chunks.")
.text(
- "So, consumer has to maintain multiple buffers to manage chunks coming from different messages.")
+ " So, consumer has to maintain multiple buffers to manage chunks coming from different messages.")
.text(
- "This mainly happens when multiple publishers are publishing messages on the topic"
- + " concurrently or publisher failed to publish all chunks of the messages.")
+ " This mainly happens when multiple publishers are publishing messages on the topic concurrently or publishers failed to publish all chunks of the messages.")
.linebreak()
- .text("eg: M1-C1, M2-C1, M1-C2, M2-C2")
.text(
- "Messages M1-C1 and M1-C2 belong to original message M1, M2-C1 and M2-C2 messages belong to M2 message.")
+ "For example, there are M1-C1, M2-C1, M1-C2, M2-C2 messages.")
+ .text(
+ "Messages M1-C1 and M1-C2 belong to the M1 original message while M2-C1 and M2-C2 belong to the M2 message.")
.linebreak()
.text(
- "Buffering large number of outstanding uncompleted chunked messages can create memory"
- + " pressure and it can be guarded by providing this %s threshold."
- + " Once, consumer reaches this threshold, it drops the outstanding unchunked-messages"
- + " by silently acking or asking broker to redeliver later by marking it unacked."
- + " This behavior can be controlled by configuration %s",
- code(
- CONSUMER_CONFIG_PREFIX
- + "maxPendingChunkedMessage"),
+ "Buffering a large number of outstanding uncompleted chunked messages can bring memory pressure and it can be guarded by providing this %s threshold.",
+ code("pulsar.consumer.maxPendingChunkedMessage"))
+ .text(
+ " Once, a consumer reaches this threshold, it drops the outstanding unchunked messages by silently acknowledging or asking the broker to redeliver messages later by marking it unacknowledged.")
+ .text(
+ " This behavior can be controlled by the %s option.",
code(
- CONSUMER_CONFIG_PREFIX
- + "autoAckOldestChunkedMessageOnQueueFull"))
+ "pulsar.consumer.autoAckOldestChunkedMessageOnQueueFull"))
.build());
public static final ConfigOption<Boolean> PULSAR_AUTO_ACK_OLDEST_CHUNKED_MESSAGE_ON_QUEUE_FULL =
@@ -398,16 +408,12 @@ public final class PulsarSourceOptions {
.withDescription(
Description.builder()
.text(
- "Buffering large number of outstanding uncompleted chunked messages can create memory pressure"
- + " and it can be guarded by providing this %s threshold."
- + " Once, consumer reaches this threshold, it drops the outstanding unchunked-messages"
- + " by silently acking if %s is true else it marks them for redelivery.",
- code(
- CONSUMER_CONFIG_PREFIX
- + "maxPendingChunkedMessage"),
+ "Buffering a large number of outstanding uncompleted chunked messages can bring memory pressure and it can be guarded by providing this %s threshold.",
+ code("pulsar.consumer.maxPendingChunkedMessage"))
+ .text(
+ " Once a consumer reaches this threshold, it drops the outstanding unchunked-messages by silently acknowledging if %s is true. Otherwise, it marks them for redelivery.",
code(
- CONSUMER_CONFIG_PREFIX
- + "autoAckOldestChunkedMessageOnQueueFull"))
+ "pulsar.consumer.autoAckOldestChunkedMessageOnQueueFull"))
.build());
public static final ConfigOption<Long> PULSAR_EXPIRE_TIME_OF_INCOMPLETE_CHUNKED_MESSAGE_MILLIS =
@@ -415,10 +421,12 @@ public final class PulsarSourceOptions {
.longType()
.defaultValue(60 * 1000L)
.withDescription(
- "If producer fails to publish all the chunks of a message then consumer"
- + " can expire incomplete chunks if consumer won't be able to"
- + " receive all chunks in expire times (default 1 hour)."
- + " It uses milliseconds as the unit of time.");
+ Description.builder()
+ .text(
+ "If a producer fails to publish all the chunks of a message,")
+ .text(
+ " the consumer can expire incomplete chunks if the consumer cannot receive all chunks in expire times (default 1 hour, in ms).")
+ .build());
public static final ConfigOption<ConsumerCryptoFailureAction> PULSAR_CRYPTO_FAILURE_ACTION =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "cryptoFailureAction")
@@ -427,23 +435,28 @@ public final class PulsarSourceOptions {
.withDescription(
Description.builder()
.text(
- "Consumer should take action when it receives a message that can not be decrypted.")
+ "The consumer should take action when it receives a message that can not be decrypted.")
.list(
text(
- "FAIL: this is the default option to fail messages until crypto succeeds."),
+ "%s: this is the default option to fail messages until crypto succeeds.",
+ code("FAIL")),
text(
- "DISCARD: silently acknowledge and not deliver message to an application."),
+ "%s: silently acknowledge but do not deliver messages to an application.",
+ code("DISCARD")),
text(
- "CONSUME: deliver encrypted messages to applications. It is the application's responsibility to decrypt the message."))
+ "%s: deliver encrypted messages to applications. It is the application's responsibility to decrypt the message.",
+ code("CONSUME")))
.linebreak()
- .text("The decompression of message fails.")
+ .text("Fail to decompress the messages.")
.linebreak()
.text(
"If messages contain batch messages, a client is not be able to retrieve individual messages in batch.")
.linebreak()
.text(
- "Delivered encrypted message contains %s which contains encryption and compression information in it using which application can decrypt consumed message payload.",
+ "The delivered encrypted message contains %s which contains encryption and compression information in.",
code("EncryptionContext"))
+ .text(
+ " You can use an application to decrypt the consumed message payload.")
.build());
public static final ConfigOption<Map<String, String>> PULSAR_CONSUMER_PROPERTIES =
@@ -453,17 +466,15 @@ public final class PulsarSourceOptions {
.withDescription(
Description.builder()
.text("A name or value property of this consumer.")
- .linebreak()
.text(
- "%s is application defined metadata attached to a consumer.",
+ " %s is application defined metadata attached to a consumer.",
code("properties"))
- .linebreak()
.text(
- "When getting a topic stats, associate this metadata with the consumer stats for easier identification.")
+ " When getting a topic stats, associate this metadata with the consumer stats for easier identification.")
.build());
public static final ConfigOption<Boolean> PULSAR_READ_COMPACTED =
- ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "readCompacted") // NOSONAR
+ ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "readCompacted")
.booleanType()
.defaultValue(false)
.withDescription(
@@ -524,47 +535,31 @@ public final class PulsarSourceOptions {
.intType()
.defaultValue(0)
.withDescription(
- "Maximum number of times that a message will be redelivered before being sent to the dead letter queue.");
+ "The maximum number of times that a message are redelivered before being sent to the dead letter queue.");
public static final ConfigOption<String> PULSAR_RETRY_LETTER_TOPIC =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "deadLetterPolicy.retryLetterTopic")
.stringType()
.noDefaultValue()
- .withDescription(
- "Name of the retry topic where the failing messages will be sent.");
+ .withDescription("Name of the retry topic where the failed messages are sent.");
public static final ConfigOption<String> PULSAR_DEAD_LETTER_TOPIC =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "deadLetterPolicy.deadLetterTopic")
.stringType()
.noDefaultValue()
- .withDescription(
- "Name of the dead topic where the failing messages will be sent.");
+ .withDescription("Name of the dead topic where the failed messages are sent.");
public static final ConfigOption<Boolean> PULSAR_RETRY_ENABLE =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "retryEnable")
.booleanType()
.defaultValue(false)
- .withDescription("If enabled, the consumer will auto retry messages.");
-
- public static final ConfigOption<Boolean> PULSAR_AUTO_UPDATE_PARTITIONS =
- ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "autoUpdatePartitions")
- .booleanType()
- .defaultValue(true)
- .withDescription(
- Description.builder()
- .text(
- "If %s is enabled, a consumer subscribes to partition increase automatically.",
- code("autoUpdatePartitions"))
- .linebreak()
- .text("Note: this is only for partitioned consumers.\t")
- .build());
+ .withDescription("If enabled, the consumer will automatically retry messages.");
public static final ConfigOption<Integer> PULSAR_AUTO_UPDATE_PARTITIONS_INTERVAL_SECONDS =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "autoUpdatePartitionsIntervalSeconds")
.intType()
.defaultValue(60)
.withDescription(
- "Set the interval (in seconds) of updating partitions."
- + " This only works if autoUpdatePartitions is enabled.");
+ "The interval (in seconds) of updating partitions. This only works if autoUpdatePartitions is enabled.");
public static final ConfigOption<Boolean> PULSAR_REPLICATE_SUBSCRIPTION_STATE =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "replicateSubscriptionState")
@@ -582,23 +577,11 @@ public final class PulsarSourceOptions {
.booleanType()
.defaultValue(false)
.withDescription(
- "Ack will return receipt but does not mean that the message will not be resent after get receipt.");
+ "Acknowledgement will return a receipt but this does not mean that the message will not be resent after getting the receipt.");
public static final ConfigOption<Boolean> PULSAR_POOL_MESSAGES =
ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "poolMessages")
.booleanType()
.defaultValue(false)
- .withDescription(
- Description.builder()
- .text(
- "Enable pooling of messages and the underlying data buffers.")
- .linebreak()
- .text(
- "When pooling is enabled, the application is responsible for calling"
- + " %s after the handling of every received message. If %s"
- + " is not called on a received message, there will be a memory leak."
- + " If an application attempts to use and already \"released\" message,"
- + " it might experience undefined behavior (eg: memory corruption, deserialization error, etc.).",
- code("Message.release()"), code("release()"))
- .build());
+ .withDescription("Enable pooling of messages and the underlying data buffers.");
}
diff --git a/flink-docs/pom.xml b/flink-docs/pom.xml
index 68364e0..3791708 100644
--- a/flink-docs/pom.xml
+++ b/flink-docs/pom.xml
@@ -121,6 +121,12 @@ under the License.
</dependency>
<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-pulsar_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-slf4j-impl</artifactId>
<scope>compile</scope>
diff --git a/flink-docs/src/main/java/org/apache/flink/docs/configuration/ConfigOptionsDocGenerator.java b/flink-docs/src/main/java/org/apache/flink/docs/configuration/ConfigOptionsDocGenerator.java
index 8f04aea..9272789 100644
--- a/flink-docs/src/main/java/org/apache/flink/docs/configuration/ConfigOptionsDocGenerator.java
+++ b/flink-docs/src/main/java/org/apache/flink/docs/configuration/ConfigOptionsDocGenerator.java
@@ -95,7 +95,13 @@ public class ConfigOptionsDocGenerator {
"flink-kubernetes", "org.apache.flink.kubernetes.configuration"),
new OptionsClassLocation("flink-clients", "org.apache.flink.client.cli"),
new OptionsClassLocation(
- "flink-table/flink-sql-client", "org.apache.flink.table.client.config")
+ "flink-table/flink-sql-client", "org.apache.flink.table.client.config"),
+ new OptionsClassLocation(
+ "flink-connectors/flink-connector-pulsar",
+ "org.apache.flink.connector.pulsar.common.config"),
+ new OptionsClassLocation(
+ "flink-connectors/flink-connector-pulsar",
+ "org.apache.flink.connector.pulsar.source")
};
static final Set<String> EXCLUSIONS =
[flink] 03/04: [FLINK-23864][connector/pulsar] Remove
PULSAR_AUTO_UPDATE_PARTITIONS option.
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 982db57dffc9ec6e6960916638b25368d86a6cda
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Fri Sep 10 12:10:48 2021 +0800
[FLINK-23864][connector/pulsar] Remove PULSAR_AUTO_UPDATE_PARTITIONS option.
---
.../flink/connector/pulsar/source/config/PulsarSourceConfigUtils.java | 2 --
1 file changed, 2 deletions(-)
diff --git a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/PulsarSourceConfigUtils.java b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/PulsarSourceConfigUtils.java
index b212aa7..7ac480a 100644
--- a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/PulsarSourceConfigUtils.java
+++ b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/PulsarSourceConfigUtils.java
@@ -48,7 +48,6 @@ import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSA
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_ACK_RECEIPT_ENABLED;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_ACK_TIMEOUT_MILLIS;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_AUTO_ACK_OLDEST_CHUNKED_MESSAGE_ON_QUEUE_FULL;
-import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_AUTO_UPDATE_PARTITIONS;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_AUTO_UPDATE_PARTITIONS_INTERVAL_SECONDS;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_CONSUMER_NAME;
import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_CONSUMER_PROPERTIES;
@@ -160,7 +159,6 @@ public final class PulsarSourceConfigUtils {
PULSAR_SUBSCRIPTION_INITIAL_POSITION,
builder::subscriptionInitialPosition);
createDeadLetterPolicy(configuration).ifPresent(builder::deadLetterPolicy);
- setOptionValue(configuration, PULSAR_AUTO_UPDATE_PARTITIONS, builder::autoUpdatePartitions);
setOptionValue(
configuration,
PULSAR_AUTO_UPDATE_PARTITIONS_INTERVAL_SECONDS,