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 dccaf82688 [Docs][Connector-V2] Improve TDengine RocketMQ and DataHub
docs (#11346)
dccaf82688 is described below
commit dccaf8268840869fb11cbe88d986f19c8db3ad6a
Author: Daniel Carter <[email protected]>
AuthorDate: Sun Jul 5 17:55:03 2026 +0800
[Docs][Connector-V2] Improve TDengine RocketMQ and DataHub docs (#11346)
Co-authored-by: DanielCarter-stack
<[email protected]>
Co-authored-by: David Zollo <[email protected]>
---
docs/en/connectors/sink/Datahub.md | 19 +++++++++----
docs/en/connectors/sink/RocketMQ.md | 7 +++++
docs/en/connectors/sink/TDengine.md | 5 ++++
docs/en/connectors/source/RocketMQ.md | 52 ++++++++++++++++++++++++++++++++++-
docs/en/connectors/source/TDengine.md | 5 ++++
docs/zh/connectors/sink/Datahub.md | 13 +++++++--
docs/zh/connectors/sink/RocketMQ.md | 5 ++++
docs/zh/connectors/sink/TDengine.md | 3 ++
docs/zh/connectors/source/RocketMQ.md | 46 ++++++++++++++++++++++++++++++-
docs/zh/connectors/source/TDengine.md | 2 ++
10 files changed, 147 insertions(+), 10 deletions(-)
diff --git a/docs/en/connectors/sink/Datahub.md
b/docs/en/connectors/sink/Datahub.md
index 88644b13cf..dccb5570a2 100644
--- a/docs/en/connectors/sink/Datahub.md
+++ b/docs/en/connectors/sink/Datahub.md
@@ -9,13 +9,21 @@ import ChangeLog from '../changelog/connector-datahub.md';
The DataHub sink writes SeaTunnel rows to Alibaba Cloud DataHub.
The connector supports single-table writes and multi-table writes. In
multi-table
-jobs, use placeholders such as `${table_name}` in `topic` to route records from
+jobs, use placeholders such as `${table}` in `topic` to route records from
different input tables to different DataHub topics.
## Key features
- [ ] [exactly-once](../../introduction/concepts/connector-v2-features.md)
+- [ ] [cdc](../../introduction/concepts/connector-v2-features.md)
- [x] [support multiple table
write](../../introduction/concepts/connector-v2-features.md)
+- [ ] [timer flush](../../introduction/concepts/connector-v2-features.md)
+
+## Before You Start
+
+Create the DataHub project and topic before running the SeaTunnel job. The
+DataHub topic schema must contain fields with the same names as the upstream
+SeaTunnel schema fields, because the sink writes values by field name.
## Options
@@ -49,10 +57,11 @@ The DataHub project name.
### topic [string]
The DataHub topic name. For multi-table writes, this value can contain
-placeholders, for example `${table_name}`.
+placeholders, for example `${table}`. `${table_name}` is only kept as a
+deprecated compatibility alias; use `${table}` for new jobs.
-The SeaTunnel field names must match the DataHub topic fields, because the
sink writes fields by
-name according to the topic schema.
+The SeaTunnel field names must match the DataHub topic fields, because the sink
+writes fields by name according to the topic schema.
### timeout [int]
@@ -147,7 +156,7 @@ sink {
accessId = "your-access-id"
accessKey = "your-access-key"
project = "demo_project"
- topic = "${table_name}"
+ topic = "${table}"
timeout = 3000
retryTimes = 3
}
diff --git a/docs/en/connectors/sink/RocketMQ.md
b/docs/en/connectors/sink/RocketMQ.md
index b11c69378a..588b04e876 100644
--- a/docs/en/connectors/sink/RocketMQ.md
+++ b/docs/en/connectors/sink/RocketMQ.md
@@ -17,6 +17,7 @@ import ChangeLog from '../changelog/connector-rocketmq.md';
## Key Features
- [x] [exactly-once](../../introduction/concepts/connector-v2-features.md)
+- [ ] [cdc](../../introduction/concepts/connector-v2-features.md)
- [ ] [timer flush](../../introduction/concepts/connector-v2-features.md)
## Description
@@ -61,6 +62,12 @@ The sink supports exactly-once writes through RocketMQ
transactional messages. T
When `format = text`, SeaTunnel serializes fields in the upstream schema order
and joins them with `field.delimiter`. When `format = json`, each row is
written as a JSON object.
+### producer.send.sync
+
+`producer.send.sync = true` makes the producer wait for RocketMQ to acknowledge
+each send request. With the default value `false`, messages are sent
+asynchronously.
+
## Task Examples
### Write JSON Messages
diff --git a/docs/en/connectors/sink/TDengine.md
b/docs/en/connectors/sink/TDengine.md
index 428d21ace7..e0bd099621 100644
--- a/docs/en/connectors/sink/TDengine.md
+++ b/docs/en/connectors/sink/TDengine.md
@@ -17,11 +17,16 @@ the target sub table name, the middle fields are normal
columns, and the last
fields are TAGS values. The connector reads the number of TAGS fields from the
target super table metadata.
+For example, if the target super table has two TAGS fields, the last two input
+fields are treated as TAGS values, and the first input field is treated as the
+sub table name.
+
## Key features
- [x] [exactly-once](../../introduction/concepts/connector-v2-features.md)
- [ ] [cdc](../../introduction/concepts/connector-v2-features.md)
- [x] [support multiple table
write](../../introduction/concepts/connector-v2-features.md)
+- [ ] [timer flush](../../introduction/concepts/connector-v2-features.md)
## Options
diff --git a/docs/en/connectors/source/RocketMQ.md
b/docs/en/connectors/source/RocketMQ.md
index a7557d56c1..9c44f68c0e 100644
--- a/docs/en/connectors/source/RocketMQ.md
+++ b/docs/en/connectors/source/RocketMQ.md
@@ -50,7 +50,7 @@ Reads messages from Apache RocketMQ topics. The connector can
read one or more t
| start.mode.offsets | Map | no | - | Required when `start.mode =
CONSUME_FROM_SPECIFIC_OFFSETS`. The key format is `topic-queueId`, for example
`test_topic-0`. |
| start.mode.timestamp | Long | no | - | Required when `start.mode =
CONSUME_FROM_TIMESTAMP`. Use a millisecond timestamp. |
| partition.discovery.interval.millis | long | no | -1 | Topic and partition
discovery interval in milliseconds. `-1` disables dynamic discovery. |
-| ignore_parse_errors | Boolean | no | false | Whether to skip messages that
cannot be parsed. |
+| ignore_parse_errors | Boolean | no | false | Whether to skip JSON messages
that cannot be parsed. |
| consumer.poll.timeout.millis | long | no | 5000 | Pull timeout in
milliseconds. |
| common-options | config | no | - | Source common options. See [Source Common
Options](../common-options/source-common-options.md). |
@@ -66,6 +66,10 @@ Reads messages from Apache RocketMQ topics. The connector
can read one or more t
- `CONSUME_FROM_TIMESTAMP`: start from the first offset at or after
`start.mode.timestamp`.
- `CONSUME_FROM_SPECIFIC_OFFSETS`: start from the offsets in
`start.mode.offsets`.
+When `start.mode = CONSUME_FROM_TIMESTAMP`, `start.mode.timestamp` must be a
+non-negative millisecond timestamp and cannot be later than the current time of
+the running job.
+
```hocon
start.mode = "CONSUME_FROM_SPECIFIC_OFFSETS"
start.mode.offsets = {
@@ -73,6 +77,21 @@ start.mode.offsets = {
}
```
+```hocon
+start.mode = "CONSUME_FROM_TIMESTAMP"
+start.mode.timestamp = 1667179890315
+```
+
+### Message Format
+
+When `format = json`, define `schema` so that SeaTunnel can parse the JSON
+message body into typed fields. `ignore_parse_errors = true` can be used to
skip
+invalid JSON messages instead of failing the job.
+
+When `format = text`, SeaTunnel splits the message body by `field.delimiter`
+and maps the values to fields in schema order. If `schema` is omitted, the
+message body is read as a single text value.
+
### Multi-Table Read
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.
@@ -226,6 +245,37 @@ sink {
}
```
+### Read From a Timestamp
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Rocketmq {
+ name.srv.addr = "rocketmq-e2e:9876"
+ topics = "test_topic_source"
+ plugin_output = "rocketmq_table"
+ format = json
+ start.mode = "CONSUME_FROM_TIMESTAMP"
+ start.mode.timestamp = 1667179890315
+ schema = {
+ fields {
+ id = bigint
+ }
+ }
+ }
+}
+
+sink {
+ Console {
+ plugin_input = "rocketmq_table"
+ }
+}
+```
+
## Changelog
<ChangeLog />
diff --git a/docs/en/connectors/source/TDengine.md
b/docs/en/connectors/source/TDengine.md
index 002bd28731..cccdad9848 100644
--- a/docs/en/connectors/source/TDengine.md
+++ b/docs/en/connectors/source/TDengine.md
@@ -92,6 +92,11 @@ When reading from a super table, put TAGS columns at the end
of the list. Do not
include `subtable_name`; the connector adds it automatically as the first
output
field.
+The order of `read_columns` decides the output field order after
+`subtable_name`. If the result is written to a TDengine sink, keep normal
columns
+before TAGS columns so the sink can split column values and TAGS values
+correctly.
+
## Examples
### Read all sub tables in a time range
diff --git a/docs/zh/connectors/sink/Datahub.md
b/docs/zh/connectors/sink/Datahub.md
index 0f558f49e3..a5a1b4f555 100644
--- a/docs/zh/connectors/sink/Datahub.md
+++ b/docs/zh/connectors/sink/Datahub.md
@@ -9,12 +9,18 @@ import ChangeLog from '../changelog/connector-datahub.md';
DataHub Sink 用于将 SeaTunnel 数据写入阿里云 DataHub。
该连接器支持单表写入和多表写入。多表写入时，可以在 `topic` 中使用
-`${table_name}` 这类占位符，将不同输入表的数据写入不同的 DataHub Topic。
+`${table}` 这类占位符，将不同输入表的数据写入不同的 DataHub Topic。
## 主要特性
- [ ] [精确一次](../../introduction/concepts/connector-v2-features.md)
+- [ ] [变更数据捕获](../../introduction/concepts/connector-v2-features.md)
- [x] [支持多表写入](../../introduction/concepts/connector-v2-features.md)
+- [ ] [定时刷新](../../introduction/concepts/connector-v2-features.md)
+
+## 使用前准备
+
+运行 SeaTunnel 任务前，请先创建 DataHub 项目和 Topic。DataHub Topic 的结构中需要包含和上游 SeaTunnel
schema 同名的字段，因为该 Sink 会按字段名写入数据。
## 选项
@@ -47,7 +53,8 @@ DataHub 项目名称。
### topic [string]
-DataHub Topic 名称。多表写入时可以使用占位符，例如 `${table_name}`。
+DataHub Topic 名称。多表写入时可以使用占位符，例如 `${table}`。
+`${table_name}` 仅作为已废弃的兼容别名保留，新任务建议使用 `${table}`。
SeaTunnel 字段名需要和 DataHub Topic 中的字段名一致，因为 sink 会按照 Topic 结构里的字段名写入数据。
@@ -142,7 +149,7 @@ sink {
accessId = "your-access-id"
accessKey = "your-access-key"
project = "demo_project"
- topic = "${table_name}"
+ topic = "${table}"
timeout = 3000
retryTimes = 3
}
diff --git a/docs/zh/connectors/sink/RocketMQ.md
b/docs/zh/connectors/sink/RocketMQ.md
index 1e1abe912c..d9a2f41767 100644
--- a/docs/zh/connectors/sink/RocketMQ.md
+++ b/docs/zh/connectors/sink/RocketMQ.md
@@ -17,6 +17,7 @@ import ChangeLog from '../changelog/connector-rocketmq.md';
## 主要特性
- [x] [精确一次](../../introduction/concepts/connector-v2-features.md)
+- [ ] [变更数据捕获](../../introduction/concepts/connector-v2-features.md)
- [ ] [定时刷新](../../introduction/concepts/connector-v2-features.md)
## 描述
@@ -61,6 +62,10 @@ Sink 支持通过 RocketMQ 事务消息实现精确一次写入。该能力默
当 `format = text` 时，SeaTunnel 会按上游 schema 的字段顺序序列化，并使用 `field.delimiter`
拼接字段。当 `format = json` 时，每行数据会写成一个 JSON 对象。
+### producer.send.sync
+
+`producer.send.sync = true` 表示生产者会等待 RocketMQ 确认每次发送请求。默认值为 `false` 时，消息会异步发送。
+
## 任务示例
### 写入 JSON 消息
diff --git a/docs/zh/connectors/sink/TDengine.md
b/docs/zh/connectors/sink/TDengine.md
index e9083a58aa..b87bd37799 100644
--- a/docs/zh/connectors/sink/TDengine.md
+++ b/docs/zh/connectors/sink/TDengine.md
@@ -12,11 +12,14 @@ import ChangeLog from '../changelog/connector-tdengine.md';
输入数据需要符合 TDengine 超级表写入结构：第一列是目标子表名，中间是普通列，最后几列是 TAGS 值。连接器会读取目标超级表元数据来判断末尾有几列
TAGS。
+例如，目标超级表有 2 个 TAGS 字段时，输入数据最后 2 列会作为 TAGS 值，第一列会作为子表名。
+
## 主要特性
- [x] [精确一次](../../introduction/concepts/connector-v2-features.md)
- [ ] [变更数据捕获](../../introduction/concepts/connector-v2-features.md)
- [x] [支持多表写入](../../introduction/concepts/connector-v2-features.md)
+- [ ] [定时刷新](../../introduction/concepts/connector-v2-features.md)
## 选项
diff --git a/docs/zh/connectors/source/RocketMQ.md
b/docs/zh/connectors/source/RocketMQ.md
index f9cb29ea74..1abca1cc75 100644
--- a/docs/zh/connectors/source/RocketMQ.md
+++ b/docs/zh/connectors/source/RocketMQ.md
@@ -50,7 +50,7 @@ import ChangeLog from '../changelog/connector-rocketmq.md';
| start.mode.offsets | Map | 否 | - | `start.mode =
CONSUME_FROM_SPECIFIC_OFFSETS` 时必填。key 格式为 `topic-queueId`，例如 `test_topic-0`。 |
| start.mode.timestamp | Long | 否 | - | `start.mode = CONSUME_FROM_TIMESTAMP`
时必填，单位是毫秒时间戳。 |
| partition.discovery.interval.millis | long | 否 | -1 | 动态发现 topic
和分区的间隔，单位毫秒。`-1` 表示不启用动态发现。 |
-| ignore_parse_errors | Boolean | 否 | false | 是否跳过解析失败的消息。 |
+| ignore_parse_errors | Boolean | 否 | false | 是否跳过解析失败的 JSON 消息。 |
| consumer.poll.timeout.millis | long | 否 | 5000 | 拉取消息的超时时间，单位毫秒。 |
| common-options | config | 否 | - | 源连接器通用参数，详情请参考
[源通用参数](../common-options/source-common-options.md)。 |
@@ -66,6 +66,8 @@ import ChangeLog from '../changelog/connector-rocketmq.md';
- `CONSUME_FROM_TIMESTAMP`：从 `start.mode.timestamp` 对应时间之后的第一条消息开始读。
- `CONSUME_FROM_SPECIFIC_OFFSETS`：从 `start.mode.offsets` 指定的位点开始读。
+当 `start.mode = CONSUME_FROM_TIMESTAMP` 时，`start.mode.timestamp`
必须是非负的毫秒时间戳，并且不能晚于任务运行时的当前时间。
+
```hocon
start.mode = "CONSUME_FROM_SPECIFIC_OFFSETS"
start.mode.offsets = {
@@ -73,6 +75,17 @@ start.mode.offsets = {
}
```
+```hocon
+start.mode = "CONSUME_FROM_TIMESTAMP"
+start.mode.timestamp = 1667179890315
+```
+
+### 消息格式
+
+当 `format = json` 时，请配置 `schema`，SeaTunnel 会按 schema 将 JSON 消息体解析成有类型的字段。配置
`ignore_parse_errors = true` 后，遇到无法解析的 JSON 消息会跳过，而不是让任务失败。
+
+当 `format = text` 时，SeaTunnel 会按 `field.delimiter` 拆分消息体，并按照 schema
字段顺序映射数据。如果不配置 `schema`，消息体会作为单个文本值读取。
+
### 多表读取
当不同 topic 的字段结构不一样时，使用 `tables_configs`。每一项都必须包含 `topics`，并且可以单独配置
`schema`、`format`、`tags` 和启动消费位置。如果没有配置 `schema.table`，输出表名默认使用 topic 名称。
@@ -226,6 +239,37 @@ sink {
}
```
+### 从指定时间戳读取
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Rocketmq {
+ name.srv.addr = "rocketmq-e2e:9876"
+ topics = "test_topic_source"
+ plugin_output = "rocketmq_table"
+ format = json
+ start.mode = "CONSUME_FROM_TIMESTAMP"
+ start.mode.timestamp = 1667179890315
+ schema = {
+ fields {
+ id = bigint
+ }
+ }
+ }
+}
+
+sink {
+ Console {
+ plugin_input = "rocketmq_table"
+ }
+}
+```
+
## 变更日志
<ChangeLog />
diff --git a/docs/zh/connectors/source/TDengine.md
b/docs/zh/connectors/source/TDengine.md
index 9afbe14d18..fe460860a7 100644
--- a/docs/zh/connectors/source/TDengine.md
+++ b/docs/zh/connectors/source/TDengine.md
@@ -78,6 +78,8 @@ TDengine 子表名称列表。不配置时读取该超级表下的所有子表
要读取的列名列表。不配置时读取所有列。读取超级表时，请将 TAGS 字段放在列表末尾。不要在这里配置
`subtable_name`，连接器会自动把它作为第一列输出。
+`read_columns` 的顺序会决定 `subtable_name` 之后的输出字段顺序。如果后续继续写入 TDengine
Sink，请保持普通列在前、TAGS 字段在后，这样 Sink 才能正确区分普通列和 TAGS 值。
+
## 示例
### 按时间范围读取所有子表
