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 84214d913b [Docs][Connector-V2] Improve RocketMQ Milvus and Qdrant
docs (#11360)
84214d913b is described below
commit 84214d913b1f23a19c62eabcc91f90a9c267b3ec
Author: Daniel Carter <[email protected]>
AuthorDate: Mon Jul 6 22:49:01 2026 +0800
[Docs][Connector-V2] Improve RocketMQ Milvus and Qdrant docs (#11360)
Co-authored-by: DanielCarter-stack
<[email protected]>
---
docs/en/connectors/sink/Milvus.md | 3 ++-
docs/en/connectors/sink/Qdrant.md | 1 +
docs/en/connectors/sink/RocketMQ.md | 10 +++++++---
docs/en/connectors/source/Milvus.md | 3 +--
docs/en/connectors/source/Qdrant.md | 2 ++
docs/en/connectors/source/RocketMQ.md | 9 ++++++++-
docs/zh/connectors/sink/Milvus.md | 3 ++-
docs/zh/connectors/sink/Qdrant.md | 1 +
docs/zh/connectors/sink/RocketMQ.md | 9 ++++++---
docs/zh/connectors/source/Milvus.md | 3 +--
docs/zh/connectors/source/Qdrant.md | 2 ++
docs/zh/connectors/source/RocketMQ.md | 9 ++++++++-
12 files changed, 41 insertions(+), 14 deletions(-)
diff --git a/docs/en/connectors/sink/Milvus.md
b/docs/en/connectors/sink/Milvus.md
index 30ba9c072c..832bb9dae1 100644
--- a/docs/en/connectors/sink/Milvus.md
+++ b/docs/en/connectors/sink/Milvus.md
@@ -60,7 +60,7 @@ Common use cases:
| enable_upsert | boolean | No | true
| Whether to write by upsert instead of insert. Upsert needs a
primary key in the collection schema.
|
| enable_dynamic_field | boolean | No | true
| Whether to enable Milvus dynamic fields when SeaTunnel creates a
collection.
|
| batch_size | int | No | 1000
| Number of records buffered before one write. In streaming jobs,
data can also be flushed when a checkpoint is triggered.
|
-| rate_limit | int | No | 100000
| Maximum write rate used by the connector retry/rate-limit control.
|
+| rate_limit | int | No | 100000
| Milvus collection write-rate limit in MB/s. When greater than 0,
the connector sets `collection.insertRate.max.mb` and
`collection.upsertRate.max.mb` while the writer is running.
|
| partition_key | String | No | -
| Field name used as the Milvus partition key when SeaTunnel creates
a collection.
|
| create_index | boolean | No | false
| Whether to create vector indexes for the target collection. When
copying from Milvus source, existing vector index metadata can be used to
create matching indexes on the target collection.
|
| load_collection | boolean | No | false
| Whether to load the target collection into Milvus memory after the
collection is created. This is useful when the data should be queried
immediately after writing.
|
@@ -74,6 +74,7 @@ Common use cases:
- When a Milvus source reads partitions, the sink can create the same
partition names on the target collection if the target collection does not use
a partition key.
- `create_index = true` only creates vector indexes when index metadata is
available from the upstream catalog, such as when the source is Milvus.
- `enable_upsert = true` writes by primary key and requires a primary key in
the target collection. Use `enable_upsert = false` when writing rows that
should be inserted directly.
+- `rate_limit` changes Milvus collection properties while the writer is open
and resets the insert/upsert limits to `-1` when the writer closes.
- `load_collection = true` loads the collection after creation so it can be
queried immediately after the write finishes.
## Task Example
diff --git a/docs/en/connectors/sink/Qdrant.md
b/docs/en/connectors/sink/Qdrant.md
index 0ab9fa6beb..9a6b0eeadf 100644
--- a/docs/en/connectors/sink/Qdrant.md
+++ b/docs/en/connectors/sink/Qdrant.md
@@ -76,6 +76,7 @@ The value of the primary key column is used as the Qdrant
point ID. Primary key
- The target collection must already exist before the job starts. The
connector does not create collections or vector indexes.
- Vector column names and dimensions must match the vector configuration of
the target Qdrant collection.
- The sink writes vector columns as named vectors. Use a target collection
with named vectors that match the SeaTunnel vector column names.
+- Avoid null values in payload and vector fields before writing to Qdrant,
because the writer converts each configured field directly into a Qdrant point
value.
- The sink writes each incoming row as an upsert request. It does not
interpret `UPDATE` or `DELETE` row kinds as CDC operations.
- The writer buffers up to 64 points and also flushes when the writer is
closed or a commit is prepared. This batch size is not currently exposed as a
connector option.
- Each sink block writes to one collection. Use separate sink blocks when
different collections need different settings.
diff --git a/docs/en/connectors/sink/RocketMQ.md
b/docs/en/connectors/sink/RocketMQ.md
index 588b04e876..f0d948d796 100644
--- a/docs/en/connectors/sink/RocketMQ.md
+++ b/docs/en/connectors/sink/RocketMQ.md
@@ -35,7 +35,7 @@ Writes SeaTunnel rows to an Apache RocketMQ topic. The sink
supports JSON and te
| secret.key | String | no | - | Secret key. Required when `acl.enabled` is
`true`. |
| producer.group | String | no | SeaTunnel-Producer-Group | RocketMQ producer
group ID. |
| tag | String | no | - | RocketMQ message tag written with each message. |
-| partition.key.fields | List | no | - | Field names used to choose the target
RocketMQ queue. Every listed field must exist in the upstream schema. |
+| partition.key.fields | List | no | - | Field names serialized as the
RocketMQ message key. Every listed field must exist in the upstream schema. |
| format | String | no | json | Message format. Supported values are `json`
and `text`. |
| field.delimiter | String | no | `,` | Field delimiter used when `format =
text`. |
| producer.send.sync | Boolean | no | false | Whether to send messages
synchronously. When `false`, messages are sent asynchronously. |
@@ -48,9 +48,13 @@ Writes SeaTunnel rows to an Apache RocketMQ topic. The sink
supports JSON and te
### partition.key.fields
-`partition.key.fields` controls which RocketMQ queue receives each message.
SeaTunnel hashes the configured field values and uses the hash to choose a
queue. If this option is not set, RocketMQ chooses the queue.
+`partition.key.fields` controls the RocketMQ message key. SeaTunnel serializes
+the configured field values as JSON and writes that value to `Message.keys`.
For
+non-transactional sends, the same key is also used by RocketMQ's hash queue
+selector, so rows with the same key are sent to the same queue. If this option
is
+not set, RocketMQ chooses the queue.
-For example, if the input schema contains `c_int`, this configuration uses
`c_int` as the queue key:
+For example, if the input schema contains `c_int`, this configuration uses
`c_int` to build the message key:
```hocon
partition.key.fields = ["c_int"]
diff --git a/docs/en/connectors/source/Milvus.md
b/docs/en/connectors/source/Milvus.md
index 590ff0eb06..8b8b5d3325 100644
--- a/docs/en/connectors/source/Milvus.md
+++ b/docs/en/connectors/source/Milvus.md
@@ -54,13 +54,12 @@ Common use cases:
| token | String | Yes | - | Milvus authentication token. For
a local Milvus server this is usually `username:password`.
|
| database | String | No | default | Source database.
|
| collection | String | No | - | Source collection. If it is set,
only this collection is read. If it is not set, all collections under
`database` are read. |
-| batch_size | Int | No | 1000 | Number of records fetched in one
source read batch.
|
-| rate_limit | Int | No | 1000000 | Maximum source read rate limit.
|
## Notes
- `database` defaults to `default`, so simple local Milvus jobs do not need to
set it.
- `collection` is optional. Set it when the job should read exactly one
collection.
+- The source currently exposes only `url`, `token`, `database`, and
`collection` as user-facing options. Read batching and retry behavior use
connector defaults.
- When `collection` is not set, the source discovers all collections in
`database` and exposes each collection as a separate SeaTunnel table.
- The source splits work by Milvus partition. Collections with a partition key
are read with one split; collections without a partition key are split by
partition name and assigned across readers.
- When the source reads a collection with partitions, downstream Milvus sink
can use that metadata to create the same partition names on the target
collection.
diff --git a/docs/en/connectors/source/Qdrant.md
b/docs/en/connectors/source/Qdrant.md
index a7067cda71..7448375397 100644
--- a/docs/en/connectors/source/Qdrant.md
+++ b/docs/en/connectors/source/Qdrant.md
@@ -46,6 +46,7 @@ Each Qdrant record is a point:
- Vector columns are read from Qdrant vectors. For named vectors, the
SeaTunnel column name must match the Qdrant vector name.
- Other supported columns are read from the Qdrant point payload.
- If the collection uses one default unnamed vector, use `default_vector` as
the SeaTunnel vector column name.
+- Include only payload or vector fields that exist on the points being read.
The source maps fields directly by name and expects configured fields to be
present.
Example:
@@ -110,6 +111,7 @@ Source plugin common parameters, see [Source Common
Options](../common-options/s
- The collection must already exist before the job starts.
- Vector field names and dimensions in Qdrant must match the vector columns in
the SeaTunnel schema.
+- For a collection with one unnamed vector, name the SeaTunnel vector column
`default_vector`.
- The source is bounded and reads with one split, so it is not a parallel
source.
- Qdrant source emits `INSERT` rows only; it does not read CDC changes.
diff --git a/docs/en/connectors/source/RocketMQ.md
b/docs/en/connectors/source/RocketMQ.md
index 9c44f68c0e..695b29c6d2 100644
--- a/docs/en/connectors/source/RocketMQ.md
+++ b/docs/en/connectors/source/RocketMQ.md
@@ -36,7 +36,7 @@ Reads messages from Apache RocketMQ topics. The connector can
read one or more t
| topics | String | no | - | Topic name list separated by commas, for example
`"topic_a,topic_b"`. Configure only one of `topics`, `tables_configs`, and
`table_list`. |
| tables_configs | List | no | - | Multi-table read configuration. Each item
must contain `topics` and can contain `format`, `schema`, `tags`, `start.mode`,
`start.mode.timestamp`, `start.mode.offsets`, and `ignore_parse_errors`. |
| table_list | List | no | - | Deprecated. Use `tables_configs` instead. |
-| tags | String | no | - | Tag list separated by commas. Only messages with
these tags are consumed. |
+| tags | String | no | - | Tag list separated by commas. Only messages whose
RocketMQ tag exactly matches one configured value are consumed. |
| acl.enabled | Boolean | no | false | Whether to enable RocketMQ ACL
authentication. |
| access.key | String | no | - | Access key. Required when `acl.enabled` is
`true`. |
| secret.key | String | no | - | Secret key. Required when `acl.enabled` is
`true`. |
@@ -92,6 +92,13 @@ 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.
+### Tags
+
+`tags` uses a comma-separated list such as `tag_a,tag_b`. The connector
compares
+the pulled message tag with these values, so do not use RocketMQ tag expression
+syntax such as `tag_a || tag_b` here. In multi-table jobs, each
`tables_configs`
+entry can set its own `tags` filter.
+
### 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.
diff --git a/docs/zh/connectors/sink/Milvus.md
b/docs/zh/connectors/sink/Milvus.md
index 9dbd15d962..bc4fdb03f6 100644
--- a/docs/zh/connectors/sink/Milvus.md
+++ b/docs/zh/connectors/sink/Milvus.md
@@ -60,7 +60,7 @@ SeaTunnel 表结构创建目标集合，并支持向量字段、动态字段、
| enable_upsert | boolean | 否 | true
| 是否使用更新插入模式写入，而不是普通插入模式。更新插入需要集合中有主键。
|
| enable_dynamic_field | boolean | 否 | true
| SeaTunnel 创建集合时，是否启用 Milvus 动态字段。
|
| batch_size | int | 否 | 1000
| 写入前缓存的记录数。流任务中，检查点触发时也会刷新写入。
|
-| rate_limit | int | 否 | 100000
| 连接器用于重试和限流控制的最大写入速率。
|
+| rate_limit | int | 否 | 100000
| Milvus collection 写入限速，单位 MB/s。大于 0 时，写入器运行期间会设置
`collection.insertRate.max.mb` 和 `collection.upsertRate.max.mb`。 |
| partition_key | String | 否 | -
| SeaTunnel 创建集合时使用的 Milvus 分区键字段名。
|
| create_index | boolean | 否 | false
| 是否为目标集合创建向量索引。从 Milvus 源端复制时，可以利用上游索引元数据在目标集合创建相同索引。
|
| load_collection | boolean | 否 | false
| 创建集合后是否把目标集合加载到 Milvus 内存中。如果写完马上要查询，可以开启。
|
@@ -74,6 +74,7 @@ SeaTunnel 表结构创建目标集合，并支持向量字段、动态字段、
- Milvus 源端读取分区后，如果目标集合没有使用分区键，接收器可以在目标集合创建相同分区名。
- `create_index = true` 只有在上游目录信息能提供索引元数据时才会创建向量索引，例如上游也是 Milvus。
- `enable_upsert = true` 会按主键更新插入，目标集合中需要有主键。希望直接追加写入时，可以设置 `enable_upsert =
false`。
+- `rate_limit` 会在写入器运行期间修改 Milvus collection 属性，并在写入器关闭时把 insert/upsert 限速重置为
`-1`。
- `load_collection = true` 会在创建集合后加载集合，这样写入完成后可以马上查询。
## 任务示例
diff --git a/docs/zh/connectors/sink/Qdrant.md
b/docs/zh/connectors/sink/Qdrant.md
index fa9e947967..fa306fb5af 100644
--- a/docs/zh/connectors/sink/Qdrant.md
+++ b/docs/zh/connectors/sink/Qdrant.md
@@ -76,6 +76,7 @@ Sink 插件通用参数，请参考 [Sink 通用选项](../common-options/sink-c
- 目标 collection 必须在作业启动前已经存在。连接器不会自动创建 collection 或向量索引。
- 向量列名和维度必须与目标 Qdrant collection 的向量配置一致。
- sink 会把向量列写成命名向量。目标 collection 中的命名向量需要和 SeaTunnel 向量列名一致。
+- 写入 Qdrant 前，请先处理 payload 和向量字段中的空值，因为写入器会把每个配置字段直接转换成 Qdrant point 值。
- sink 会把每条输入行作为 upsert 写入，不会按 `UPDATE` 或 `DELETE` 行类型执行 CDC 语义。
- 写入器最多缓存 64 个 point，并且在关闭写入器或准备提交时也会刷新。这一批次大小目前不是可配置项。
- 每个 sink 配置块写入一个 collection。如果不同 collection 需要不同配置，请使用多个 sink 配置块。
diff --git a/docs/zh/connectors/sink/RocketMQ.md
b/docs/zh/connectors/sink/RocketMQ.md
index d9a2f41767..86450321f6 100644
--- a/docs/zh/connectors/sink/RocketMQ.md
+++ b/docs/zh/connectors/sink/RocketMQ.md
@@ -35,7 +35,7 @@ import ChangeLog from '../changelog/connector-rocketmq.md';
| secret.key | String | 否 | - | 秘密密钥。`acl.enabled = true` 时必填。 |
| producer.group | String | 否 | SeaTunnel-Producer-Group | RocketMQ 生产者组 ID。 |
| tag | String | 否 | - | 写入每条消息时使用的 RocketMQ tag。 |
-| partition.key.fields | List | 否 | - | 用来选择 RocketMQ 队列的字段名。配置的字段必须存在于上游
schema 中。 |
+| partition.key.fields | List | 否 | - | 会被序列化为 RocketMQ 消息 key
的字段名。配置的字段必须存在于上游 schema 中。 |
| format | String | 否 | json | 消息格式。支持 `json` 和 `text`。 |
| field.delimiter | String | 否 | `,` | `format = text` 时使用的字段分隔符。 |
| producer.send.sync | Boolean | 否 | false | 是否同步发送消息。为 `false` 时异步发送。 |
@@ -48,9 +48,12 @@ import ChangeLog from '../changelog/connector-rocketmq.md';
### partition.key.fields
-`partition.key.fields` 控制消息写入哪个 RocketMQ 队列。SeaTunnel
会对这些字段的值做哈希，并用哈希结果选择队列。如果不配置该参数，则由 RocketMQ 自行选择队列。
+`partition.key.fields` 控制 RocketMQ 消息 key。SeaTunnel 会把这些字段的值序列化成
+JSON，并写入 `Message.keys`。在非事务发送时，同一个 key 还会交给 RocketMQ
+的哈希队列选择器，因此相同 key 的数据会进入同一个队列。如果不配置该参数，则由
+RocketMQ 自行选择队列。
-例如，上游字段中有 `c_int` 时，可以这样配置：
+例如，上游字段中有 `c_int` 时，可以这样把 `c_int` 用作消息 key：
```hocon
partition.key.fields = ["c_int"]
diff --git a/docs/zh/connectors/source/Milvus.md
b/docs/zh/connectors/source/Milvus.md
index c2543cdf8f..e1166d5a1c 100644
--- a/docs/zh/connectors/source/Milvus.md
+++ b/docs/zh/connectors/source/Milvus.md
@@ -54,13 +54,12 @@ Milvus 源连接器用于从 Milvus 或 Zilliz Cloud 读取数据。它可以读
| token | String | 是 | - | Milvus 认证令牌。本地 Milvus 通常使用
`username:password`。 |
| database | String | 否 | default | 源数据库。
|
| collection | String | 否 | - | 源集合。配置后只读取这个集合；不配置时读取 `database`
下的所有集合。 |
-| batch_size | Int | 否 | 1000 | 每批从 Milvus 读取的数据条数。
|
-| rate_limit | Int | 否 | 1000000 | 源端读取速率限制。
|
## 注意事项
- `database` 默认是 `default`，本地 Milvus 的简单任务通常不用配置。
- `collection` 是可选项。只想读一个集合时再配置。
+- 源端目前对用户开放的参数只有 `url`、`token`、`database` 和 `collection`。读取批次和重试行为使用连接器内置默认值。
- 不配置 `collection` 时，源端会发现 `database` 下的所有集合，并把每个集合作为一张独立的 SeaTunnel 表输出。
- 源端会按 Milvus 分区拆分读取任务。有分区键的集合使用一个 split 读取；没有分区键的集合会按分区名拆分，并分配给多个 reader。
- 源端读取带分区的集合时，下游 Milvus 接收器可以利用这些元数据在目标集合创建相同分区名。
diff --git a/docs/zh/connectors/source/Qdrant.md
b/docs/zh/connectors/source/Qdrant.md
index c40ec13d66..f73aba300a 100644
--- a/docs/zh/connectors/source/Qdrant.md
+++ b/docs/zh/connectors/source/Qdrant.md
@@ -46,6 +46,7 @@ Qdrant 中的每条记录叫作 point：
- 向量列会从 Qdrant 向量中读取。读取命名向量时，SeaTunnel 列名必须和 Qdrant 向量名一致。
- 其他支持的列会从 Qdrant point payload 中读取。
- 如果 collection 使用默认的未命名向量，请使用 `default_vector` 作为 SeaTunnel 向量列名。
+- schema 中只配置实际存在的 payload 或向量字段。Source 会按字段名直接取值，配置的字段需要在读取到的 point 中存在。
示例：
@@ -110,6 +111,7 @@ Source 插件通用参数，请参考 [Source 通用选项](../common-options/so
- 作业启动前，collection 必须已经存在。
- Qdrant 中的向量字段名和维度必须与 SeaTunnel schema 中的向量列一致。
+- 如果 collection 只有一个未命名向量，请把 SeaTunnel 向量列命名为 `default_vector`。
- Qdrant source 是有界读取，并且只使用一个 split，不支持并行读取。
- Qdrant source 只输出 `INSERT` 行，不读取 CDC 变更。
diff --git a/docs/zh/connectors/source/RocketMQ.md
b/docs/zh/connectors/source/RocketMQ.md
index 1abca1cc75..08888b3cfb 100644
--- a/docs/zh/connectors/source/RocketMQ.md
+++ b/docs/zh/connectors/source/RocketMQ.md
@@ -36,7 +36,7 @@ import ChangeLog from '../changelog/connector-rocketmq.md';
| topics | String | 否 | - | topic 名称，多个 topic 使用逗号分隔，例如
`"topic_a,topic_b"`。`topics`、`tables_configs` 和 `table_list` 只能配置其中一个。 |
| tables_configs | List | 否 | - | 多表读取配置。每一项必须包含 `topics`，并可配置
`format`、`schema`、`tags`、`start.mode`、`start.mode.timestamp`、`start.mode.offsets`
和 `ignore_parse_errors`。 |
| table_list | List | 否 | - | 已废弃，请使用 `tables_configs`。 |
-| tags | String | 否 | - | tag 名称，多个 tag 使用逗号分隔。只消费匹配这些 tag 的消息。 |
+| tags | String | 否 | - | tag 名称，多个 tag 使用逗号分隔。只消费 RocketMQ tag 与配置值完全匹配的消息。 |
| acl.enabled | Boolean | 否 | false | 是否启用 RocketMQ ACL 鉴权。 |
| access.key | String | 否 | - | 访问密钥。`acl.enabled = true` 时必填。 |
| secret.key | String | 否 | - | 秘密密钥。`acl.enabled = true` 时必填。 |
@@ -86,6 +86,13 @@ start.mode.timestamp = 1667179890315
当 `format = text` 时，SeaTunnel 会按 `field.delimiter` 拆分消息体，并按照 schema
字段顺序映射数据。如果不配置 `schema`，消息体会作为单个文本值读取。
+### 标签
+
+`tags` 使用逗号分隔，例如 `tag_a,tag_b`。连接器会把拉取到的消息 tag
+和这些值逐个做精确匹配，因此这里不要使用 RocketMQ 的 tag 表达式语法，例如
+`tag_a || tag_b`。在多表读取任务中，每个 `tables_configs` 条目都可以配置自己的
+`tags` 过滤条件。
+
### 多表读取
当不同 topic 的字段结构不一样时，使用 `tables_configs`。每一项都必须包含 `topics`，并且可以单独配置
`schema`、`format`、`tags` 和启动消费位置。如果没有配置 `schema.table`，输出表名默认使用 topic 名称。
