This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12675-5008be82623ef599c24c541638904ffd461e5245 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit c033fb3190f9ace9a8a823a940abc6460287f757 Author: chovygo <[email protected]> AuthorDate: Thu Oct 8 13:23:39 2026 +0000 [Docs][Connector-V2] Fix Mqtt field_delimiter table rendering and align en/zh source docs (#12675) Co-authored-by: zhaowei <[email protected]> --- docs/en/connectors/sink/Mqtt.md | 2 +- docs/en/connectors/source/Mqtt.md | 78 +++++++++++++++++++++++++++++++++++---- docs/zh/connectors/sink/Mqtt.md | 2 +- docs/zh/connectors/source/Mqtt.md | 75 +++++++++++++++++++++++++++++++++++++ 4 files changed, 148 insertions(+), 9 deletions(-) diff --git a/docs/en/connectors/sink/Mqtt.md b/docs/en/connectors/sink/Mqtt.md index 7a2efd9667..52833ae0ba 100644 --- a/docs/en/connectors/sink/Mqtt.md +++ b/docs/en/connectors/sink/Mqtt.md @@ -41,7 +41,7 @@ from job restarts, rely on upstream replay and MQTT QoS rather than a stable sin | password | string | no | - | MQTT broker password. Leave unset for anonymous access. | | qos | int | no | 1 | MQTT Quality of Service level. `0` is at-most-once, `1` is at-least-once. | | format | string | no | json | Serialization format. `json` serializes each row as a JSON object; `text` serializes each row as delimited text. | -| field_delimiter | string | no | , | Field delimiter used when `format = "text"`, for example `,`, `|`, or `\t`. | +| field_delimiter | string | no | , | Field delimiter used when `format = "text"`, for example `,`, `\|`, or `\t`. | | batch_size | int | no | 1 | Number of messages to buffer before sending to the broker. The buffer is also flushed at each checkpoint. | | retry_timeout | int | no | 5000 | Maximum time in milliseconds to retry publishing on transient network failures before failing the task. | | connection_timeout | int | no | 30 | MQTT connection establishment timeout in seconds. | diff --git a/docs/en/connectors/source/Mqtt.md b/docs/en/connectors/source/Mqtt.md index e82e7edeb1..cdbc55ce0a 100644 --- a/docs/en/connectors/source/Mqtt.md +++ b/docs/en/connectors/source/Mqtt.md @@ -46,7 +46,7 @@ The source uses MQTT auto-reconnect. If the client remains disconnected longer t | password | string | no | - | MQTT broker authentication password. Leave unset for anonymous access. | | qos | int | no | 1 | MQTT subscribe QoS level. `0` is at-most-once, `1` is at-least-once. | | format | string | no | json | Message deserialization format: `json` or `text`. | -| field_delimiter | string | no | , | Field delimiter for `text` format. Only used when `format=text`. | +| field_delimiter | string | no | , | Field delimiter used when `format=text`, e.g. `,`, `\|`, `\t`. | | client_id | string | no | - | MQTT client id. Required when `clean_session=false`; generated automatically otherwise. | | clean_session | boolean | no | true | Whether to use a clean MQTT session. | | connection_timeout | int | no | 30 | MQTT connection timeout in seconds. | @@ -55,6 +55,12 @@ The source uses MQTT auto-reconnect. If the client remains disconnected longer t | max_queue_size | int | no | 1000 | Maximum number of MQTT messages buffered in memory before deserialization. | | common-options | | no | - | Source plugin common parameters. See [Source Common Options](../common-options/source-common-options.md) for details. | +:::tip + +This connector currently supports only a single topic subscription. For multiple topics, use multiple MQTT sources. MQTT 5.0 features (such as shared subscriptions, message properties) are not yet enabled. + +::: + ### url [string] The MQTT broker connection URL. Must include protocol, host, and port. @@ -97,12 +103,6 @@ The deserialization format for incoming messages. Supported values: - `json` — Deserialize each message as a JSON object (default) - `text` — Deserialize each message as delimited plain text (delimiter controlled by `field_delimiter`) -### field_delimiter [string] - -The field delimiter used when `format` is set to `text`. Default is `,`. - -Examples: `,`, `|`, `\t` - ### client_id [string] The MQTT client id. If omitted while `clean_session=true`, the connector generates a random client id. @@ -200,6 +200,70 @@ sink { } ``` +### TLS/SSL source + +```hocon +env { + parallelism = 1 + job.mode = "STREAMING" +} + +source { + MQTT { + url = "ssl://broker.example.com:8883" + topic = "factory/line-1/status" + username = "seatunnel" + password = "broker-token" + client_id = "seatunnel-mqtt-tls" + qos = 1 + connection_timeout = 30 + keep_alive_interval = 60 + reconnect_timeout = 180 + format = "json" + schema = { + fields { + device_id = string + status = string + ts = bigint + } + } + } +} + +sink { + Console {} +} +``` + +### Text delimited source + +```hocon +env { + parallelism = 1 + job.mode = "STREAMING" +} + +source { + MQTT { + url = "tcp://broker.example.com:1883" + topic = "factory/line-2/log" + format = "text" + field_delimiter = "|" + schema = { + fields { + device_id = string + level = string + message = string + } + } + } +} + +sink { + Console {} +} +``` + ## Changelog <ChangeLog /> diff --git a/docs/zh/connectors/sink/Mqtt.md b/docs/zh/connectors/sink/Mqtt.md index 49435b5da9..93efa64c22 100644 --- a/docs/zh/connectors/sink/Mqtt.md +++ b/docs/zh/connectors/sink/Mqtt.md @@ -40,7 +40,7 @@ import ChangeLog from '../changelog/connector-mqtt.md'; | password | string | 否 | - | MQTT broker 认证密码。匿名访问时可以不配置。 | | qos | int | 否 | 1 | 发布消息时使用的 MQTT QoS 等级。`0` 表示最多一次，`1` 表示至少一次。 | | format | string | 否 | json | 输出消息的序列化格式。`json` 把每一行序列化为 JSON 对象；`text` 按分隔符拼接为纯文本。 | -| field_delimiter | string | 否 | , | 当 `format = "text"` 时使用的字段分隔符，例如 `,`、`|`、`\t`。 | +| field_delimiter | string | 否 | , | 当 `format = "text"` 时使用的字段分隔符，例如 `,`、`\|`、`\t`。 | | batch_size | int | 否 | 1 | 发送到 broker 前缓存的消息数量；每个 checkpoint 和 writer 关闭时也会自动 flush。 | | retry_timeout | int | 否 | 5000 | 发布消息遇到临时网络故障时，最多重试多久，单位为毫秒。 | | connection_timeout | int | 否 | 30 | 建立 MQTT 连接的超时时间，单位为秒。 | diff --git a/docs/zh/connectors/source/Mqtt.md b/docs/zh/connectors/source/Mqtt.md index b3b129a9e6..0caf630f7f 100644 --- a/docs/zh/connectors/source/Mqtt.md +++ b/docs/zh/connectors/source/Mqtt.md @@ -61,6 +61,81 @@ import ChangeLog from '../changelog/connector-mqtt.md'; ::: +### url [string] + +MQTT broker 连接 URL。必须包含协议、主机和端口。 + +示例：`tcp://broker.example.com:1883` + +### topic [string] + +要订阅消息的 MQTT topic。 + +示例：`iot/sensors/temperature` + +### schema [config] + +上游数据的 schema 字段。更多详情请参考 [Schema 特性](../../introduction/concepts/schema-feature.md)。 + +### username [string] + +MQTT broker 认证用户名。匿名访问时可不填。 + +### password [string] + +MQTT broker 认证密码。匿名访问时可不填。 + +### qos [int] + +订阅 topic 时使用的 MQTT QoS 等级。 + +该设置只控制 MQTT broker 和 MQTT client 之间的交付，不在 SeaTunnel 中提供端到端交付保证。 + +支持的取值： + +- `0` — MQTT QoS 0 +- `1` — MQTT QoS 1 + +### format [string] + +输入消息的反序列化格式。支持的取值： + +- `json` — 将每条消息反序列化为 JSON 对象（默认） +- `text` — 将每条消息按分隔符切分为纯文本（分隔符由 `field_delimiter` 控制） + +### client_id [string] + +MQTT client id。当 `clean_session=true` 且未配置该选项时，连接器会生成随机 client id。 + +当 `clean_session=false` 时必须配置该选项，因为持久 MQTT 会话需要稳定的 client id。 + +### clean_session [boolean] + +是否使用 clean MQTT session。默认为 `true`。 + +- `true` — broker 丢弃之前的会话状态。适用于无状态场景。 +- `false` — broker 可以保留会话状态和订阅。需要稳定的 `client_id`。 + +### connection_timeout [int] + +MQTT 连接建立超时时间，单位为秒。 + +### keep_alive_interval [int] + +MQTT keep alive 间隔，单位为秒。 + +### reconnect_timeout [int] + +等待 MQTT 自动重连的最长时间，单位为秒。如果 MQTT 客户端断开连接的时间超过该超时时间，`pollNext()` 会让源任务失败，而不是无限期静默等待。 + +### max_queue_size [int] + +反序列化之前在内存中缓存的 MQTT 消息最大数量。 + +### common options + +源插件通用参数，详情请参考 [源通用选项](../common-options/source-common-options.md)。 + ## 示例 ### JSON 源
