This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 498502d4db [Docs][Connector-V2] Improve RabbitMQ RocketMQ and MQTT
docs (#11344)
498502d4db is described below
commit 498502d4db13ff6dd19df28dc4d79871e10ea0a3
Author: Daniel Carter <[email protected]>
AuthorDate: Sun Jul 5 11:31:32 2026 +0800
[Docs][Connector-V2] Improve RabbitMQ RocketMQ and MQTT docs (#11344)
Co-authored-by: DanielCarter-stack
<[email protected]>
---
docs/en/connectors/sink/Mqtt.md | 17 +++++++++++++++
docs/en/connectors/sink/Rabbitmq.md | 18 ++++++++--------
docs/en/connectors/source/Mqtt.md | 15 ++++++++++++++
docs/en/connectors/source/Rabbitmq.md | 18 ++++++++++------
docs/en/connectors/source/RocketMQ.md | 5 ++++-
docs/zh/connectors/sink/Mqtt.md | 17 +++++++++++++++
docs/zh/connectors/sink/Rabbitmq.md | 26 +++++++++++------------
docs/zh/connectors/source/Mqtt.md | 39 ++++++++++++++++++++++++-----------
docs/zh/connectors/source/Rabbitmq.md | 38 ++++++++++++++++++++--------------
docs/zh/connectors/source/RocketMQ.md | 5 ++++-
10 files changed, 140 insertions(+), 58 deletions(-)
diff --git a/docs/en/connectors/sink/Mqtt.md b/docs/en/connectors/sink/Mqtt.md
index e00b0e31ce..fa13cd3e8e 100644
--- a/docs/en/connectors/sink/Mqtt.md
+++ b/docs/en/connectors/sink/Mqtt.md
@@ -166,6 +166,23 @@ message because `format` is set to `json`.
### Authenticated broker with text format
```hocon
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ FakeSource {
+ row.num = 10
+ schema = {
+ fields {
+ id = bigint
+ content = string
+ }
+ }
+ }
+}
+
sink {
MQTT {
url = "tcp://secure-broker.example.com:1883"
diff --git a/docs/en/connectors/sink/Rabbitmq.md
b/docs/en/connectors/sink/Rabbitmq.md
index 1468a35488..c28349e864 100644
--- a/docs/en/connectors/sink/Rabbitmq.md
+++ b/docs/en/connectors/sink/Rabbitmq.md
@@ -1,12 +1,12 @@
import ChangeLog from '../changelog/connector-rabbitmq.md';
-# Rabbitmq
+# RabbitMQ
-> Rabbitmq sink connector
+> RabbitMQ sink connector
## Description
-Used to write data to Rabbitmq.
+Used to write data to RabbitMQ queues.
## Key features
@@ -123,7 +123,7 @@ Sink plugin common parameters, please refer to [Sink Common
Options](../common-o
## Example
-simple:
+### Write Messages to a Queue
```hocon
env {
@@ -159,9 +159,9 @@ sink {
}
```
-### Example 2
+### Declare Queue Options
-queue with durable, exclusive, auto_delete:
+Queue with `durable`, `exclusive`, and `auto_delete`:
```hocon
env {
@@ -189,9 +189,9 @@ sink {
username = "guest"
password = "guest"
queue_name = "test1"
- durable = "true"
- exclusive = "false"
- auto_delete = "false"
+ durable = true
+ exclusive = false
+ auto_delete = false
rabbitmq.config = {
requested-heartbeat = 10
connection-timeout = 10
diff --git a/docs/en/connectors/source/Mqtt.md
b/docs/en/connectors/source/Mqtt.md
index d810978b74..0a1faf9129 100644
--- a/docs/en/connectors/source/Mqtt.md
+++ b/docs/en/connectors/source/Mqtt.md
@@ -10,6 +10,12 @@ Used to read messages from an MQTT broker. Supports MQTT
3.1.1 protocol via the
This connector subscribes to a configured MQTT topic, deserializes message
payloads as JSON or text, and converts them into SeaTunnel rows.
+## Supported Engines
+
+> SeaTunnel Zeta<br/>
+> Flink<br/>
+> Spark<br/>
+
## Key features
- [ ] [batch](../../introduction/concepts/connector-v2-features.md)
@@ -167,6 +173,11 @@ sink {
### Persistent session source
```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+}
+
source {
MQTT {
url = "tcp://broker.example.com:1883"
@@ -183,6 +194,10 @@ source {
}
}
}
+
+sink {
+ Console {}
+}
```
## Changelog
diff --git a/docs/en/connectors/source/Rabbitmq.md
b/docs/en/connectors/source/Rabbitmq.md
index 9d6ed7020b..07fba40b87 100644
--- a/docs/en/connectors/source/Rabbitmq.md
+++ b/docs/en/connectors/source/Rabbitmq.md
@@ -1,12 +1,12 @@
import ChangeLog from '../changelog/connector-rabbitmq.md';
-# Rabbitmq
+# RabbitMQ
-> Rabbitmq source connector
+> RabbitMQ source connector
## Description
-Used to read data from Rabbitmq.
+Used to read data from RabbitMQ queues.
## Key features
@@ -16,6 +16,7 @@ Used to read data from Rabbitmq.
- [ ] [column projection](../../introduction/concepts/connector-v2-features.md)
- [ ] [parallelism](../../introduction/concepts/connector-v2-features.md)
- [ ] [support user-defined
split](../../introduction/concepts/connector-v2-features.md)
+- [x] [support multiple table
read](../../introduction/concepts/connector-v2-features.md)
:::tip
@@ -99,7 +100,7 @@ the schema fields of upstream data. For more details, please
refer to [Schema Fe
### tables_configs [array]
-Used to read from multiple queues simultaneously. Each object in the array
must contain a queue_name and a schema.
+Used to read from multiple queues simultaneously. Each object in the array
must contain `queue_name` and `schema`.
### network_recovery_interval [int]
@@ -169,9 +170,10 @@ Source plugin common parameters, please refer to [Source
Common Options](../comm
If you are upgrading from a previous version that only supported single-table
reads, your existing configuration will work without any changes.
**Configuration Priority:**
-- You cannot configure both `tables_configs` and the root-level
`queue_name`/`schema` at the same time. They are mutually exclusive. Doing so
will result in a configuration validation error.
+- You cannot configure both `tables_configs` and the root-level `queue_name`
at the same time. They are mutually exclusive. Doing so will result in a
configuration validation error.
- Use `tables_configs` for multi-table mode.
-- Use root-level `queue_name` and `schema` for single-table mode.
+- Use root-level `queue_name` and `schema` for single-queue mode.
+- In multi-table mode, put each queue's `schema` inside its own
`tables_configs` item.
- If you configure `username`, you must also configure `password`, and vice
versa.
- `host` and `port` are always required. `virtual_host` is optional unless
your RabbitMQ deployment requires a non-default virtual host.
@@ -193,6 +195,9 @@ source {
username = "guest"
password = "guest"
queue_name = "test"
+ durable = true
+ exclusive = false
+ auto_delete = false
schema = {
fields {
id = bigint
@@ -209,6 +214,7 @@ sink {
Console {}
}
```
+
### Multi-table Read Example
You can use the `tables_configs` option to consume messages from multiple
RabbitMQ queues simultaneously within a single job. The connector will
automatically assign the correct table identifier to each row based on the
queue it originated from, allowing you to route them to different sinks using
`plugin_input`.
diff --git a/docs/en/connectors/source/RocketMQ.md
b/docs/en/connectors/source/RocketMQ.md
index 3dce2b4420..a7557d56c1 100644
--- a/docs/en/connectors/source/RocketMQ.md
+++ b/docs/en/connectors/source/RocketMQ.md
@@ -22,6 +22,7 @@ import ChangeLog from '../changelog/connector-rocketmq.md';
- [ ] [column projection](../../introduction/concepts/connector-v2-features.md)
- [x] [parallelism](../../introduction/concepts/connector-v2-features.md)
- [ ] [support user-defined
split](../../introduction/concepts/connector-v2-features.md)
+- [x] [support multiple table
read](../../introduction/concepts/connector-v2-features.md)
## Description
@@ -74,10 +75,12 @@ start.mode.offsets = {
### Multi-Table Read
-Use `tables_configs` when different topics have different schemas. Each item
can define its own `topics`, `schema`, `format`, `tags`, and startup position.
If `schema.table` is not set, the output table name defaults to the topic name.
+Use `tables_configs` when different topics have different schemas. Each item
must contain `topics` and can define its own `schema`, `format`, `tags`, and
startup position. If `schema.table` is not set, the output table name defaults
to the topic name.
`topics`, `tables_configs`, and the deprecated `table_list` are mutually
exclusive. In `tables_configs`, options that are not set on an item inherit the
top-level defaults, so each item only needs to override the topic-specific
schema, tags, or startup position.
+When a `tables_configs` item uses `start.mode = CONSUME_FROM_TIMESTAMP`, it
must also set `start.mode.timestamp`. When it uses `start.mode =
CONSUME_FROM_SPECIFIC_OFFSETS`, it must also set a non-empty
`start.mode.offsets` map.
+
## Task Examples
### Read JSON Messages
diff --git a/docs/zh/connectors/sink/Mqtt.md b/docs/zh/connectors/sink/Mqtt.md
index 1de1cadc4c..6c638a3cd0 100644
--- a/docs/zh/connectors/sink/Mqtt.md
+++ b/docs/zh/connectors/sink/Mqtt.md
@@ -161,6 +161,23 @@ sink {
### 使用认证并写入文本格式
```hocon
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ FakeSource {
+ row.num = 10
+ schema = {
+ fields {
+ id = bigint
+ content = string
+ }
+ }
+ }
+}
+
sink {
MQTT {
url = "tcp://secure-broker.example.com:1883"
diff --git a/docs/zh/connectors/sink/Rabbitmq.md
b/docs/zh/connectors/sink/Rabbitmq.md
index e621cb8468..93bdc889ca 100644
--- a/docs/zh/connectors/sink/Rabbitmq.md
+++ b/docs/zh/connectors/sink/Rabbitmq.md
@@ -1,16 +1,16 @@
import ChangeLog from '../changelog/connector-rabbitmq.md';
-# Rabbitmq
+# RabbitMQ
-> Rabbitmq 数据接收器
+> RabbitMQ Sink 连接器
## 描述
-该数据接收器是将数据写入Rabbitmq。
+用于将数据写入 RabbitMQ 队列。
## 主要特性
-- [ ] [精准一次](../../introduction/concepts/connector-v2-features.md)
+- [ ] [精确一次](../../introduction/concepts/connector-v2-features.md)
- [ ] [定时刷新](../../introduction/concepts/connector-v2-features.md)
## 接收器选项
@@ -38,23 +38,23 @@ import ChangeLog from '../changelog/connector-rabbitmq.md';
### host [string]
-Rabbitmq服务器地址
+RabbitMQ 服务器地址
### port [int]
-Rabbitmq服务器端口
+RabbitMQ 服务器端口
### virtual_host [string]
-virtual host – 连接broker使用的vhost
+virtual host，连接 broker 使用的 vhost
### username [string]
-连接broker时使用的用户名
+连接 broker 时使用的用户名
### password [string]
-连接broker时使用的密码
+连接 broker 时使用的密码
`username` 和 `password` 需要一起配置。
@@ -123,7 +123,7 @@ Sink插件常用参数，请参考[Sink常用选项](../common-options/sink-comm
## 示例
-simple:
+### 写入队列
```hocon
env {
@@ -189,9 +189,9 @@ sink {
username = "guest"
password = "guest"
queue_name = "test1"
- durable = "true"
- exclusive = "false"
- auto_delete = "false"
+ durable = true
+ exclusive = false
+ auto_delete = false
rabbitmq.config = {
requested-heartbeat = 10
connection-timeout = 10
diff --git a/docs/zh/connectors/source/Mqtt.md
b/docs/zh/connectors/source/Mqtt.md
index b057c8f3b4..f6624aa725 100644
--- a/docs/zh/connectors/source/Mqtt.md
+++ b/docs/zh/connectors/source/Mqtt.md
@@ -10,22 +10,28 @@ import ChangeLog from '../changelog/connector-mqtt.md';
该连接器会订阅配置的 MQTT topic，将消息 payload 按 JSON 或 text 格式反序列化，并转换为 SeaTunnel Row。
-## 关键特性
+## 支持引擎
-- [ ] [批](../../introduction/concepts/connector-v2-features.md)
-- [x] [流](../../introduction/concepts/connector-v2-features.md)
+> SeaTunnel Zeta<br/>
+> Flink<br/>
+> Spark<br/>
+
+## 主要特性
+
+- [ ] [批处理](../../introduction/concepts/connector-v2-features.md)
+- [x] [流处理](../../introduction/concepts/connector-v2-features.md)
- [ ] [精确一次](../../introduction/concepts/connector-v2-features.md)
- [ ] [列投影](../../introduction/concepts/connector-v2-features.md)
-- [ ] [并行性](../../introduction/concepts/connector-v2-features.md)
-- [ ] [支持用户自定义 split](../../introduction/concepts/connector-v2-features.md)
+- [ ] [并行度](../../introduction/concepts/connector-v2-features.md)
+- [ ] [支持用户自定义分片](../../introduction/concepts/connector-v2-features.md)
:::caution 交付语义
-`qos` 选项只控制 MQTT broker 和 MQTT client 之间的交付语义。它没有与 SeaTunnel checkpoint 集成，因此该
Source 不提供端到端的精确一次或至少一次保证。
+`qos` 选项只控制 MQTT broker 和 MQTT client 之间的交付语义。它没有与 SeaTunnel checkpoint
集成，因此该源连接器不提供端到端的精确一次或至少一次保证。
-如需使用 MQTT 持久会话，请设置 `clean_session=false` 并配置稳定的 `client_id`。当
`clean_session=false` 时，Source 在关闭时只断开连接，不会取消订阅，因此 broker 可以根据 MQTT 会话语义保留订阅。
+如需使用 MQTT 持久会话，请设置 `clean_session=false` 并配置稳定的 `client_id`。当
`clean_session=false` 时，源连接器在关闭时只断开连接，不会取消订阅，因此 broker 可以根据 MQTT 会话语义保留订阅。
-Source 使用 MQTT 自动重连。如果 client 断开连接的时间超过 `reconnect_timeout`，Source task
会失败，以避免静默停止摄取。
+源连接器使用 MQTT 自动重连。如果客户端断开连接的时间超过 `reconnect_timeout`，源任务会失败，以避免静默停止摄取。
:::
@@ -120,7 +126,7 @@ MQTT keep alive 间隔，单位为秒。
### reconnect_timeout [int]
-等待 MQTT 自动重连的最长时间，单位为秒。如果 MQTT client 断开连接的时间超过该超时时间，`pollNext()` 会使 Source
task 失败，避免无限期静默等待。
+等待 MQTT 自动重连的最长时间，单位为秒。如果 MQTT 客户端断开连接的时间超过该超时时间，`pollNext()`
会使源任务失败，避免无限期静默等待。
### max_queue_size [int]
@@ -128,11 +134,11 @@ MQTT keep alive 间隔，单位为秒。
### common options
-源插件通用参数，请参考 [源通用选项](../common-options/source-common-options.md) 详见。
+源插件通用参数，详情请参考 [源通用选项](../common-options/source-common-options.md)。
## 示例
-### JSON Source
+### JSON 源
```hocon
env {
@@ -164,9 +170,14 @@ sink {
}
```
-### 持久会话 Source
+### 持久会话源
```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+}
+
source {
MQTT {
url = "tcp://broker.example.com:1883"
@@ -183,6 +194,10 @@ source {
}
}
}
+
+sink {
+ Console {}
+}
```
## 变更日志
diff --git a/docs/zh/connectors/source/Rabbitmq.md
b/docs/zh/connectors/source/Rabbitmq.md
index c483853212..7edf6416aa 100644
--- a/docs/zh/connectors/source/Rabbitmq.md
+++ b/docs/zh/connectors/source/Rabbitmq.md
@@ -1,21 +1,22 @@
import ChangeLog from '../changelog/connector-rabbitmq.md';
-# Rabbitmq
+# RabbitMQ
-> Rabbitmq 源连接器
+> RabbitMQ 源连接器
## 描述
-用于从 Rabbitmq 读取数据。
+用于从 RabbitMQ 队列读取数据。
-## 关键特性
+## 主要特性
-- [ ] [批](../../introduction/concepts/connector-v2-features.md)
-- [x] [流](../../introduction/concepts/connector-v2-features.md)
+- [ ] [批处理](../../introduction/concepts/connector-v2-features.md)
+- [x] [流处理](../../introduction/concepts/connector-v2-features.md)
- [x] [精确一次](../../introduction/concepts/connector-v2-features.md)
- [ ] [列投影](../../introduction/concepts/connector-v2-features.md)
-- [ ] [并行性](../../introduction/concepts/connector-v2-features.md)
-- [ ] [支持用户自定义split](../../introduction/concepts/connector-v2-features.md)
+- [ ] [并行度](../../introduction/concepts/connector-v2-features.md)
+- [ ] [支持用户自定义分片](../../introduction/concepts/connector-v2-features.md)
+- [x] [支持多表读取](../../introduction/concepts/connector-v2-features.md)
:::tip
@@ -42,9 +43,9 @@ import ChangeLog from '../changelog/connector-rabbitmq.md';
| topology_recovery_enabled | boolean | 否 | - | 如果为 true，启用拓扑恢复
|
| AUTOMATIC_RECOVERY_ENABLED | boolean | 否 | - | 如果为 true，启用连接恢复
|
| connection_timeout | int | 否 | - | 连接 tcp 建立超时（毫秒）；零表示无限
|
-| requested_channel_max | int | 否 | - |
最初请求的最大通道数；零表示无限制。**注意：值必须在 0 到 65535 之间（AMQP 0-9-1 中的无符号短整数）。 |
+| requested_channel_max | int | 否 | - |
最初请求的最大通道数；零表示无限制。**注意：值必须在 0 到 65535 之间（AMQP 0-9-1 中的无符号短整数）。** |
| requested_frame_max | int | 否 | - | 请求的最大帧大小
|
-| requested_heartbeat | int | 否 | - | 设置请求的心跳超时。**注意：值必须在 0 到
65535 之间（AMQP 0-9-1 中的无符号短整数）。 |
+| requested_heartbeat | int | 否 | - | 设置请求的心跳超时。**注意：值必须在 0 到
65535 之间（AMQP 0-9-1 中的无符号短整数）。** |
| prefetch_count | int | 否 | - | 预取计数，无需确认即可接收的最大消息数
|
| delivery_timeout | int | 否 | - | 交付超时，等待下一条消息交付的最大时间（毫秒）
|
| use_correlation_id | boolean | 否 | - | 消息是否带有可用于去重的唯一
correlation id |
@@ -99,7 +100,7 @@ RabbitMQ 共享配置中的可选 exchange。普通队列消费不需要配置
### tables_configs [array]
-用于同时从多个队列读取消息。数组中的每个对象必须包含 queue_name 和 schema。
+用于同时从多个队列读取消息。数组中的每个对象必须包含 `queue_name` 和 `schema`。
### network_recovery_interval [int]
@@ -121,7 +122,7 @@ RabbitMQ 共享配置中的可选 exchange。普通队列消费不需要配置
### requested_channel_max [int]
-最初请求的最大通道数；零表示无限制。**注意：值必须在 0 到 65535 之间（AMQP 0-9-1 中的无符号短整数）。
+最初请求的最大通道数；零表示无限制。**注意：值必须在 0 到 65535 之间（AMQP 0-9-1 中的无符号短整数）。**
### requested_frame_max [int]
@@ -129,7 +130,7 @@ RabbitMQ 共享配置中的可选 exchange。普通队列消费不需要配置
### requested_heartbeat [int]
-设置请求的心跳超时。**注意：值必须在 0 到 65535 之间（AMQP 0-9-1 中的无符号短整数）。
+设置请求的心跳超时。**注意：值必须在 0 到 65535 之间（AMQP 0-9-1 中的无符号短整数）。**
### prefetch_count [int]
@@ -145,7 +146,7 @@ RabbitMQ 共享配置中的可选 exchange。普通队列消费不需要配置
### common options
-源插件通用参数，请参考 [源通用选项](../common-options/source-common-options.md) 详见。
+源插件通用参数，详情请参考 [源通用选项](../common-options/source-common-options.md)。
### durable
@@ -167,9 +168,10 @@ RabbitMQ 共享配置中的可选 exchange。普通队列消费不需要配置
如果您从仅支持单表读取的早期版本升级，您现有的配置无需任何更改即可正常工作。
**配置优先级：**
-- 您不能同时配置 `tables_configs` 和根级别的 `queue_name`/`schema`。它们是互斥的。这样做将导致配置验证错误。
+- 不能同时配置 `tables_configs` 和根级别的 `queue_name`。它们是互斥的，同时配置会导致校验失败。
- 使用 `tables_configs` 进行多表模式。
-- 使用根级别的 `queue_name` 和 `schema` 进行单表模式。
+- 使用根级别的 `queue_name` 和 `schema` 进行单队列模式。
+- 多表模式下，每个队列自己的 `schema` 应放在对应的 `tables_configs` 条目里。
- 如果配置了 `username`，也必须配置 `password`，反过来也一样。
- `host` 和 `port` 总是必填。`virtual_host` 是可选项，除非您的 RabbitMQ 环境要求使用非默认虚拟主机。
@@ -191,6 +193,9 @@ source {
username = "guest"
password = "guest"
queue_name = "test"
+ durable = true
+ exclusive = false
+ auto_delete = false
schema = {
fields {
id = bigint
@@ -207,6 +212,7 @@ sink {
Console {}
}
```
+
### 多表读取示例
您可以使用 `tables_configs` 选项在一个作业中同时从多个 RabbitMQ
队列消费消息。连接器将根据消息来源的队列自动为每行数据分配正确的表标识符，允许您使用 `plugin_input` 将它们路由到不同的 sink。
diff --git a/docs/zh/connectors/source/RocketMQ.md
b/docs/zh/connectors/source/RocketMQ.md
index b21f7565e8..f9cb29ea74 100644
--- a/docs/zh/connectors/source/RocketMQ.md
+++ b/docs/zh/connectors/source/RocketMQ.md
@@ -22,6 +22,7 @@ import ChangeLog from '../changelog/connector-rocketmq.md';
- [ ] [列裁剪](../../introduction/concepts/connector-v2-features.md)
- [x] [并行度](../../introduction/concepts/connector-v2-features.md)
- [ ] [支持用户自定义分片](../../introduction/concepts/connector-v2-features.md)
+- [x] [支持多表读取](../../introduction/concepts/connector-v2-features.md)
## 描述
@@ -74,10 +75,12 @@ start.mode.offsets = {
### 多表读取
-当不同 topic 的字段结构不一样时，使用 `tables_configs`。每一项都可以单独配置
`topics`、`schema`、`format`、`tags` 和启动消费位置。如果没有配置 `schema.table`，输出表名默认使用 topic
名称。
+当不同 topic 的字段结构不一样时，使用 `tables_configs`。每一项都必须包含 `topics`，并且可以单独配置
`schema`、`format`、`tags` 和启动消费位置。如果没有配置 `schema.table`，输出表名默认使用 topic 名称。
`topics`、`tables_configs` 和已废弃的 `table_list` 互斥，只能配置其中一个。在 `tables_configs`
中，单个条目未配置的参数会沿用顶层默认值，因此每个条目只需要覆盖该 topic 特有的 schema、tag 或启动位置。
+如果某个 `tables_configs` 条目使用 `start.mode = CONSUME_FROM_TIMESTAMP`，必须同时配置
`start.mode.timestamp`。如果使用 `start.mode =
CONSUME_FROM_SPECIFIC_OFFSETS`，必须同时配置非空的 `start.mode.offsets`。
+
## 任务示例
### 读取 JSON 消息
