This is an automated email from the ASF dual-hosted git repository.
zhangshenghang 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 9f50372836 [Docs][Connector-V2] Improve RocketMQ connector docs
(#11272)
9f50372836 is described below
commit 9f50372836607cabb4e5cad7eb65f29ce242ea19
Author: Daniel Carter <[email protected]>
AuthorDate: Fri Jul 3 08:19:18 2026 +0800
[Docs][Connector-V2] Improve RocketMQ connector docs (#11272)
Co-authored-by: DanielCarter-stack
<[email protected]>
Co-authored-by: Shenghang <[email protected]>
---
docs/en/connectors/sink/RocketMQ.md | 188 +++++++++++-------------
docs/en/connectors/source/RocketMQ.md | 261 ++++++++++------------------------
docs/zh/connectors/sink/RocketMQ.md | 206 +++++++++++++--------------
docs/zh/connectors/source/RocketMQ.md | 211 ++++++++++++++-------------
4 files changed, 367 insertions(+), 499 deletions(-)
diff --git a/docs/en/connectors/sink/RocketMQ.md
b/docs/en/connectors/sink/RocketMQ.md
index f7dff4c70d..f9a22f0a7f 100644
--- a/docs/en/connectors/sink/RocketMQ.md
+++ b/docs/en/connectors/sink/RocketMQ.md
@@ -6,7 +6,7 @@ import ChangeLog from '../changelog/connector-rocketmq.md';
## Support Apache RocketMQ Version
-- 4.9.0 (Or a newer version, for reference)
+- 4.9.0 or newer
## Support These Engines
@@ -17,127 +17,127 @@ import ChangeLog from
'../changelog/connector-rocketmq.md';
## Key Features
- [x] [exactly-once](../../introduction/concepts/connector-v2-features.md)
-
-By default, we will use 2pc to guarantee the message is sent to RocketMQ
exactly once.
- [ ] [timer flush](../../introduction/concepts/connector-v2-features.md)
## Description
-Write Rows to a Apache RocketMQ topic.
+Writes SeaTunnel rows to an Apache RocketMQ topic. The sink supports JSON and
text message bodies, optional message tags, synchronous sending, partition key
fields, and transactional messages when `exactly.once = true`.
## Sink Options
-| Name | Type | Required | Default |
Description
|
-|----------------------|---------|----------|--------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| topic | string | yes | - |
`RocketMQ topic` name.
|
-| name.srv.addr | string | yes | - |
`RocketMQ` name server cluster address.
|
-| acl.enabled | Boolean | no | false | If
true, access control is enabled, and access key and secret key need to be
configured.
|
-| access.key | String | no | | When
ACL_ENABLED is true, access key cannot be empty
|
-| secret.key | String | no | | When
ACL_ENABLED is true, secret key cannot be empty
|
-| producer.group | String | no | SeaTunnel-Producer-Group |
RocketMQ producer group id.
|
-| tag | String | no | - |
`RocketMQ` message tag.
|
-| partition.key.fields | array | no | - | -
|
-| format | String | no | json | Data
format. The default format is json. Optional text format. The default field
separator is ",".If you customize the delimiter, add the "field_delimiter"
option. |
-| field.delimiter | String | no | , |
Customize the field delimiter for data format.
|
-| producer.send.sync | Boolean | no | false | If
true, the message will be sync sent.
|
-| exactly.once | Boolean | no | false | If
true, the transaction message will be sent.
|
-| max.message.size | int | no | 4194304 |
Maximum allowed message body size in bytes.
|
-| send.message.timeout | int | no | 3000 |
Timeout for sending messages in milliseconds.
|
-| common-options | config | no | - | Sink
plugin common parameters, please refer to [Sink Common
Options](../common-options/sink-common-options.md) for details.
|
+| Name | Type | Required | Default | Description |
+|------|------|----------|---------|-------------|
+| topic | String | yes | - | RocketMQ topic name. |
+| name.srv.addr | String | yes | - | RocketMQ name server address, for example
`localhost:9876`. |
+| 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`. |
+| 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. |
+| 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. |
+| exactly.once | Boolean | no | false | Whether to send transactional messages
for exactly-once delivery. |
+| max.message.size | int | no | 4194304 | Maximum message body size in bytes. |
+| send.message.timeout | int | no | 3000 | Send timeout in milliseconds. |
+| common-options | config | no | - | Sink common options. See [Sink Common
Options](../common-options/sink-common-options.md). |
-### partition.key.fields [array]
+## Option Notes
-Configure which fields are used as the key of the RocketMQ message.
+### partition.key.fields
-For example, if you want to use value of fields from upstream data as key, you
can assign field names to this property.
+`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.
-Upstream data is the following:
+For example, if the input schema contains `c_int`, this configuration uses
`c_int` as the queue key:
-| name | age | data |
-|------|-----|---------------|
-| Jack | 16 | data-example1 |
-| Mary | 23 | data-example2 |
+```hocon
+partition.key.fields = ["c_int"]
+```
-If name is set as the key, then the hash value of the name column will
determine which partition the message is sent to.
+### exactly.once
-## Task Example
+The sink supports exactly-once writes through RocketMQ transactional messages.
This behavior is disabled by default. Set `exactly.once = true` when the
RocketMQ cluster and the job checkpoint settings are ready for transactional
writes.
-### Fake to Rocketmq Simple
+## Task Examples
-> The data is randomly generated and asynchronously sent to the test topic
+### Write JSON Messages
```hocon
env {
parallelism = 1
+ job.mode = "BATCH"
}
source {
FakeSource {
+ row.num = 10
schema = {
fields {
- c_map = "map<string, string>"
- c_array = "array<int>"
c_string = string
- c_boolean = boolean
- c_tinyint = tinyint
- c_smallint = smallint
c_int = int
- c_bigint = bigint
- c_float = float
- c_double = double
- c_decimal = "decimal(30, 8)"
- c_bytes = bytes
- c_date = date
c_timestamp = timestamp
}
}
}
}
-transform {
- # If you would like to get more information about how to configure seatunnel
and see full list of transform plugins,
- # please go to https://seatunnel.apache.org/docs/transforms
-}
-
sink {
Rocketmq {
- name.srv.addr = "localhost:9876"
+ name.srv.addr = "rocketmq-e2e:9876"
topic = "test_topic"
+ partition.key.fields = ["c_int"]
+ producer.send.sync = true
}
}
-
```
-### Rocketmq To Rocketmq Simple
-
-> Consuming Rocketmq writes to c_int field Hash number of partitions written
to different partitions This is the default asynchronous way to write
+### Write Text Messages
```hocon
env {
parallelism = 1
+ job.mode = "BATCH"
}
source {
+ FakeSource {
+ row.num = 10
+ schema = {
+ fields {
+ id = bigint
+ content = string
+ }
+ }
+ }
+}
+
+sink {
Rocketmq {
- name.srv.addr = "localhost:9876"
- topics = "test_topic"
- plugin_output = "rocketmq_table"
+ name.srv.addr = "rocketmq-e2e:9876"
+ topic = "test_text_topic"
+ format = text
+ field.delimiter = ","
+ producer.send.sync = true
+ }
+}
+```
+
+### Write Messages With a Tag
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ FakeSource {
+ row.num = 10
schema = {
fields {
- c_map = "map<string, string>"
- c_array = "array<int>"
c_string = string
- c_boolean = boolean
- c_tinyint = tinyint
- c_smallint = smallint
c_int = int
- c_bigint = bigint
- c_float = float
- c_double = double
- c_decimal = "decimal(30, 8)"
- c_bytes = bytes
- c_date = date
- c_timestamp = timestamp
}
}
}
@@ -145,64 +145,48 @@ source {
sink {
Rocketmq {
- name.srv.addr = "localhost:9876"
- topic = "test_topic_sink"
- partition.key.fields = ["c_int"]
+ name.srv.addr = "rocketmq-e2e:9876"
+ topic = "test_topic_message_tag"
+ tag = "test_tag"
+ partition.key.fields = ["c_string"]
+ producer.send.sync = true
}
}
```
-### Timestamp consumption write Simple
-
-> This is a stream consumption specified time stamp consumption, when there
are new partitions added the program will refresh the perception and
consumption at intervals, and write to another topic type
+### Read and Write RocketMQ
```hocon
-
env {
parallelism = 1
job.mode = "STREAMING"
+ checkpoint.interval = 5000
}
source {
Rocketmq {
- name.srv.addr = "localhost:9876"
- topics = "test_topic"
+ name.srv.addr = "rocketmq-e2e:9876"
+ topics = "test_topic_source"
plugin_output = "rocketmq_table"
+ format = json
start.mode = "CONSUME_FROM_FIRST_OFFSET"
- batch.size = "400"
- consumer.group = "test_topic_group"
- format = "json"
+ consumer.group = "rocketmq_to_rocketmq_group"
schema = {
fields {
- c_map = "map<string, string>"
- c_array = "array<int>"
+ id = bigint
c_string = string
- c_boolean = boolean
- c_tinyint = tinyint
- c_smallint = smallint
- c_int = int
- c_bigint = bigint
- c_float = float
- c_double = double
- c_decimal = "decimal(30, 8)"
- c_bytes = bytes
- c_date = date
- c_timestamp = timestamp
}
}
}
}
-transform {
- # If you would like to get more information about how to configure seatunnel
and see full list of transform plugins,
- # please go to https://seatunnel.apache.org/docs/transforms
-}
sink {
Rocketmq {
- name.srv.addr = "localhost:9876"
- topic = "test_topic"
- partition.key.fields = ["c_int"]
- producer.send.sync = true
+ plugin_input = "rocketmq_table"
+ name.srv.addr = "rocketmq-e2e:9876"
+ topic = "test_topic_sink"
+ partition.key.fields = ["id"]
+ exactly.once = true
}
}
```
diff --git a/docs/en/connectors/source/RocketMQ.md
b/docs/en/connectors/source/RocketMQ.md
index 279ad13c1c..27b55ad39f 100644
--- a/docs/en/connectors/source/RocketMQ.md
+++ b/docs/en/connectors/source/RocketMQ.md
@@ -6,7 +6,7 @@ import ChangeLog from '../changelog/connector-rocketmq.md';
## Support Apache RocketMQ Version
-- 4.9.0 (Or a newer version, for reference)
+- 4.9.0 or newer
## Support These Engines
@@ -25,53 +25,60 @@ import ChangeLog from '../changelog/connector-rocketmq.md';
## Description
-Source connector for Apache RocketMQ.
+Reads messages from Apache RocketMQ topics. The connector can read one or more
topics with one schema, or use `tables_configs` to read multiple topics with
different schemas.
## Source Options
-| Name | Type | Required | Default
| Description
|
-|-------------------------------------|---------|----------|----------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| topics | String | no | -
| `RocketMQ topic` name. If there are multiple `topics`, use `,` to
split, for example: `"tpc1,tpc2"`. You can configure only one of `topics` and
`tables_configs` at the same time. |
-| tables_configs | List | no | -
| Multi-table mode config list. Each item configures one table and
supports: `topics`, `format`, `schema`, `tags`, `start.mode`,
`start.mode.timestamp`, `start.mode.offsets`, `ignore_parse_errors`. You can
configure only one of `topics` and `tables_configs` at the same time. |
-| table_list | List | no | -
| Deprecated, use `tables_configs` instead.
|
-| name.srv.addr | String | yes | -
| `RocketMQ` name server cluster address.
|
-| tags | String | no | -
| `RocketMQ tag` name. If there are multiple `tags`, use `,` to split,
for example: `"tag1,tag2"`.
|
-| acl.enabled | Boolean | no | false
| If true, access control is enabled, and access key and secret key
need to be configured.
|
-| access.key | String | no |
|
|
-| secret.key | String | no |
| When ACL_ENABLED is true, secret key cannot be empty.
|
-| batch.size | int | no | 100
| `RocketMQ` consumer pull batch size
|
-| consumer.group | String | no |
SeaTunnel-Consumer-Group | `RocketMQ consumer group id`, used to distinguish
different consumer groups.
|
-| commit.on.checkpoint | Boolean | no | true
| If true the consumer's offset will be periodically committed in the
background.
|
-| schema | | no | -
| The structure of the data, including field names and field types.
For more details, please refer to [Schema
Feature](../../introduction/concepts/schema-feature.md).
|
-| format | String | no | json
| Data format. The default format is json. Optional text format. The
default field separator is ",".If you customize the delimiter, add the
"field.delimiter" option. |
-| field.delimiter | String | no | ,
| Customize the field delimiter for data format
|
-| start.mode | String | no |
CONSUME_FROM_GROUP_OFFSETS | The initial consumption pattern of consumers,there
are several types:
[CONSUME_FROM_LAST_OFFSET],[CONSUME_FROM_FIRST_OFFSET],[CONSUME_FROM_GROUP_OFFSETS],[CONSUME_FROM_TIMESTAMP],[CONSUME_FROM_SPECIFIC_OFFSETS]
|
-| start.mode.offsets | | no |
|
|
-| start.mode.timestamp | Long | no |
| The time required for consumption mode to be
"CONSUME_FROM_TIMESTAMP".
|
-| partition.discovery.interval.millis | long | no | -1
| The interval for dynamically discovering topics and partitions.
|
-| ignore_parse_errors | Boolean | no | false
| Optional flag to skip parse errors instead of failing.
|
-| consumer.poll.timeout.millis | long | no | 5000
| The poll timeout in milliseconds.
|
-| common-options | config | no | -
| Source plugin common parameters, please refer to [Source Common
Options](../common-options/source-common-options.md) for details.
|
-
-### start.mode.offsets
-
-The offset required for consumption mode to be "CONSUME_FROM_SPECIFIC_OFFSETS".
-
-for example:
+| Name | Type | Required | Default | Description |
+|------|------|----------|---------|-------------|
+| name.srv.addr | String | yes | - | RocketMQ name server address, for example
`localhost:9876`. |
+| 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. |
+| 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`. |
+| batch.size | int | no | 100 | Maximum number of messages pulled each time. |
+| consumer.group | String | no | SeaTunnel-Consumer-Group | RocketMQ consumer
group ID. |
+| commit.on.checkpoint | Boolean | no | true | Whether to commit offsets when
SeaTunnel checkpoints are completed. |
+| schema | config | no | - | Message schema. See [Schema
Feature](../../introduction/concepts/schema-feature.md). If omitted, the
connector reads message bodies as text. |
+| format | String | no | json | Message format. Supported values are `json`
and `text`. |
+| field.delimiter | String | no | `,` | Field delimiter used when `format =
text`. |
+| start.mode | String | no | CONSUME_FROM_GROUP_OFFSETS | Startup position.
Supported values: `CONSUME_FROM_LAST_OFFSET`, `CONSUME_FROM_FIRST_OFFSET`,
`CONSUME_FROM_GROUP_OFFSETS`, `CONSUME_FROM_TIMESTAMP`,
`CONSUME_FROM_SPECIFIC_OFFSETS`. |
+| 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. |
+| 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). |
+
+## Option Notes
+
+### Startup Position
+
+`start.mode` controls where the source starts reading:
+
+- `CONSUME_FROM_GROUP_OFFSETS`: start from committed offsets of the consumer
group.
+- `CONSUME_FROM_FIRST_OFFSET`: start from the earliest available offset.
+- `CONSUME_FROM_LAST_OFFSET`: start from the latest available offset.
+- `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`.
```hocon
+start.mode = "CONSUME_FROM_SPECIFIC_OFFSETS"
start.mode.offsets = {
- topic1-0 = 70
- topic1-1 = 10
- topic1-2 = 10
+ test_topic-0 = 50
}
```
-## Task Example
+### Multi-Table Read
-### Simple
+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.
-> Consumer reads Rocketmq data and prints it to the console type
+## Task Examples
+
+### Read JSON Messages
```hocon
env {
@@ -84,42 +91,26 @@ source {
name.srv.addr = "rocketmq-e2e:9876"
topics = "test_topic_json"
plugin_output = "rocketmq_table"
+ format = json
schema = {
fields {
id = bigint
- c_map = "map<string, smallint>"
- c_array = "array<tinyint>"
c_string = string
- c_boolean = boolean
- c_tinyint = tinyint
- c_smallint = smallint
c_int = int
- c_bigint = bigint
- c_float = float
- c_double = double
- c_decimal = "decimal(2, 1)"
- c_bytes = bytes
- c_date = date
c_timestamp = timestamp
}
}
}
}
-transform {
- # If you would like to get more information about how to configure seatunnel
and see full list of transform plugins,
- # please go to https://seatunnel.apache.org/docs/transforms
-}
-
sink {
Console {
+ plugin_input = "rocketmq_table"
}
}
```
-### Specified format consumption simple
-
-> When I consume the topic data in json format parsing and pulling the number
of bars each time is 400, the consumption starts from the original location
+### Read Text Messages With Tags
```hocon
env {
@@ -129,154 +120,54 @@ env {
source {
Rocketmq {
- name.srv.addr = "localhost:9876"
- topics = "test_topic"
+ name.srv.addr = "rocketmq-e2e:9876"
+ topics = "test_topic_text"
plugin_output = "rocketmq_table"
- start.mode = "CONSUME_FROM_FIRST_OFFSET"
- batch.size = "400"
- consumer.group = "test_topic_group"
- format = "json"
- format = json
+ format = text
+ field.delimiter = ","
+ tags = "tag_a,tag_b"
schema = {
fields {
- c_map = "map<string, string>"
- c_array = "array<int>"
- c_string = string
- c_boolean = boolean
- c_tinyint = tinyint
- c_smallint = smallint
- c_int = int
- c_bigint = bigint
- c_float = float
- c_double = double
- c_decimal = "decimal(30, 8)"
- c_bytes = bytes
- c_date = date
- c_timestamp = timestamp
+ id = bigint
+ content = string
}
}
}
}
-transform {
- # If you would like to get more information about how to configure seatunnel
and see full list of transform plugins,
- # please go to https://seatunnel.apache.org/docs/transforms
-}
sink {
Console {
+ plugin_input = "rocketmq_table"
}
}
```
-### Specified timestamp simple
-
-> This is to specify a time to consume, and I dynamically sense the existence
of a new partition every 1000 milliseconds to pull the consumption
+### Read From Specific Offsets
```hocon
env {
parallelism = 1
- spark.app.name = "SeaTunnel"
- spark.executor.instances = 2
- spark.executor.cores = 1
- spark.executor.memory = "1g"
- spark.master = local
job.mode = "BATCH"
}
source {
Rocketmq {
- name.srv.addr = "localhost:9876"
- topics = "test_topic"
- partition.discovery.interval.millis = "1000"
- start.mode.timestamp="1694508382000"
- consumer.group="test_topic_group"
- format="json"
+ name.srv.addr = "rocketmq-e2e:9876"
+ topics = "test_topic_source"
+ plugin_output = "rocketmq_table"
format = json
- schema = {
- fields {
- c_map = "map<string, string>"
- c_array = "array<int>"
- c_string = string
- c_boolean = boolean
- c_tinyint = tinyint
- c_smallint = smallint
- c_int = int
- c_bigint = bigint
- c_float = float
- c_double = double
- c_decimal = "decimal(30, 8)"
- c_bytes = bytes
- c_date = date
- c_timestamp = timestamp
- }
+ start.mode = "CONSUME_FROM_SPECIFIC_OFFSETS"
+ start.mode.offsets = {
+ test_topic_source-0 = 50
}
- }
-}
-
-transform {
- # If you would like to get more information about how to configure seatunnel
and see full list of transform plugins,
- # please go to https://seatunnel.apache.org/docs/transforms
-}
-
-sink {
- Console {
- }
-}
-```
-
-### Specified tag example
-
-> Here you can specify a tag to consume data. If there are multiple tags, use
`,` to separate them, for example: "tag1,tag2"
-
-```hocon
-env {
- parallelism = 1
- job.mode = "BATCH"
-
- # You can set spark configuration here
- spark.app.name = "SeaTunnel"
- spark.executor.instances = 2
- spark.executor.cores = 1
- spark.executor.memory = "1g"
- spark.master = local
-}
-
-source {
- Rocketmq {
- plugin_output = "rocketmq_table"
- name.srv.addr = "localhost:9876"
- topics = "test_topic"
- format = text
- # The default field delimiter is ","
- field_delimiter = ","
- tags = "test_tag"
schema = {
fields {
id = bigint
- c_map = "map<string, smallint>"
- c_array = "array<tinyint>"
- c_string = string
- c_boolean = boolean
- c_tinyint = tinyint
- c_smallint = smallint
- c_int = int
- c_bigint = bigint
- c_float = float
- c_double = double
- c_decimal = "decimal(2, 1)"
- c_bytes = bytes
- c_date = date
- c_timestamp = timestamp
}
}
}
}
-transform {
- # If you would like to get more information about how to configure seatunnel
and see full list of transform plugins,
- # please go to https://seatunnel.apache.org/docs/transforms
-}
-
sink {
Console {
plugin_input = "rocketmq_table"
@@ -284,9 +175,7 @@ sink {
}
```
-### Multiple RocketMQ Source
-
-> Read from multiple topics with different schemas. Use `tables_configs` to
configure each topic independently.
+### Read Multiple Topics With Different Schemas
```hocon
env {
@@ -296,28 +185,30 @@ env {
source {
Rocketmq {
- name.srv.addr = "localhost:9876"
- consumer.group = "multi_table_group"
- start.mode = "CONSUME_FROM_FIRST_OFFSET"
+ name.srv.addr = "rocketmq-e2e:9876"
+ start.mode = "CONSUME_FROM_LAST_OFFSET"
tables_configs = [
{
- topics = "topic_1"
- format = "json"
+ topics = "test_topic_multi_a"
+ start.mode = "CONSUME_FROM_FIRST_OFFSET"
+ format = json
schema = {
fields {
- id = int
- name = string
+ id = bigint
+ c_string = string
}
}
},
{
- topics = "topic_2"
- format = "json"
+ topics = "test_topic_multi_b"
+ start.mode = "CONSUME_FROM_FIRST_OFFSET"
+ tags = "tag_b"
+ format = json
schema = {
+ table = "rocketmq_multi_custom"
fields {
- id = int
+ id = bigint
description = string
- weight = double
}
}
}
diff --git a/docs/zh/connectors/sink/RocketMQ.md
b/docs/zh/connectors/sink/RocketMQ.md
index db23ad0c08..dce3ed440a 100644
--- a/docs/zh/connectors/sink/RocketMQ.md
+++ b/docs/zh/connectors/sink/RocketMQ.md
@@ -2,13 +2,13 @@ import ChangeLog from '../changelog/connector-rocketmq.md';
# RocketMQ
-> RocketMQ sink 连接器
+> RocketMQ Sink 连接器
-## 支持Apache RocketMQ版本
+## 支持的 Apache RocketMQ 版本
-- 4.9.0 (或更新版本，供参考)
+- 4.9.0 或更新版本
-## 支持这些引擎
+## 支持的引擎
> Spark<br/>
> Flink<br/>
@@ -17,125 +17,127 @@ import ChangeLog from
'../changelog/connector-rocketmq.md';
## 主要特性
- [x] [精确一次](../../introduction/concepts/connector-v2-features.md)
-
-默认情况下，我们将使用2pc来保证消息精确一次到RocketMQ。
-- [ ] [timer flush](../../introduction/concepts/connector-v2-features.md)
+- [ ] [定时刷新](../../introduction/concepts/connector-v2-features.md)
## 描述
-将数据行写入Apache RocketMQ主题
+将 SeaTunnel 数据行写入 Apache RocketMQ topic。该 Sink 支持 JSON 和文本消息体、消息
tag、同步发送、按字段选择分区，以及在 `exactly.once = true` 时使用事务消息保证精确一次写入。
## Sink 参数
-| 名称 | 类型 | 是否必填 | 默认值 |
描述
|
-|----------------------|---------|----------|--------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| topic | string | 是 | - |
`RocketMQ topic` 名称.
|
-| name.srv.addr | string | 是 | - |
`RocketMQ`名称服务器集群地址。
|
-| acl.enabled | Boolean | 否 | false | 如果为
true，启用访问控制，需要配置 access key 和 secret key。
|
-| access.key | String | 否 | | 当
ACL_ENABLED 为 true 时，access key 不能为空。
|
-| secret.key | String | 否 | | 当
ACL_ENABLED 为 true 时，secret key 不能为空。
|
-| producer.group | String | 否 | SeaTunnel-Producer-Group |
RocketMQ 生产者组 ID。
|
-| tag | String | 否 | - |
`RocketMQ`消息标签。
|
-| partition.key.fields | array | 否 | - | -
|
-| format | String | 否 | json |
数据格式。默认格式为json。可选text格式。默认字段分隔符为“，”。如果自定义分隔符，请添加“field_delimiter”选项。 |
-| field.delimiter | String | 否 | , |
自定义数据格式的字段分隔符。
|
-| producer.send.sync | Boolean | 否 | false | 如果为
true, 则消息将同步发送。
|
-| exactly.once | Boolean | 否 | false | 如果为
true，将发送事务消息。
|
-| max.message.size | int | 否 | 4194304 |
允许的最大消息体大小（字节）。
|
-| send.message.timeout | int | 否 | 3000 |
发送消息的超时时间（毫秒）。
|
-| common-options | config | 否 | - |
Sink插件常用参数，请参考[sink common
options](../common-options/sink-common-options.md)了解详细信息。
|
-
-### partition.key.fields [array]
-
-配置哪些字段用作RocketMQ消息的键。
-例如，如果要使用上游数据中的字段值作为键，可以为此属性指定字段名。
-上游数据如下：
-
-| name | age | data |
-|------|-----|---------------|
-| Jack | 16 | data-example1 |
-| Mary | 23 | data-example2 |
-
-如果name被设置为主键，那么name列的哈希值将决定消息被发送到哪个分区。
+| 参数名 | 类型 | 是否必填 | 默认值 | 描述 |
+|--------|------|----------|--------|------|
+| topic | String | 是 | - | RocketMQ topic 名称。 |
+| name.srv.addr | String | 是 | - | RocketMQ NameServer 地址，例如 `localhost:9876`。
|
+| acl.enabled | Boolean | 否 | false | 是否启用 RocketMQ ACL 鉴权。 |
+| access.key | String | 否 | - | 访问密钥。`acl.enabled = true` 时必填。 |
+| secret.key | String | 否 | - | 秘密密钥。`acl.enabled = true` 时必填。 |
+| producer.group | String | 否 | SeaTunnel-Producer-Group | RocketMQ 生产者组 ID。 |
+| tag | String | 否 | - | 写入每条消息时使用的 RocketMQ tag。 |
+| partition.key.fields | List | 否 | - | 用来选择 RocketMQ 队列的字段名。配置的字段必须存在于上游
schema 中。 |
+| format | String | 否 | json | 消息格式。支持 `json` 和 `text`。 |
+| field.delimiter | String | 否 | `,` | `format = text` 时使用的字段分隔符。 |
+| producer.send.sync | Boolean | 否 | false | 是否同步发送消息。为 `false` 时异步发送。 |
+| exactly.once | Boolean | 否 | false | 是否使用事务消息实现精确一次写入。 |
+| max.message.size | int | 否 | 4194304 | 最大消息体大小，单位字节。 |
+| send.message.timeout | int | 否 | 3000 | 发送消息超时时间，单位毫秒。 |
+| common-options | config | 否 | - | Sink 连接器通用参数，详情请参考 [Sink
通用参数](../common-options/sink-common-options.md)。 |
-## 任务示例
+## 参数说明
+
+### partition.key.fields
+
+`partition.key.fields` 控制消息写入哪个 RocketMQ 队列。SeaTunnel
会对这些字段的值做哈希，并用哈希结果选择队列。如果不配置该参数，则由 RocketMQ 自行选择队列。
+
+例如，上游字段中有 `c_int` 时，可以这样配置：
+
+```hocon
+partition.key.fields = ["c_int"]
+```
+
+### exactly.once
-### Fake 到 RocketMQ 简单示例
+Sink 支持通过 RocketMQ 事务消息实现精确一次写入。该能力默认关闭。确认 RocketMQ 集群和作业 checkpoint
配置满足事务写入要求后，可设置 `exactly.once = true`。
+
+## 任务示例
->数据是随机生成的，并异步发送到测试主题
+### 写入 JSON 消息
```hocon
env {
parallelism = 1
+ job.mode = "BATCH"
}
source {
FakeSource {
+ row.num = 10
schema = {
fields {
- c_map = "map<string, string>"
- c_array = "array<int>"
c_string = string
- c_boolean = boolean
- c_tinyint = tinyint
- c_smallint = smallint
c_int = int
- c_bigint = bigint
- c_float = float
- c_double = double
- c_decimal = "decimal(30, 8)"
- c_bytes = bytes
- c_date = date
c_timestamp = timestamp
}
}
}
}
-transform {
- #如果你想了解更多关于如何配置seatunnel的信息，并查看转换插件的完整列表，
- #请前往https://seatunnel.apache.org/docs/transforms
-}
-
sink {
Rocketmq {
- name.srv.addr = "localhost:9876"
+ name.srv.addr = "rocketmq-e2e:9876"
topic = "test_topic"
+ partition.key.fields = ["c_int"]
+ producer.send.sync = true
}
}
-
```
-### Rocketmq 到 Rocketmq 简单示例
-
-> 使用RocketMQ时，会向c_int字段写入哈希数，该哈希数表示写入不同分区的分区数量。这是默认的异步写入方式
+### 写入文本消息
```hocon
env {
parallelism = 1
+ job.mode = "BATCH"
}
source {
+ FakeSource {
+ row.num = 10
+ schema = {
+ fields {
+ id = bigint
+ content = string
+ }
+ }
+ }
+}
+
+sink {
Rocketmq {
- name.srv.addr = "localhost:9876"
- topics = "test_topic"
- plugin_output = "rocketmq_table"
+ name.srv.addr = "rocketmq-e2e:9876"
+ topic = "test_text_topic"
+ format = text
+ field.delimiter = ","
+ producer.send.sync = true
+ }
+}
+```
+
+### 写入带 tag 的消息
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ FakeSource {
+ row.num = 10
schema = {
fields {
- c_map = "map<string, string>"
- c_array = "array<int>"
c_string = string
- c_boolean = boolean
- c_tinyint = tinyint
- c_smallint = smallint
c_int = int
- c_bigint = bigint
- c_float = float
- c_double = double
- c_decimal = "decimal(30, 8)"
- c_bytes = bytes
- c_date = date
- c_timestamp = timestamp
}
}
}
@@ -143,64 +145,48 @@ source {
sink {
Rocketmq {
- name.srv.addr = "localhost:9876"
- topic = "test_topic_sink"
- partition.key.fields = ["c_int"]
+ name.srv.addr = "rocketmq-e2e:9876"
+ topic = "test_topic_message_tag"
+ tag = "test_tag"
+ partition.key.fields = ["c_string"]
+ producer.send.sync = true
}
}
```
-### 时间戳消费写入示例
-
->这是流消费中特定的时间戳消费，当添加新分区时，程序将定期刷新感知和消费，并写入另一个主题类型
+### RocketMQ 读写 RocketMQ
```hocon
-
env {
parallelism = 1
job.mode = "STREAMING"
+ checkpoint.interval = 5000
}
source {
Rocketmq {
- name.srv.addr = "localhost:9876"
- topics = "test_topic"
+ name.srv.addr = "rocketmq-e2e:9876"
+ topics = "test_topic_source"
plugin_output = "rocketmq_table"
+ format = json
start.mode = "CONSUME_FROM_FIRST_OFFSET"
- batch.size = "400"
- consumer.group = "test_topic_group"
- format = "json"
+ consumer.group = "rocketmq_to_rocketmq_group"
schema = {
fields {
- c_map = "map<string, string>"
- c_array = "array<int>"
+ id = bigint
c_string = string
- c_boolean = boolean
- c_tinyint = tinyint
- c_smallint = smallint
- c_int = int
- c_bigint = bigint
- c_float = float
- c_double = double
- c_decimal = "decimal(30, 8)"
- c_bytes = bytes
- c_date = date
- c_timestamp = timestamp
}
}
}
}
-transform {
- #如果你想了解更多关于如何配置seatunnel的信息，并查看转换插件的完整列表，
- #请前往https://seatunnel.apache.org/docs/transforms
-}
sink {
Rocketmq {
- name.srv.addr = "localhost:9876"
- topic = "test_topic"
- partition.key.fields = ["c_int"]
- producer.send.sync = true
+ plugin_input = "rocketmq_table"
+ name.srv.addr = "rocketmq-e2e:9876"
+ topic = "test_topic_sink"
+ partition.key.fields = ["id"]
+ exactly.once = true
}
}
```
diff --git a/docs/zh/connectors/source/RocketMQ.md
b/docs/zh/connectors/source/RocketMQ.md
index 4bb0d158b6..097ff56266 100644
--- a/docs/zh/connectors/source/RocketMQ.md
+++ b/docs/zh/connectors/source/RocketMQ.md
@@ -6,72 +6,79 @@ import ChangeLog from '../changelog/connector-rocketmq.md';
## 支持的 Apache RocketMQ 版本
-- 4.9.0（或更新版本，供参考）
+- 4.9.0 或更新版本
-## 支持这些引擎
+## 支持的引擎
> Spark<br/>
> Flink<br/>
> SeaTunnel Zeta<br/>
-## 关键特性
+## 主要特性
-- [x] [批](../../introduction/concepts/connector-v2-features.md)
-- [x] [流](../../introduction/concepts/connector-v2-features.md)
+- [x] [批处理](../../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)
-- [x] [并行性](../../introduction/concepts/connector-v2-features.md)
-- [ ] [支持用户自定义split](../../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)
## 描述
-Apache RocketMQ 的源连接器。
-
-## 源选项
-
-| 参数名 | 类型 | 必须 | 默认值
| 描述
|
-|-------------------------------------|---------|----|----------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| topics | String | 否 | -
| RocketMQ 主题名称。如果有多个主题，使用 `,` 分隔，例如：`"tpc1,tpc2"`。`topics` 与
`tables_configs` 同时只能配置一个。
|
-| tables_configs | List | 否 | -
|
多表模式配置列表。每项配置一张表，支持：`topics`、`format`、`schema`、`tags`、`start.mode`、`start.mode.timestamp`、`start.mode.offsets`、`ignore_parse_errors`。`topics`
与 `tables_configs` 同时只能配置一个。 |
-| table_list | List | 否 | -
| 已废弃，请使用 `tables_configs` 代替。
|
-| name.srv.addr | String | 是 | -
| RocketMQ 名称服务器集群地址。
|
-| tags | String | 否 | -
| RocketMQ 标签名称。如果有多个标签，使用 `,` 分隔，例如：`"tag1,tag2"`。
|
-| acl.enabled | Boolean | 否 | false
| 如果为 true，启用访问控制，需要配置访问密钥和秘密密钥。
|
-| access.key | String | 否 |
| 访问密钥
|
-| secret.key | String | 否 |
| 当 ACL_ENABLED 为 true 时，秘密密钥不能为空。
|
-| batch.size | int | 否 | 100
| RocketMQ 消费者拉取批大小
|
-| consumer.group | String | 否 |
SeaTunnel-Consumer-Group | RocketMQ 消费者组 ID，用于区分不同的消费者组。
|
-| commit.on.checkpoint | Boolean | 否 | true
| 如果为 true，消费者的偏移量将在后台定期提交。
|
-| schema | | 否 | -
| 数据的结构，包括字段名称和字段类型。更多详情请参考 [Schema
特性](../../introduction/concepts/schema-feature.md)。
|
-| format | String | 否 | json
| 数据格式。默认格式是 json。可选 text 格式。默认字段分隔符是 ","。如果自定义分隔符，添加 "field.delimiter" 选项。
|
-| field.delimiter | String | 否 | ,
| 自定义数据格式的字段分隔符
|
-| start.mode | String | 否 |
CONSUME_FROM_GROUP_OFFSETS |
消费者的初始消费模式，有几种类型：[CONSUME_FROM_LAST_OFFSET],[CONSUME_FROM_FIRST_OFFSET],[CONSUME_FROM_GROUP_OFFSETS],[CONSUME_FROM_TIMESTAMP],[CONSUME_FROM_SPECIFIC_OFFSETS]
|
-| start.mode.offsets | | 否 |
| 消费模式为 "CONSUME_FROM_SPECIFIC_OFFSETS" 所需的偏移量
|
-| start.mode.timestamp | Long | 否 |
| 消费模式为 "CONSUME_FROM_TIMESTAMP" 所需的时间。
|
-| partition.discovery.interval.millis | long | 否 | -1
| 动态发现主题和分区的间隔。
|
-| ignore_parse_errors | Boolean | 否 | false
| 可选标志，跳过解析错误而不是失败。
|
-| consumer.poll.timeout.millis | long | 否 | 5000
| 拉取消息的超时时间（毫秒）。
|
-| common-options | config | 否 | -
| 源插件通用参数，请参考 [源通用选项](../common-options/source-common-options.md) 详见。
|
-
-### start.mode.offsets
-
-消费模式为 "CONSUME_FROM_SPECIFIC_OFFSETS" 所需的偏移量。
-
-例如：
+从 Apache RocketMQ topic 读取消息。连接器既可以用一套 schema 读取一个或多个 topic，也可以通过
`tables_configs` 读取多张不同结构的表。
+
+## 源参数
+
+| 参数名 | 类型 | 是否必填 | 默认值 | 描述 |
+|--------|------|----------|--------|------|
+| name.srv.addr | String | 是 | - | RocketMQ NameServer 地址，例如 `localhost:9876`。
|
+| 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 的消息。 |
+| acl.enabled | Boolean | 否 | false | 是否启用 RocketMQ ACL 鉴权。 |
+| access.key | String | 否 | - | 访问密钥。`acl.enabled = true` 时必填。 |
+| secret.key | String | 否 | - | 秘密密钥。`acl.enabled = true` 时必填。 |
+| batch.size | int | 否 | 100 | 每次最多拉取的消息数。 |
+| consumer.group | String | 否 | SeaTunnel-Consumer-Group | RocketMQ 消费者组 ID。 |
+| commit.on.checkpoint | Boolean | 否 | true | 是否在 SeaTunnel checkpoint
完成后提交消费位点。 |
+| schema | config | 否 | - | 消息结构。详情请参考 [Schema
特性](../../introduction/concepts/schema-feature.md)。不配置时，连接器按文本读取消息体。 |
+| format | String | 否 | json | 消息格式。支持 `json` 和 `text`。 |
+| field.delimiter | String | 否 | `,` | `format = text` 时使用的字段分隔符。 |
+| start.mode | String | 否 | CONSUME_FROM_GROUP_OFFSETS |
启动消费位置。支持：`CONSUME_FROM_LAST_OFFSET`、`CONSUME_FROM_FIRST_OFFSET`、`CONSUME_FROM_GROUP_OFFSETS`、`CONSUME_FROM_TIMESTAMP`、`CONSUME_FROM_SPECIFIC_OFFSETS`。
|
+| 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 | 是否跳过解析失败的消息。 |
+| consumer.poll.timeout.millis | long | 否 | 5000 | 拉取消息的超时时间，单位毫秒。 |
+| common-options | config | 否 | - | 源连接器通用参数，详情请参考
[源通用参数](../common-options/source-common-options.md)。 |
+
+## 参数说明
+
+### 启动消费位置
+
+`start.mode` 用来控制从哪里开始读：
+
+- `CONSUME_FROM_GROUP_OFFSETS`：从消费者组已提交的位点开始读。
+- `CONSUME_FROM_FIRST_OFFSET`：从最早可用位点开始读。
+- `CONSUME_FROM_LAST_OFFSET`：从最新位点开始读。
+- `CONSUME_FROM_TIMESTAMP`：从 `start.mode.timestamp` 对应时间之后的第一条消息开始读。
+- `CONSUME_FROM_SPECIFIC_OFFSETS`：从 `start.mode.offsets` 指定的位点开始读。
```hocon
+start.mode = "CONSUME_FROM_SPECIFIC_OFFSETS"
start.mode.offsets = {
- topic1-0 = 70
- topic1-1 = 10
- topic1-2 = 10
+ test_topic-0 = 50
}
```
-## 任务示例
+### 多表读取
-### 简单
+当不同 topic 的字段结构不一样时，使用 `tables_configs`。每一项都可以单独配置
`topics`、`schema`、`format`、`tags` 和启动消费位置。如果没有配置 `schema.table`，输出表名默认使用 topic
名称。
-> 消费者读取 Rocketmq 数据并将其打印到控制台
+## 任务示例
+
+### 读取 JSON 消息
```hocon
env {
@@ -84,42 +91,58 @@ source {
name.srv.addr = "rocketmq-e2e:9876"
topics = "test_topic_json"
plugin_output = "rocketmq_table"
+ format = json
schema = {
fields {
id = bigint
- c_map = "map<string, smallint>"
- c_array = "array<tinyint>"
c_string = string
- c_boolean = boolean
- c_tinyint = tinyint
- c_smallint = smallint
c_int = int
- c_bigint = bigint
- c_float = float
- c_double = double
- c_decimal = "decimal(2, 1)"
- c_bytes = bytes
- c_date = date
c_timestamp = timestamp
}
}
}
}
-transform {
- # 如果您想了解有关如何配置 seatunnel 的更多信息并查看完整的转换插件列表，
- # 请访问 https://seatunnel.apache.org/docs/transforms
+sink {
+ Console {
+ plugin_input = "rocketmq_table"
+ }
+}
+```
+
+### 按 tag 读取文本消息
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Rocketmq {
+ name.srv.addr = "rocketmq-e2e:9876"
+ topics = "test_topic_text"
+ plugin_output = "rocketmq_table"
+ format = text
+ field.delimiter = ","
+ tags = "tag_a,tag_b"
+ schema = {
+ fields {
+ id = bigint
+ content = string
+ }
+ }
+ }
}
sink {
Console {
+ plugin_input = "rocketmq_table"
}
}
```
-### 指定格式消费简单
-
-> 当我以 json 格式消费主题数据并解析，每次拉取的条数是 400，消费从原始位置开始
+### 从指定 offset 读取
```hocon
env {
@@ -129,48 +152,30 @@ env {
source {
Rocketmq {
- name.srv.addr = "localhost:9876"
- topics = "test_topic"
+ name.srv.addr = "rocketmq-e2e:9876"
+ topics = "test_topic_source"
plugin_output = "rocketmq_table"
- start.mode = "CONSUME_FROM_FIRST_OFFSET"
- batch.size = "400"
- consumer.group = "test_topic_group"
format = json
+ start.mode = "CONSUME_FROM_SPECIFIC_OFFSETS"
+ start.mode.offsets = {
+ test_topic_source-0 = 50
+ }
schema = {
fields {
- c_map = "map<string, string>"
- c_array = "array<int>"
- c_string = string
- c_boolean = boolean
- c_tinyint = tinyint
- c_smallint = smallint
- c_int = int
- c_bigint = bigint
- c_float = float
- c_double = double
- c_decimal = "decimal(30, 8)"
- c_bytes = bytes
- c_date = date
- c_timestamp = timestamp
+ id = bigint
}
}
}
}
-transform {
- # 如果您想了解有关如何配置 seatunnel 的更多信息并查看完整的转换插件列表，
- # 请访问 https://seatunnel.apache.org/docs/transforms
-}
-
sink {
Console {
+ plugin_input = "rocketmq_table"
}
}
```
-### 多表读取
-
-> 从不同 topic 读取不同结构的数据，使用 `tables_configs` 为每个 topic 独立配置。
+### 读取多个不同结构的 topic
```hocon
env {
@@ -180,28 +185,30 @@ env {
source {
Rocketmq {
- name.srv.addr = "localhost:9876"
- consumer.group = "multi_table_group"
- start.mode = "CONSUME_FROM_FIRST_OFFSET"
+ name.srv.addr = "rocketmq-e2e:9876"
+ start.mode = "CONSUME_FROM_LAST_OFFSET"
tables_configs = [
{
- topics = "topic_1"
- format = "json"
+ topics = "test_topic_multi_a"
+ start.mode = "CONSUME_FROM_FIRST_OFFSET"
+ format = json
schema = {
fields {
- id = int
- name = string
+ id = bigint
+ c_string = string
}
}
},
{
- topics = "topic_2"
- format = "json"
+ topics = "test_topic_multi_b"
+ start.mode = "CONSUME_FROM_FIRST_OFFSET"
+ tags = "tag_b"
+ format = json
schema = {
+ table = "rocketmq_multi_custom"
fields {
- id = int
+ id = bigint
description = string
- weight = double
}
}
}
