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 6cb263141c [Docs][Connector-V2] Improve RabbitMQ connector docs
(#11246)
6cb263141c is described below
commit 6cb263141cefa8d2ad1cbb7ef22c83727be4425f
Author: Daniel Carter <[email protected]>
AuthorDate: Fri Jul 3 08:12:09 2026 +0800
[Docs][Connector-V2] Improve RabbitMQ connector docs (#11246)
Co-authored-by: DanielCarter-stack
<[email protected]>
Co-authored-by: Shenghang <[email protected]>
---
docs/en/connectors/sink/Rabbitmq.md | 85 +++++++++++++--------
docs/en/connectors/source/Rabbitmq.md | 66 +++++++++++------
docs/zh/connectors/sink/Rabbitmq.md | 134 +++++++++++++++++++++++++++-------
docs/zh/connectors/source/Rabbitmq.md | 69 +++++++++++------
4 files changed, 254 insertions(+), 100 deletions(-)
diff --git a/docs/en/connectors/sink/Rabbitmq.md
b/docs/en/connectors/sink/Rabbitmq.md
index 7e21e6e583..1468a35488 100644
--- a/docs/en/connectors/sink/Rabbitmq.md
+++ b/docs/en/connectors/sink/Rabbitmq.md
@@ -19,14 +19,15 @@ Used to write data to Rabbitmq.
| host | string | yes | - |
| port | int | yes | - |
| virtual_host | string | yes | - |
-| username | string | yes | - |
-| password | string | yes | - |
+| username | string | no | - |
+| password | string | no | - |
| queue_name | string | yes | - |
| url | string | no | - |
+| routing_key | string | no | - |
+| exchange | string | no | - |
| network_recovery_interval | int | no | - |
| topology_recovery_enabled | boolean | no | - |
-| automatic_recovery_enabled | boolean | no | - |
-| use_correlation_id | boolean | no | false |
+| AUTOMATIC_RECOVERY_ENABLED | boolean | no | - |
| connection_timeout | int | no | - |
| rabbitmq.config | map | no | - |
| common-options | | no | - |
@@ -54,34 +55,23 @@ the AMQP user name to use when connecting to the broker
the password to use when connecting to the broker
+`username` and `password` should be configured together.
+
### url [string]
convenience method for setting the fields in an AMQP URI: host, port,
username, password and virtual host
### queue_name [string]
-the queue to write the message to
-
-### durable [boolean]
-
-true: The queue will survive a server restart.
-false: The queue will be deleted on server restart.
-
-### exclusive [boolean]
-
-true: The queue is used only by the current connection and will be deleted
when the connection closes.
-false: The queue can be used by multiple connections.
+the queue to write the message to. If `routing_key` is not configured, the
connector publishes messages to this queue through the default exchange.
-### auto_delete [boolean]
+### routing_key [string]
-true: The queue will be deleted automatically when the last consumer
unsubscribes.
-false: The queue will not be automatically deleted.
+The routing key used to publish messages. Configure it together with
`exchange` when you want to publish through a specific exchange instead of
directly to `queue_name`.
-### schema [Config]
+### exchange [string]
-#### fields [Config]
-
-the schema fields of upstream data.
+The exchange used when `routing_key` is configured.
### network_recovery_interval [int]
@@ -91,13 +81,11 @@ how long will automatic recovery wait before attempting to
reconnect, in ms
if true, enables topology recovery
-### automatic_recovery_enabled [boolean]
-
-if true, enables connection recovery
+### AUTOMATIC_RECOVERY_ENABLED [boolean]
-### use_correlation_id [boolean]
+If true, enables connection recovery.
-whether the messages received are supplied with a unique id to deduplicate
messages (in case of failed acknowledgments).
+The option key is currently uppercase in the connector configuration. Use
`AUTOMATIC_RECOVERY_ENABLED`, not `automatic_recovery_enabled`.
### connection_timeout [int]
@@ -121,17 +109,40 @@ Sink plugin common parameters, please refer to [Sink
Common Options](../common-o
- true: The queue is used only by the current connection and will be deleted
when the connection closes.
- false: The queue can be used by multiple connections.
-### auto-delete
+### auto_delete
- true: The queue will be deleted automatically when the last consumer
unsubscribes.
- false: The queue will not be automatically deleted.
+## Configuration Notes
+
+- If you configure `username`, you must also configure `password`, and vice
versa.
+- `host`, `port`, `virtual_host`, and `queue_name` are required connector
options. `url` can additionally provide the AMQP URI used by the RabbitMQ
client.
+- `durable`, `exclusive`, and `auto_delete` are used when the connector
declares the target queue.
+
## Example
simple:
```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+}
+
+source {
+ FakeSource {
+ row.num = 10
+ schema = {
+ fields {
+ id = bigint
+ c_string = string
+ }
+ }
+ }
+}
+
sink {
RabbitMQ {
host = "rabbitmq-e2e"
@@ -153,6 +164,23 @@ sink {
queue with durable, exclusive, auto_delete:
```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+}
+
+source {
+ FakeSource {
+ row.num = 10
+ schema = {
+ fields {
+ id = bigint
+ c_string = string
+ }
+ }
+ }
+}
+
sink {
RabbitMQ {
host = "rabbitmq-e2e"
@@ -175,4 +203,3 @@ sink {
## Changelog
<ChangeLog />
-
diff --git a/docs/en/connectors/source/Rabbitmq.md
b/docs/en/connectors/source/Rabbitmq.md
index 3201e2b69d..9d6ed7020b 100644
--- a/docs/en/connectors/source/Rabbitmq.md
+++ b/docs/en/connectors/source/Rabbitmq.md
@@ -29,9 +29,9 @@ The source must be non-parallel (parallelism set to 1) in
order to achieve exact
| -------------------------- | ------- | -------- | ------------- |
| host | string | yes | - |
| port | int | yes | - |
-| virtual_host | string | yes | - |
-| username | string | yes | - |
-| password | string | yes | - |
+| virtual_host | string | no | - |
+| username | string | no | - |
+| password | string | no | - |
| queue_name | string | no | - |
| schema | config | no | - |
| tables_configs | array | no | - |
@@ -40,13 +40,14 @@ The source must be non-parallel (parallelism set to 1) in
order to achieve exact
| exchange | string | no | - |
| network_recovery_interval | int | no | - |
| topology_recovery_enabled | boolean | no | - |
-| automatic_recovery_enabled | boolean | no | - |
+| AUTOMATIC_RECOVERY_ENABLED | boolean | no | - |
| connection_timeout | int | no | - |
| requested_channel_max | int | no | - |
| requested_frame_max | int | no | - |
| requested_heartbeat | int | no | - |
| prefetch_count | int | no | - |
-| delivery_timeout | long | no | - |
+| delivery_timeout | int | no | - |
+| use_correlation_id | boolean | no | - |
| common-options | | no | - |
| durable | boolean | no | true |
| exclusive | boolean | no | false |
@@ -72,6 +73,8 @@ the AMQP user name to use when connecting to the broker
the password to use when connecting to the broker
+`username` and `password` should be configured together.
+
### url [string]
convenience method for setting the fields in an AMQP URI: host, port,
username, password and virtual host
@@ -82,11 +85,11 @@ the queue to consume messages from. *Note: Required if
`tables_configs` is not c
### routing_key [string]
-the routing key to publish the message to
+Optional RabbitMQ routing key inherited from the shared RabbitMQ
configuration. It is not required for normal queue consumption.
### exchange [string]
-the exchange to publish the message to
+Optional RabbitMQ exchange inherited from the shared RabbitMQ configuration.
It is not required for normal queue consumption.
### schema [Config]
@@ -106,9 +109,11 @@ how long will automatic recovery wait before attempting to
reconnect, in ms
if true, enables topology recovery
-### automatic_recovery_enabled [boolean]
+### AUTOMATIC_RECOVERY_ENABLED [boolean]
+
+If true, enables connection recovery.
-if true, enables connection recovery
+The option key is currently uppercase in the connector configuration. Use
`AUTOMATIC_RECOVERY_ENABLED`, not `automatic_recovery_enabled`.
### connection_timeout [int]
@@ -132,10 +137,14 @@ Set the requested heartbeat timeout
prefetchCount the max number of messages to receive without acknowledgement
-### delivery_timeout [long]
+### delivery_timeout [int]
deliveryTimeout maximum wait time, in milliseconds, for the next message
delivery
+### use_correlation_id [boolean]
+
+Whether the consumed messages provide a unique correlation id that can be used
to deduplicate messages when acknowledgments fail.
+
### common options
Source plugin common parameters, please refer to [Source Common
Options](../common-options/source-common-options.md) for details
@@ -150,7 +159,7 @@ Source plugin common parameters, please refer to [Source
Common Options](../comm
- true: The queue is used only by the current connection and will be deleted
when the connection closes.
- false: The queue can be used by multiple connections.
-### auto-delete
+### auto_delete
- true: The queue will be deleted automatically when the last consumer
unsubscribes.
- false: The queue will not be automatically deleted.
@@ -163,12 +172,19 @@ If you are upgrading from a previous version that only
supported single-table re
- 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.
- Use `tables_configs` for multi-table mode.
- Use root-level `queue_name` and `schema` for single-table mode.
+- 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.
## Example
### Single-table Read Example
```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+}
+
source {
RabbitMQ {
host = "rabbitmq-e2e"
@@ -182,20 +198,32 @@ source {
id = bigint
c_map = "map<string, smallint>"
c_array = "array<tinyint>"
+ c_string = string
+ c_boolean = boolean
}
}
}
}
+
+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`.
```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+}
+
source {
RabbitMQ {
- host = "localhost"
+ host = "rabbitmq-e2e"
port = 5672
+ virtual_host = "/"
username = "guest"
password = "guest"
@@ -226,20 +254,14 @@ source {
}
sink {
- # The first sink will ONLY receive data from the users_queue
- Jdbc {
+ # The first sink will only receive data from users_table
+ Console {
plugin_input = "users_table"
- driver = "com.mysql.cj.jdbc.Driver"
- url = "jdbc:mysql://localhost:3306/mydb"
- query = "insert into users (user_id, name) values (?, ?)"
}
- # The second sink will ONLY receive data from the orders_queue
- Jdbc {
+ # The second sink will only receive data from orders_table
+ Console {
plugin_input = "orders_table"
- driver = "com.mysql.cj.jdbc.Driver"
- url = "jdbc:mysql://localhost:3306/mydb"
- query = "insert into orders (order_id, amount) values (?, ?)"
}
}
```
diff --git a/docs/zh/connectors/sink/Rabbitmq.md
b/docs/zh/connectors/sink/Rabbitmq.md
index 3a40dc4c1f..e621cb8468 100644
--- a/docs/zh/connectors/sink/Rabbitmq.md
+++ b/docs/zh/connectors/sink/Rabbitmq.md
@@ -11,26 +11,30 @@ import ChangeLog from '../changelog/connector-rabbitmq.md';
## 主要特性
- [ ] [精准一次](../../introduction/concepts/connector-v2-features.md)
-- [ ] [timer flush](../../introduction/concepts/connector-v2-features.md)
+- [ ] [定时刷新](../../introduction/concepts/connector-v2-features.md)
## 接收器选项
| 名称 | 类型 | 是否必须 | 默认值 |
|----------------------------|---------|------|-------|
-| host | string | yes | - |
-| port | int | yes | - |
-| virtual_host | string | yes | - |
-| username | string | yes | - |
-| password | string | yes | - |
-| queue_name | string | yes | - |
-| url | string | no | - |
-| network_recovery_interval | int | no | - |
-| topology_recovery_enabled | boolean | no | - |
-| automatic_recovery_enabled | boolean | no | - |
-| use_correlation_id | boolean | no | false |
-| connection_timeout | int | no | - |
-| rabbitmq.config | map | no | - |
-| common-options | | no | - |
+| host | string | 是 | - |
+| port | int | 是 | - |
+| virtual_host | string | 是 | - |
+| username | string | 否 | - |
+| password | string | 否 | - |
+| queue_name | string | 是 | - |
+| url | string | 否 | - |
+| routing_key | string | 否 | - |
+| exchange | string | 否 | - |
+| network_recovery_interval | int | 否 | - |
+| topology_recovery_enabled | boolean | 否 | - |
+| AUTOMATIC_RECOVERY_ENABLED | boolean | 否 | - |
+| connection_timeout | int | 否 | - |
+| rabbitmq.config | map | 否 | - |
+| durable | boolean | 否 | true |
+| exclusive | boolean | 否 | false |
+| auto_delete | boolean | 否 | false |
+| common-options | | 否 | - |
### host [string]
@@ -52,19 +56,38 @@ virtual host – 连接broker使用的vhost
连接broker时使用的密码
+`username` 和 `password` 需要一起配置。
+
### url [string]
设置host、port、username、password和virtual host的简便方式。
### queue_name [string]
-数据写入的队列名。
+数据写入的队列名。如果没有配置 `routing_key`，连接器会通过默认 exchange 将消息直接写入该队列。
+
+### routing_key [string]
+
+发布消息时使用的路由键。如果希望通过指定 exchange 发布消息，而不是直接写入 `queue_name`，请同时配置 `routing_key` 和
`exchange`。
+
+### exchange [string]
+
+配置 `routing_key` 时使用的 exchange。
-### schema [Config]
+### durable [boolean]
-#### fields [Config]
+- true：队列将在服务器重启时保留。
+- false：队列将在服务器重启时删除。
-上游数据的模式字段。
+### exclusive [boolean]
+
+- true：队列仅由当前连接使用，连接关闭时将删除。
+- false：队列可以由多个连接使用。
+
+### auto_delete [boolean]
+
+- true：队列将在最后一个消费者取消订阅时自动删除。
+- false：队列不会自动删除。
### network_recovery_interval [int]
@@ -74,13 +97,11 @@ virtual host – 连接broker使用的vhost
设置为true，表示启用拓扑恢复。
-### automatic_recovery_enabled [boolean]
+### AUTOMATIC_RECOVERY_ENABLED [boolean]
-设置为true，表示启用连接恢复。
+设置为 true，表示启用连接恢复。
-### use_correlation_id [boolean]
-
-接收到的消息是否都提供唯一ID，来删除重复的消息达到幂等（在失败的情况下）
+当前连接器配置项名称使用大写形式。请写成 `AUTOMATIC_RECOVERY_ENABLED`，不要写成
`automatic_recovery_enabled`。
### connection_timeout [int]
@@ -88,18 +109,78 @@ TCP连接建立的超时时间，单位为毫秒；0代表不限制。
### rabbitmq.config [map]
-In addition to the above parameters that must be specified by the RabbitMQ
client, the user can also specify multiple non-mandatory parameters for the
client, covering [all the parameters specified in the official RabbitMQ
document](https://www.rabbitmq.com/configure.html).
除了上面提及必须设置的RabbitMQ客户端参数，你也还可以为客户端指定多个非强制参数，参见
[RabbitMQ官方文档参数设置](https://www.rabbitmq.com/configure.html)。
### common options
Sink插件常用参数，请参考[Sink常用选项](../common-options/sink-common-options.md)获取更多细节信息。
+## 配置说明
+
+- 如果配置了 `username`，也必须配置 `password`，反过来也一样。
+- `host`、`port`、`virtual_host` 和 `queue_name` 是连接器必填项。`url` 可额外提供 RabbitMQ
客户端使用的 AMQP URI。
+- `durable`、`exclusive` 和 `auto_delete` 用于连接器声明目标队列。
+
## 示例
simple:
```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+}
+
+source {
+ FakeSource {
+ row.num = 10
+ schema = {
+ fields {
+ id = bigint
+ c_string = string
+ }
+ }
+ }
+}
+
+sink {
+ RabbitMQ {
+ host = "rabbitmq-e2e"
+ port = 5672
+ virtual_host = "/"
+ username = "guest"
+ password = "guest"
+ queue_name = "test1"
+ rabbitmq.config = {
+ requested-heartbeat = 10
+ connection-timeout = 10
+ }
+ }
+}
+```
+
+### 示例 2
+
+配置队列的 durable、exclusive、auto_delete：
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+}
+
+source {
+ FakeSource {
+ row.num = 10
+ schema = {
+ fields {
+ id = bigint
+ c_string = string
+ }
+ }
+ }
+}
+
sink {
RabbitMQ {
host = "rabbitmq-e2e"
@@ -108,6 +189,9 @@ sink {
username = "guest"
password = "guest"
queue_name = "test1"
+ durable = "true"
+ exclusive = "false"
+ auto_delete = "false"
rabbitmq.config = {
requested-heartbeat = 10
connection-timeout = 10
diff --git a/docs/zh/connectors/source/Rabbitmq.md
b/docs/zh/connectors/source/Rabbitmq.md
index cd5dc33d42..c483853212 100644
--- a/docs/zh/connectors/source/Rabbitmq.md
+++ b/docs/zh/connectors/source/Rabbitmq.md
@@ -29,24 +29,25 @@ import ChangeLog from '../changelog/connector-rabbitmq.md';
|----------------------------|---------|----|-------|-----------------------------------------------------------------------------|
| host | string | 是 | - | 连接的默认主机
|
| port | int | 是 | - | 连接的默认端口
|
-| virtual_host | string | 是 | - | 虚拟主机 – 连接到代理时使用的虚拟主机
|
-| username | string | 是 | - | 连接到代理时使用的 AMQP 用户名
|
-| password | string | 是 | - | 连接到代理时使用的密码
|
+| virtual_host | string | 否 | - | 虚拟主机 – 连接到代理时使用的虚拟主机
|
+| username | string | 否 | - | 连接到代理时使用的 AMQP 用户名
|
+| password | string | 否 | - | 连接到代理时使用的密码
|
| queue_name | string | 否 | - | 要消费消息的队列
|
| schema | config | 否 | - | 上游数据的模式。更多详情请参考 [Schema
特性](../../introduction/concepts/schema-feature.md)。 |
| tables_configs | array | 否 | - |
用于同时从多个队列读取消息。数组中的每个对象必须包含 queue_name 和 schema。 |
| url | string | 否 | - | 便捷方法，用于设置 AMQP URI
中的字段：主机、端口、用户名、密码和虚拟主机 |
-| routing_key | string | 否 | - | 要发布消息的路由密钥
|
-| exchange | string | 否 | - | 要发布消息的交换机
|
+| routing_key | string | 否 | - | RabbitMQ 共享配置中的可选路由键
|
+| exchange | string | 否 | - | RabbitMQ 共享配置中的可选
exchange |
| network_recovery_interval | int | 否 | - | 自动恢复在尝试重新连接之前等待多长时间（毫秒）
|
| topology_recovery_enabled | boolean | 否 | - | 如果为 true，启用拓扑恢复
|
-| automatic_recovery_enabled | boolean | 否 | - | 如果为 true，启用连接恢复
|
+| AUTOMATIC_RECOVERY_ENABLED | boolean | 否 | - | 如果为 true，启用连接恢复
|
| connection_timeout | int | 否 | - | 连接 tcp 建立超时（毫秒）；零表示无限
|
| requested_channel_max | int | 否 | - |
最初请求的最大通道数；零表示无限制。**注意：值必须在 0 到 65535 之间（AMQP 0-9-1 中的无符号短整数）。 |
| requested_frame_max | int | 否 | - | 请求的最大帧大小
|
| requested_heartbeat | int | 否 | - | 设置请求的心跳超时。**注意：值必须在 0 到
65535 之间（AMQP 0-9-1 中的无符号短整数）。 |
| prefetch_count | int | 否 | - | 预取计数，无需确认即可接收的最大消息数
|
-| delivery_timeout | long | 否 | - | 交付超时，等待下一条消息交付的最大时间（毫秒）
|
+| delivery_timeout | int | 否 | - | 交付超时，等待下一条消息交付的最大时间（毫秒）
|
+| use_correlation_id | boolean | 否 | - | 消息是否带有可用于去重的唯一
correlation id |
| durable | boolean | 否 | true | 队列是否在服务器重启时保留
|
| exclusive | boolean | 否 | false | 队列是否仅由当前连接使用
|
| auto_delete | boolean | 否 | false | 队列是否在最后一个消费者取消订阅时自动删除
|
@@ -72,6 +73,8 @@ import ChangeLog from '../changelog/connector-rabbitmq.md';
连接到代理时使用的密码
+`username` 和 `password` 需要一起配置。
+
### url [string]
便捷方法，用于设置 AMQP URI 中的字段：主机、端口、用户名、密码和虚拟主机
@@ -82,11 +85,11 @@ import ChangeLog from '../changelog/connector-rabbitmq.md';
### routing_key [string]
-要发布消息的路由密钥
+RabbitMQ 共享配置中的可选路由键。普通队列消费不需要配置它。
### exchange [string]
-要发布消息的交换机
+RabbitMQ 共享配置中的可选 exchange。普通队列消费不需要配置它。
### schema [Config]
@@ -106,9 +109,11 @@ import ChangeLog from '../changelog/connector-rabbitmq.md';
如果为 true，启用拓扑恢复
-### automatic_recovery_enabled [boolean]
+### AUTOMATIC_RECOVERY_ENABLED [boolean]
+
+如果为 true，启用连接恢复。
-如果为 true，启用连接恢复
+当前连接器配置项名称使用大写形式。请写成 `AUTOMATIC_RECOVERY_ENABLED`，不要写成
`automatic_recovery_enabled`。
### connection_timeout [int]
@@ -130,10 +135,14 @@ import ChangeLog from
'../changelog/connector-rabbitmq.md';
预取计数，无需确认即可接收的最大消息数
-### delivery_timeout [long]
+### delivery_timeout [int]
交付超时，等待下一条消息交付的最大时间（毫秒）
+### use_correlation_id [boolean]
+
+消费到的消息是否带有唯一 correlation id。开启后，当确认消息失败时，可用这个 id 辅助去重。
+
### common options
源插件通用参数，请参考 [源通用选项](../common-options/source-common-options.md) 详见。
@@ -148,7 +157,7 @@ import ChangeLog from '../changelog/connector-rabbitmq.md';
- true：队列仅由当前连接使用，连接关闭时将删除。
- false：队列可以由多个连接使用。
-### auto-delete
+### auto_delete
- true：队列将在最后一个消费者取消订阅时自动删除。
- false：队列不会自动删除。
@@ -161,12 +170,19 @@ import ChangeLog from
'../changelog/connector-rabbitmq.md';
- 您不能同时配置 `tables_configs` 和根级别的 `queue_name`/`schema`。它们是互斥的。这样做将导致配置验证错误。
- 使用 `tables_configs` 进行多表模式。
- 使用根级别的 `queue_name` 和 `schema` 进行单表模式。
+- 如果配置了 `username`，也必须配置 `password`，反过来也一样。
+- `host` 和 `port` 总是必填。`virtual_host` 是可选项，除非您的 RabbitMQ 环境要求使用非默认虚拟主机。
## 示例
### 单表读取示例
```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+}
+
source {
RabbitMQ {
host = "rabbitmq-e2e"
@@ -180,20 +196,31 @@ source {
id = bigint
c_map = "map<string, smallint>"
c_array = "array<tinyint>"
+ c_string = string
+ c_boolean = boolean
}
}
}
}
+sink {
+ Console {}
+}
```
### 多表读取示例
您可以使用 `tables_configs` 选项在一个作业中同时从多个 RabbitMQ
队列消费消息。连接器将根据消息来源的队列自动为每行数据分配正确的表标识符，允许您使用 `plugin_input` 将它们路由到不同的 sink。
```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+}
+
source {
RabbitMQ {
- host = "localhost"
+ host = "rabbitmq-e2e"
port = 5672
+ virtual_host = "/"
username = "guest"
password = "guest"
@@ -224,20 +251,14 @@ source {
}
sink {
- # 第一个 sink 将仅接收来自 users_queue 的数据
- Jdbc {
+ # 第一个 sink 将仅接收 users_table 的数据
+ Console {
plugin_input = "users_table"
- driver = "com.mysql.cj.jdbc.Driver"
- url = "jdbc:mysql://localhost:3306/mydb"
- query = "insert into users (user_id, name) values (?, ?)"
}
- # 第二个 sink 将仅接收来自 orders_queue 的数据
- Jdbc {
+ # 第二个 sink 将仅接收 orders_table 的数据
+ Console {
plugin_input = "orders_table"
- driver = "com.mysql.cj.jdbc.Driver"
- url = "jdbc:mysql://localhost:3306/mydb"
- query = "insert into orders (order_id, amount) values (?, ?)"
}
}
```
