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 3678ae3d60 [Docs][Connector-V2] Improve MongoDB CDC connector docs
(#11731)
3678ae3d60 is described below
commit 3678ae3d6095334cf69da4164394ce6f7c95abdd
Author: Daniel Carter <[email protected]>
AuthorDate: Mon Aug 10 21:51:51 2026 +0800
[Docs][Connector-V2] Improve MongoDB CDC connector docs (#11731)
Co-authored-by: DanielCarter-stack
<[email protected]>
---
docs/en/connectors/sink/Feishu.md | 98 +++++++++++++-
docs/en/connectors/sink/Fluss.md | 69 ++++++++++
docs/en/connectors/sink/GraphQL.md | 50 +++++++
docs/en/connectors/sink/Milvus.md | 123 ++++++++++++++++++
docs/en/connectors/sink/Prometheus.md | 100 ++++++++++++++
docs/en/connectors/source/Fluss.md | 62 +++++++++
docs/en/connectors/source/GraphQL.md | 91 +++++++++++++
docs/en/connectors/source/Milvus.md | 79 ++++++++++-
docs/en/connectors/source/MongoDB-CDC.md | 194 ++++++++++++++++++++++++++++
docs/en/connectors/source/MySQL-CDC.md | 84 ++++++++++++
docs/en/connectors/source/Oracle-CDC.md | 30 ++++-
docs/en/connectors/source/PostgreSQL-CDC.md | 86 ++++++++++++
docs/en/connectors/source/Prometheus.md | 71 ++++++++++
docs/zh/connectors/sink/Kafka.md | 43 ++++++
docs/zh/connectors/source/MongoDB-CDC.md | 191 +++++++++++++++++++++++++++
docs/zh/connectors/source/MySQL-CDC.md | 84 ++++++++++++
docs/zh/connectors/source/Oracle-CDC.md | 30 ++++-
docs/zh/connectors/source/PostgreSQL-CDC.md | 86 ++++++++++++
18 files changed, 1556 insertions(+), 15 deletions(-)
diff --git a/docs/en/connectors/sink/Feishu.md
b/docs/en/connectors/sink/Feishu.md
index 28df8f9f28..8c5ca6491a 100644
--- a/docs/en/connectors/sink/Feishu.md
+++ b/docs/en/connectors/sink/Feishu.md
@@ -47,7 +47,9 @@ Feishu webhook URLs and custom authentication headers are
sensitive. Do not prin
| DECIMAL | BigDecimal |
| BYTES | byte[] |
| STRING | String |
-| TIME<br/>TIMESTAMP<br/>TIME | String |
+| DATE | String |
+| TIME | String |
+| TIMESTAMP | String |
| ARRAY | JsonArray |
## Sink Options
@@ -86,7 +88,7 @@ env {
source {
FakeSource {
- row_num = 1
+ row.num = 1
schema = {
fields {
name = string
@@ -143,6 +145,98 @@ Feishu {
}
```
+### Stream alerts to a Feishu bot
+
+For continuous alerting, run the sink in streaming mode and use
`request_interval_ms`
+plus `array_mode = true` to batch multiple alerts into one webhook call. The
+example reads events from a Kafka topic and posts them as a single JSON array
per
+batch.
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 30000
+}
+
+source {
+ Kafka {
+ plugin_output = "alerts"
+ bootstrap.servers = "kafka:9092"
+ topic = "service_alerts"
+ format = "json"
+ schema = {
+ fields {
+ name = string
+ age = int
+ }
+ }
+ }
+}
+
+sink {
+ Feishu {
+ plugin_input = "alerts"
+ url = "https://open.feishu.cn/open-apis/bot/v2/hook/<your-hook-token>"
+ array_mode = true
+ batch_size = 20
+ request_interval_ms = 1000
+ retry = 5
+ retry_backoff_multiplier_ms = 200
+ retry_backoff_max_ms = 10000
+ }
+}
+```
+
+### Send a Feishu Rich Text Message
+
+Feishu bot webhooks expect a `msg_type` envelope such as `text`, `post`, or
+`interactive`. Use a `Transform` upstream to wrap each row before the Feishu
+sink sends it. The example below sends each row as a `text` envelope.
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ FakeSource {
+ plugin_output = "raw"
+ row.num = 1
+ schema = {
+ fields {
+ name = string
+ age = int
+ }
+ }
+ rows = [
+ {
+ fields = [tyrantlucifer, 12]
+ kind = INSERT
+ }
+ ]
+ }
+}
+
+transform {
+ Sql {
+ plugin_input = "raw"
+ query = "SELECT 'text' AS msg_type, named_struct('text', concat('User ',
name, ' is ', cast(age as string), ' years old')) AS content FROM raw"
+ }
+}
+
+sink {
+ Feishu {
+ plugin_input = "Sql"
+ url = "https://open.feishu.cn/open-apis/bot/v2/hook/<your-hook-token>"
+ headers {
+ Content-Type = "application/json"
+ }
+ }
+}
+```
+
## Changelog
<ChangeLog />
diff --git a/docs/en/connectors/sink/Fluss.md b/docs/en/connectors/sink/Fluss.md
index 30bb784626..fd977b3fc4 100644
--- a/docs/en/connectors/sink/Fluss.md
+++ b/docs/en/connectors/sink/Fluss.md
@@ -202,6 +202,75 @@ The sink resolves the target table separately for each
upstream table. Before ru
In multi-table mode, the target `database` and `table` values can combine
fixed text with placeholders. For example, `database =
"fluss_db_${database_name}"` and `table = "fluss_tb_${table_name}"` route
upstream table `test2.table1` to `fluss_db_test2.fluss_tb_table1`.
+### Stream upserts into a primary-key Fluss table
+
+When the target Fluss table has a primary key, the sink routes each row based
on
+its `RowKind`. The example below reads from a CDC source and writes inserts,
+updates, and deletes to a primary-key Fluss table.
+
+```hocon
+env {
+ parallelism = 2
+ job.mode = "STREAMING"
+ checkpoint.interval = 30000
+}
+
+source {
+ MySQL-CDC {
+ plugin_output = "orders_cdc"
+ base-url = "jdbc:mysql://mysql:3306/shop"
+ username = "cdc"
+ password = "${secret}"
+ database-names = ["shop"]
+ table-names = ["shop.orders"]
+ }
+}
+
+sink {
+ Fluss {
+ plugin_input = "orders_cdc"
+ bootstrap.servers = "fluss-coordinator:9123"
+ database = "shop"
+ table = "orders"
+ }
+}
+```
+
+### Replicate CDC changes across databases
+
+Use placeholders to replicate CDC changes from one Fluss namespace to another.
+The `schema_name` placeholder resolves to the upstream database name, and
+`table_name` to the upstream table name, so a single sink configuration can fan
+out many tables.
+
+```hocon
+env {
+ parallelism = 2
+ job.mode = "STREAMING"
+ checkpoint.interval = 30000
+}
+
+source {
+ Fluss {
+ plugin_output = "fluss_cdc"
+ bootstrap.servers = "fluss-coordinator:9123"
+ database = "source_db"
+ table = "source_table"
+ start_mode = "earliest"
+ }
+}
+
+sink {
+ Fluss {
+ plugin_input = "fluss_cdc"
+ bootstrap.servers = "fluss-coordinator:9123"
+ database = "sink_db_${schema_name}"
+ table = "sink_${table_name}"
+ multi_table_sink_replica = 2
+ }
+}
+```
+
## Changelog
<ChangeLog />
diff --git a/docs/en/connectors/sink/GraphQL.md
b/docs/en/connectors/sink/GraphQL.md
index c199a06f13..0b849f34bc 100644
--- a/docs/en/connectors/sink/GraphQL.md
+++ b/docs/en/connectors/sink/GraphQL.md
@@ -158,6 +158,56 @@ sink {
}
```
+### Stream Mutations to a GraphQL Service
+
+Run the same mutation continuously in streaming mode. The sink sends one
mutation
+per row and uses `retry` plus exponential backoff to absorb transient HTTP
+failures.
+
+```hocon
+env {
+ parallelism = 2
+ job.mode = "STREAMING"
+ checkpoint.interval = 30000
+}
+
+source {
+ FakeSource {
+ plugin_output = "events"
+ schema = {
+ fields {
+ id = int
+ val_string = string
+ }
+ }
+ rows = [
+ { kind = INSERT, fields = [1, "first"] }
+ { kind = INSERT, fields = [2, "second"] }
+ ]
+ }
+}
+
+sink {
+ GraphQL {
+ plugin_input = "events"
+ url = "http://graphql:8080/v1/graphql";
+ headers = {
+ Authorization = "Bearer ${secret}"
+ }
+ query = """
+ mutation MyMutation($id: Int!, $val_string: String!) {
+ insert_event(objects: {id: $id, val_string: $val_string}) {
+ affected_rows
+ }
+ }
+ """
+ retry = 5
+ retry_backoff_multiplier_ms = 200
+ retry_backoff_max_ms = 5000
+ }
+}
+```
+
### Write Multiple Upstream Tables
```hocon
diff --git a/docs/en/connectors/sink/Milvus.md
b/docs/en/connectors/sink/Milvus.md
index d75d438115..84f6698e11 100644
--- a/docs/en/connectors/sink/Milvus.md
+++ b/docs/en/connectors/sink/Milvus.md
@@ -228,6 +228,129 @@ sink {
}
```
+### Write With a Partition Key
+
+This example creates a new Milvus collection on the fly with `partition_key`
set to a
+SeaTunnel scalar field. SeaTunnel uses this field as the Milvus partition key
when it
+creates the collection, so each value of `category` becomes a separate
partition.
+
+```bash
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ FakeSource {
+ row.num = 10
+ vector.dimension = 4
+ schema = {
+ table = "simple_example_with_partitionkey"
+ columns = [
+ {
+ name = book_id
+ type = bigint
+ nullable = false
+ defaultValue = 0
+ comment = "primary key id"
+ },
+ {
+ name = book_intro
+ type = float_vector
+ columnScale = 4
+ comment = "vector"
+ },
+ {
+ name = category
+ type = string
+ nullable = false
+ comment = "partition key"
+ }
+ ]
+ primaryKey {
+ name = book_id
+ columnNames = [book_id]
+ }
+ }
+ }
+}
+
+sink {
+ Milvus {
+ url = "http://127.0.0.1:19530";
+ token = "username:password"
+ database = "test"
+ partition_key = "category"
+ }
+}
+```
+
+### Write With Nullable Fields
+
+This example enables nullable Milvus fields. It requires Milvus 2.5 or later.
The
+nullable scalar column is declared in the schema with `nullable = true` and the
+sink is configured with `enable_nullable_field = true` so that `null` values
can be
+written.
+
+```bash
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ FakeSource {
+ schema = {
+ table = "simple_example_nullable"
+ columns = [
+ {
+ name = book_id
+ type = bigint
+ nullable = false
+ comment = "primary key id"
+ },
+ {
+ name = book_intro
+ type = float_vector
+ columnScale = 4
+ nullable = false
+ comment = "vector"
+ },
+ {
+ name = book_title
+ type = string
+ nullable = true
+ comment = "nullable title"
+ }
+ ]
+ primaryKey {
+ name = book_id
+ columnNames = [book_id]
+ }
+ }
+ rows = [
+ {
+ kind = INSERT
+ fields = [1, [0.1, 0.2, 0.3, 0.4], null]
+ },
+ {
+ kind = INSERT
+ fields = [2, [0.2, 0.3, 0.4, 0.5], "nullable title"]
+ }
+ ]
+ }
+}
+
+sink {
+ Milvus {
+ url = "http://127.0.0.1:19530";
+ token = "username:password"
+ database = "test_nullable"
+ enable_nullable_field = true
+ }
+}
+```
+
### Streaming Write with Checkpoint Flush
```bash
diff --git a/docs/en/connectors/sink/Prometheus.md
b/docs/en/connectors/sink/Prometheus.md
index 9505528259..4bcf885b8d 100644
--- a/docs/en/connectors/sink/Prometheus.md
+++ b/docs/en/connectors/sink/Prometheus.md
@@ -138,6 +138,106 @@ sink {
}
```
+## Streaming Remote Write With Batched Flush
+
+This example reads from Kafka in streaming mode and writes to a Prometheus
remote
+write endpoint. The sink buffers up to `batch_size` rows or waits up to
+`flush_interval` milliseconds before issuing the HTTP write, which keeps
network
+overhead low when the upstream flow is bursty.
+
+```hocon
+env {
+ parallelism = 2
+ job.mode = "STREAMING"
+ checkpoint.interval = 30000
+}
+
+source {
+ Kafka {
+ plugin_output = "metrics_topic"
+ bootstrap.servers = "kafka:9092"
+ topic = "metrics"
+ format = "json"
+ schema = {
+ fields {
+ c_map = "map<string, string>"
+ c_double = double
+ c_timestamp = bigint
+ }
+ }
+ }
+}
+
+sink {
+ Prometheus {
+ plugin_input = "metrics_topic"
+ url = "http://prometheus:9090/api/v1/write";
+ key_label = "c_map"
+ key_value = "c_double"
+ key_timestamp = "c_timestamp"
+ batch_size = 2048
+ flush_interval = 10000
+ retry = 5
+ retry_backoff_multiplier_ms = 200
+ retry_backoff_max_ms = 10000
+ }
+}
+```
+
+## Multi-Table Remote Write
+
+When a single job reads from multiple upstream tables and writes to a single
+Prometheus remote write endpoint, set `multi_table_sink_replica` to control how
+many writer tasks each table gets. The default of `1` is fine when tables are
+small; raise it only when one table needs more parallelism than the others.
+
+```hocon
+env {
+ parallelism = 2
+ job.mode = "BATCH"
+}
+
+source {
+ FakeSource {
+ plugin_output = "fake_app_a"
+ schema = {
+ fields {
+ c_map = "map<string, string>"
+ c_double = double
+ c_timestamp = timestamp
+ }
+ }
+ rows = [
+ { kind = INSERT, fields = [{"__name__" : "app_a_metric"}, 1.0,
CURRENT_TIMESTAMP] }
+ ]
+ }
+ FakeSource {
+ plugin_output = "fake_app_b"
+ schema = {
+ fields {
+ c_map = "map<string, string>"
+ c_double = double
+ c_timestamp = timestamp
+ }
+ }
+ rows = [
+ { kind = INSERT, fields = [{"__name__" : "app_b_metric"}, 2.0,
CURRENT_TIMESTAMP] }
+ ]
+ }
+}
+
+sink {
+ Prometheus {
+ plugin_input = ["fake_app_a", "fake_app_b"]
+ url = "http://prometheus:9090/api/v1/write";
+ key_label = "c_map"
+ key_value = "c_double"
+ key_timestamp = "c_timestamp"
+ multi_table_sink_replica = 2
+ }
+}
+```
+
## Changelog
<ChangeLog />
diff --git a/docs/en/connectors/source/Fluss.md
b/docs/en/connectors/source/Fluss.md
index c7199df56b..5e421ca1a2 100644
--- a/docs/en/connectors/source/Fluss.md
+++ b/docs/en/connectors/source/Fluss.md
@@ -174,6 +174,68 @@ sink {
}
```
+### Stream one Fluss table into another
+
+This example copies a Fluss source table into a Fluss sink table in streaming
+mode. The source starts from the earliest available log offset and the
connector
+commits the per-bucket read position with each checkpoint so the job can resume
+after a restart.
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 30000
+}
+
+source {
+ Fluss {
+ bootstrap.servers = "fluss-coordinator:9123"
+ database = "fluss_stream_db"
+ table = "fluss_stream_src"
+ start_mode = "earliest"
+ poll.timeout.ms = 10000
+ plugin_output = "fluss_stream"
+ }
+}
+
+sink {
+ Fluss {
+ bootstrap.servers = "fluss-coordinator:9123"
+ database = "fluss_stream_db"
+ table = "fluss_stream_sink"
+ plugin_input = "fluss_stream"
+ }
+}
+```
+
+### Tune the poll timeout for high-latency clusters
+
+When the Fluss coordinator sits on a high-latency network or returns large
+batches, increase `poll.timeout.ms` so the log scanner waits longer between
+empty polls and reduces the number of round trips.
+
+```hocon
+env {
+ parallelism = 2
+ job.mode = "STREAMING"
+ checkpoint.interval = 60000
+}
+
+source {
+ Fluss {
+ bootstrap.servers = "fluss-coordinator:9123"
+ database = "fluss_db"
+ table = "fluss_table"
+ start_mode = "latest"
+ poll.timeout.ms = 60000
+ client.config = {
+ request.timeout = "30s"
+ }
+ }
+}
+```
+
## Changelog
<ChangeLog />
diff --git a/docs/en/connectors/source/GraphQL.md
b/docs/en/connectors/source/GraphQL.md
index 9d404739ee..a829ec54f7 100644
--- a/docs/en/connectors/source/GraphQL.md
+++ b/docs/en/connectors/source/GraphQL.md
@@ -116,6 +116,97 @@ sink {
}
```
+### Query GraphQL With Authentication Headers
+
+When the GraphQL endpoint requires authentication, pass the bearer token
through
+`headers`. Any HTTP header supported by the underlying client can be set this
way.
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ GraphQL {
+ plugin_output = "graphql_auth"
+ url = "https://graphql.example.com/v1/graphql";
+ format = "json"
+ content_field = "$.data.source"
+ headers = {
+ Authorization = "Bearer ${secret}"
+ X-Tenant = "acme"
+ }
+ query = """
+ query MyQuery {
+ source {
+ id
+ val_bool
+ }
+ }
+ """
+ schema = {
+ fields {
+ id = "int"
+ val_bool = "boolean"
+ }
+ }
+ }
+}
+
+sink {
+ Console {
+ plugin_input = "graphql_auth"
+ }
+}
+```
+
+### Streaming Polling Query
+
+For HTTP endpoints that publish new data over time but do not offer a
subscription,
+run the same query repeatedly in streaming mode. SeaTunnel sends the request
every
+`poll_interval_millis` and forwards the new rows downstream.
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 60000
+}
+
+source {
+ GraphQL {
+ plugin_output = "graphql_streaming"
+ url = "http://graphql:8080/v1/graphql";
+ format = "json"
+ content_field = "$.data.source"
+ poll_interval_millis = 10000
+ query = """
+ query MyQuery {
+ source {
+ id
+ val_bool
+ val_double
+ }
+ }
+ """
+ schema = {
+ fields {
+ id = "int"
+ val_bool = "boolean"
+ val_double = "double"
+ }
+ }
+ }
+}
+
+sink {
+ Console {
+ plugin_input = "graphql_streaming"
+ }
+}
+```
+
### Subscribe to GraphQL Data
```hocon
diff --git a/docs/en/connectors/source/Milvus.md
b/docs/en/connectors/source/Milvus.md
index 8b8b5d3325..85bb6e87df 100644
--- a/docs/en/connectors/source/Milvus.md
+++ b/docs/en/connectors/source/Milvus.md
@@ -48,22 +48,26 @@ Common use cases:
## Source Options
-| Name | Type | Required | Default | Description
|
-|------------|--------|----------|---------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| url | String | Yes | - | The URL to connect to Milvus or
Zilliz Cloud, for example `http://127.0.0.1:19530`.
|
-| 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. |
+| Name | Type | Required | Default | Description
|
+|-------------|---------|----------|------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| url | String | Yes | - | The URL to connect to Milvus
or Zilliz Cloud, for example `http://127.0.0.1:19530`.
|
+| 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. The legacy alias `collection_name` is also accepted.
|
+| batch_size | Integer | No | 1000 | Number of records to fetch
from Milvus in one batch. A larger value improves throughput but uses more
memory; set it to a smaller value when records contain large vector payloads.
|
+| rate_limit | Integer | No | 1000000 | Maximum number of records
the reader requests from Milvus per second. Use this to throttle a streaming
job against the Milvus quota (QPS) or gRPC message-size limit. Set to `-1` to
disable throttling. |
## 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.
+- `batch_size` controls the per-fetch page size, not the parallelism of
readers. Tune it together with `parallelism` to balance throughput and memory.
+- `rate_limit` is a server-side hint that protects against Milvus `GRPC limit`
errors when reading large volumes of vector data. Leave it at the default
unless you see rate-limit or gRPC errors in the logs.
- 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.
- When the source reads vector indexes, downstream Milvus sink can use that
metadata with `create_index = true` to create matching vector indexes.
+- Streaming jobs should set a checkpoint interval and reuse the same Milvus
token across restarts so that incremental reads resume correctly from the
committed offset.
## Task Example
@@ -163,6 +167,67 @@ sink {
}
```
+### Stream From One Collection With Checkpoints
+
+This example runs the source in `STREAMING` mode with a 30 second checkpoint
interval.
+The downstream sink uses `enable_upsert = false` so each row is inserted once
and
+duplicates are rejected.
+
+```bash
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 30000
+}
+
+source {
+ Milvus {
+ url = "http://127.0.0.1:19530";
+ token = "username:password"
+ database = "streaming_test"
+ collection = "simple_example"
+ batch_size = 500
+ rate_limit = 200000
+ }
+}
+
+sink {
+ Milvus {
+ url = "http://127.0.0.1:19530";
+ token = "username:password"
+ database = "streaming_test"
+ enable_upsert = false
+ batch_size = 1000
+ }
+}
+```
+
+### Throttle Reads Against a Shared Cluster
+
+When the Milvus cluster is shared with other jobs, lower `rate_limit` and
`batch_size`
+so the source does not exceed the cluster's gRPC message-size limit.
+
+```bash
+env {
+ parallelism = 2
+ job.mode = "BATCH"
+}
+
+source {
+ Milvus {
+ url = "http://127.0.0.1:19530";
+ token = "username:password"
+ database = "shared"
+ batch_size = 200
+ rate_limit = 100000
+ }
+}
+
+sink {
+ Console {}
+}
+```
+
## Changelog
<ChangeLog />
diff --git a/docs/en/connectors/source/MongoDB-CDC.md
b/docs/en/connectors/source/MongoDB-CDC.md
index 165cef3005..f3586311d8 100644
--- a/docs/en/connectors/source/MongoDB-CDC.md
+++ b/docs/en/connectors/source/MongoDB-CDC.md
@@ -286,6 +286,200 @@ sink {
}
```
+### CDC Data Write to Another MongoDB
+
+You can also route CDC events to a sink MongoDB collection. The example below
mirrors `inventory.products` from the source cluster to a target cluster named
`mongo1`:
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 5000
+}
+
+source {
+ MongoDB-CDC {
+ hosts = "mongo0:27017"
+ database = ["inventory"]
+ collection = ["inventory.products"]
+ schema = {
+ fields {
+ "_id" : string,
+ "name" : string,
+ "description" : string,
+ "weight" : string
+ }
+ }
+ }
+}
+
+sink {
+ MongoDB {
+ uri = "mongodb://mongo1:27017"
+ database = "inventory"
+ collection = "products_mirror"
+ }
+}
+```
+
+## Startup From a Specific Timestamp
+
+If you want to skip the snapshot and resume the change stream from a known
point in time, set `startup.mode` to `timestamp` and provide
`startup.timestamp` in epoch milliseconds. This is useful when re-processing a
backlog of changes after a maintenance window or when bootstrapping a new sink
that should ignore historical writes.
+
+```hocon
+source {
+ MongoDB-CDC {
+ hosts = "mongo0:27017"
+ database = ["inventory"]
+ collection = ["inventory.products"]
+ startup.mode = "timestamp"
+ # 2026-08-01 00:00:00 UTC
+ startup.timestamp = 1785542400000
+ schema = {
+ fields {
+ "_id" : string,
+ "name" : string,
+ "description" : string,
+ "weight" : string
+ }
+ }
+ }
+}
+```
+
+## Using the SRV Connection URI
+
+For MongoDB Atlas or any deployment that exposes a `mongodb+srv://` connection
string, pass the URI directly to the `hosts` option. Authentication
credentials, the replica set name, and other URI options are forwarded to the
driver as-is, so you do not need to also set `connection.options`:
+
+```hocon
+source {
+ MongoDB-CDC {
+ hosts = "mongodb+srv://cluster0.example.net"
+ username = "stuser"
+ password = "stpw"
+ database = ["inventory"]
+ collection = ["inventory.products"]
+ schema = {
+ fields {
+ "_id" : string,
+ "name" : string,
+ "description" : string,
+ "weight" : string
+ }
+ }
+ }
+}
+```
+
+## Heartbeat and Resume Token Maintenance
+
+Change stream resume tokens can expire if no source records are published for
a long time (for example, on a low-traffic collection). Set
`heartbeat.interval.ms` to a non-zero value so the connector periodically
advances the resume token and keeps the change stream open across checkpoint
restores:
+
+```hocon
+source {
+ MongoDB-CDC {
+ hosts = "mongo0:27017"
+ database = ["inventory"]
+ collection = ["inventory.products"]
+ # Send a heartbeat every 30 seconds when no records are flowing
+ heartbeat.interval.ms = 30000
+ schema = {
+ fields {
+ "_id" : string,
+ "name" : string,
+ "description" : string,
+ "weight" : string
+ }
+ }
+ }
+}
+```
+
+## Reading From Multiple MongoDB Sources
+
+A SeaTunnel job only accepts one source block per `source { ... }`, so the
supported way to
+fan in CDC streams from several MongoDB clusters is to submit one job per
source and have them
+write to the same sink table. Each job preserves its own parallelism, schema,
and restart
+tokens; the sink table deduplicates by primary key.
+
+```hocon
+# Job A: CDC from cluster `mongo0` writing to `inventory_a.products_a`
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 5000
+}
+
+source {
+ MongoDB-CDC {
+ hosts = "mongo0:27017"
+ database = ["inventory_a"]
+ collection = ["inventory_a.products_a"]
+ username = superuser
+ password = superpw
+ schema = {
+ fields {
+ "_id": string,
+ "name": string,
+ "price": int
+ }
+ }
+ }
+}
+
+sink {
+ jdbc {
+ url = "jdbc:mysql://mysql_e2e:3306/mongodb_cdc"
+ driver = "com.mysql.cj.jdbc.Driver"
+ username = "st_user"
+ password = "seatunnel"
+ generate_sink_sql = true
+ database = mongodb_cdc
+ table = "${table_name}"
+ primary_keys = ["_id"]
+ }
+}
+```
+
+```hocon
+# Job B: CDC from cluster `mongo1` writing to `inventory_b.products_b`
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 5000
+}
+
+source {
+ MongoDB-CDC {
+ hosts = "mongo1:27017"
+ database = ["inventory_b"]
+ collection = ["inventory_b.products_b"]
+ username = superuser
+ password = superpw
+ schema = {
+ fields {
+ "_id": string,
+ "name": string,
+ "price": int
+ }
+ }
+ }
+}
+
+sink {
+ jdbc {
+ url = "jdbc:mysql://mysql_e2e:3306/mongodb_cdc"
+ driver = "com.mysql.cj.jdbc.Driver"
+ username = "st_user"
+ password = "seatunnel"
+ generate_sink_sql = true
+ database = mongodb_cdc
+ table = "${table_name}"
+ primary_keys = ["_id"]
+ }
+}
+```
+
## Multi-table Synchronization
The following example demonstrates how to read CDC data from multiple MongoDB
collections and write each collection to the matching MySQL table. The sink
table uses `${table_name}`, so `inventory.products` and `inventory.orders` are
routed to their own target tables.
diff --git a/docs/en/connectors/source/MySQL-CDC.md
b/docs/en/connectors/source/MySQL-CDC.md
index 8ce4b5279a..266f7cc923 100644
--- a/docs/en/connectors/source/MySQL-CDC.md
+++ b/docs/en/connectors/source/MySQL-CDC.md
@@ -415,6 +415,90 @@ sink {
}
```
+### Configure Debezium heartbeat
+
+For low-traffic tables, the MySQL binlog only advances when row changes occur.
Use a Debezium heartbeat to keep the binlog position moving so downstream
checkpoints record fresh offsets and replication lag stays measurable. The
heartbeat table must exist on the MySQL server before the job starts.
+
+```hocon
+source {
+ MySQL-CDC {
+ username = "st_user_source"
+ password = "mysqlpw"
+ table-names = ["mysql_cdc.mysql_cdc_e2e_source_table"]
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc"
+ debezium {
+ heartbeat.interval.ms = 100
+ heartbeat.action.query = "INSERT INTO mysql_cdc.heartbeat (ts) VALUES
(NOW())"
+ }
+ }
+}
+```
+
+### Flush on a timer without waiting for `batch_size`
+
+When the source has very low write volume, the JDBC sink can sit idle until a
checkpoint fires. Enable timer flush in the sink so buffered rows are written
even if `batch_size` is not reached.
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 300000
+ sink.flush.interval = 500
+}
+
+source {
+ MySQL-CDC {
+ server-id = 5680-5690
+ username = "st_user_source"
+ password = "mysqlpw"
+ table-names = ["mysql_cdc.timer_flush_src"]
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc"
+ }
+}
+
+sink {
+ Jdbc {
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc"
+ driver = "com.mysql.cj.jdbc.Driver"
+ username = "st_user_sink"
+ password = "mysqlpw"
+ generate_sink_sql = true
+ database = mysql_cdc
+ table = timer_flush_sink
+ primary_keys = ["id"]
+ batch_size = 100000000
+ batch_interval_ms = 0
+ }
+}
+```
+
+`sink.flush.interval` is configured in the `env` block and applies to the sink
pipeline regardless of `batch_size`.
+
+### Read tables without a primary key
+
+For tables without a physical primary key, set `exactly_once = false` and
supply a unique column via `table-names-config.primaryKeys` when you need
stable row identity for downstream upserts.
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 5000
+}
+
+source {
+ MySQL-CDC {
+ server-id = 5652
+ username = "st_user_source"
+ password = "mysqlpw"
+ table-names = ["mysql_cdc.mysql_cdc_e2e_source_table_no_primary_key"]
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc"
+ exactly_once = false
+ }
+}
+```
+
+Without a usable primary key (configured or physical) the connector cannot
safely apply UPDATE/DELETE events. Use this mode only for append-only workloads
or when downstream sink behavior does not depend on row identity.
+
### Start From a Specific Binlog Offset
Use `startup.mode = "specific"` when the first record must be read from a
known binlog file and position.
diff --git a/docs/en/connectors/source/Oracle-CDC.md
b/docs/en/connectors/source/Oracle-CDC.md
index bf1607ecff..0200a496ed 100644
--- a/docs/en/connectors/source/Oracle-CDC.md
+++ b/docs/en/connectors/source/Oracle-CDC.md
@@ -423,12 +423,11 @@ source {
### Configure Debezium heartbeat
-For low-traffic tables, Debezium heartbeat can help the Oracle LogMiner
position move forward regularly.
+For low-traffic tables, the Oracle LogMiner SCN only advances when redo log
changes occur. Use a Debezium heartbeat to keep the SCN moving so checkpoint
offsets are recorded regularly and replication lag stays observable. The
heartbeat table must exist on the Oracle server before the job starts.
```hocon
source {
Oracle-CDC {
- plugin_output = "customers"
username = "system"
password = "top_secret"
database-names = ["ORCLCDB"]
@@ -437,13 +436,38 @@ source {
url = "jdbc:oracle:thin:@//oracle-host:1521/ORCLCDB"
debezium {
database.oracle.jdbc.timezoneAsRegion = "false"
- heartbeat.interval.ms = 1000
+ heartbeat.interval.ms = 100
heartbeat.action.query = "INSERT INTO DEBEZIUM.heartbeat (ts) VALUES
(SYSTIMESTAMP)"
}
}
}
```
+### Read tables without a primary key
+
+For tables without a physical primary key, set `exactly_once = false` and
supply a unique column via `table-names-config.primaryKeys` when you need
stable row identity for downstream upserts.
+
+```hocon
+source {
+ Oracle-CDC {
+ username = "system"
+ password = "top_secret"
+ database-names = ["ORCLCDB"]
+ schema-names = ["DEBEZIUM"]
+ url = "jdbc:oracle:thin:@//oracle-host:1521/ORCLCDB"
+ table-names = ["ORCLCDB.DEBEZIUM.FULL_TYPES_NO_PRIMARY_KEY"]
+ table-names-config = [
+ {
+ table = "ORCLCDB.DEBEZIUM.FULL_TYPES_NO_PRIMARY_KEY"
+ primaryKeys = ["ID"]
+ }
+ ]
+ }
+}
+```
+
+Without a usable primary key, the connector cannot safely apply UPDATE/DELETE
events. Use this mode only for append-only workloads.
+
### Schema change event filtering
When `schema-changes.enabled = true`, you can further control which schema
change event types are
diff --git a/docs/en/connectors/source/PostgreSQL-CDC.md
b/docs/en/connectors/source/PostgreSQL-CDC.md
index 7c1b5f54e4..5cd1fa8757 100644
--- a/docs/en/connectors/source/PostgreSQL-CDC.md
+++ b/docs/en/connectors/source/PostgreSQL-CDC.md
@@ -195,6 +195,92 @@ source {
}
```
+### Configure Debezium heartbeat
+
+For low-traffic tables, the Postgres logical decoding slot position only
advances when row changes are written to the WAL. A Debezium heartbeat keeps
the slot advancing so checkpoint offsets are recorded regularly and replication
lag stays observable. The heartbeat table must exist on the Postgres server
before the job starts.
+
+```hocon
+source {
+ Postgres-CDC {
+ username = "postgres"
+ password = "postgres"
+ database-names = ["postgres_cdc"]
+ schema-names = ["inventory"]
+ table-names = ["postgres_cdc.inventory.postgres_cdc_table_1"]
+ url =
"jdbc:postgresql://postgres_cdc_e2e:5432/postgres_cdc?loggerLevel=OFF"
+ decoding.plugin.name = "decoderbufs"
+ slot.name = "seatunnel_postgres_cdc"
+ debezium {
+ heartbeat.interval.ms = 100
+ heartbeat.action.query = "INSERT INTO inventory.heartbeat (ts) VALUES
(NOW())"
+ }
+ }
+}
+```
+
+### Run a snapshot-only batch
+
+Use `startup.mode = "snapshot-only"` when the job must perform an initial
snapshot and stop without entering WAL streaming. This is useful for one-time
backfills.
+
+```hocon
+env {
+ execution.parallelism = 1
+ job.mode = "BATCH"
+ checkpoint.interval = 5000
+}
+
+source {
+ Postgres-CDC {
+ username = "postgres"
+ password = "postgres"
+ database-names = ["postgres_cdc"]
+ schema-names = ["inventory"]
+ table-names = ["postgres_cdc.inventory.postgres_cdc_table_1"]
+ url =
"jdbc:postgresql://postgres_cdc_e2e:5432/postgres_cdc?loggerLevel=OFF"
+ decoding.plugin.name = "decoderbufs"
+ slot.name = "seatunnel_postgres_cdc"
+ startup.mode = "snapshot-only"
+ }
+}
+
+sink {
+ Jdbc {
+ url =
"jdbc:postgresql://postgres_cdc_e2e:5432/postgres_cdc?loggerLevel=OFF"
+ driver = "org.postgresql.Driver"
+ username = "postgres"
+ password = "postgres"
+ generate_sink_sql = true
+ database = postgres_cdc
+ table = inventory.sink_postgres_cdc_table_1
+ primary_keys = ["id"]
+ }
+}
+```
+
+In `snapshot-only` mode, the connector skips WAL streaming entirely; configure
`slot.name` if you need a dedicated slot for the snapshot read.
+
+### Read tables without a primary key
+
+For tables without a physical primary key, set `exactly_once = false` and
supply a unique column via `table-names-config.primaryKeys` when you need
stable row identity for downstream upserts.
+
+```hocon
+source {
+ Postgres-CDC {
+ username = "postgres"
+ password = "postgres"
+ database-names = ["postgres_cdc"]
+ schema-names = ["inventory"]
+ table-names = ["postgres_cdc.inventory.full_types_no_primary_key"]
+ url =
"jdbc:postgresql://postgres_cdc_e2e:5432/postgres_cdc?loggerLevel=OFF"
+ decoding.plugin.name = "decoderbufs"
+ exactly_once = false
+ slot.name = "seatunnel_postgres_cdc"
+ }
+}
+```
+
+Without a usable primary key, the connector cannot safely apply UPDATE/DELETE
events. Use this mode only for append-only workloads.
+
## CDC Metadata Fields
PostgreSQL CDC exposes metadata fields that can be used by the `Metadata`
transform:
diff --git a/docs/en/connectors/source/Prometheus.md
b/docs/en/connectors/source/Prometheus.md
index de815189b5..7eff2ceafd 100644
--- a/docs/en/connectors/source/Prometheus.md
+++ b/docs/en/connectors/source/Prometheus.md
@@ -158,6 +158,77 @@ source {
}
```
+## Streaming Range Query Example
+
+Run a range query continuously against a Prometheus or VictoriaMetrics server.
The
+source re-runs the same range query every `poll_interval_millis` and forwards
the
+newest samples to downstream operators.
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 60000
+}
+
+source {
+ Prometheus {
+ plugin_output = "live_metrics"
+ url = "http://prometheus:9090";
+ query = "rate(node_cpu_seconds_total{mode!=\"idle\"}[1m])"
+ query_type = "Range"
+ start = "2026-08-10T00:00:00Z"
+ end = "now"
+ step = "30s"
+ content_field = "$.data.result.*"
+ format = "json"
+ poll_interval_millis = 15000
+ retry = 3
+ retry_backoff_multiplier_ms = 200
+ retry_backoff_max_ms = 5000
+ schema = {
+ fields {
+ metric = "map<string, string>"
+ value = double
+ time = long
+ }
+ }
+ }
+}
+
+sink {
+ Console {}
+}
+```
+
+## Range Query With Explicit Unix Timestamps
+
+Both `start` and `end` accept Unix timestamps in seconds. This is useful when
you
+want to backfill a fixed window from a known point in time.
+
+```hocon
+source {
+ Prometheus {
+ plugin_output = "backfill"
+ url = "http://prometheus:9090";
+ query = "sum(up) by (job)"
+ query_type = "Range"
+ start = "1754851200"
+ end = "1754937600"
+ step = "60s"
+ content_field = "$.data.result.*"
+ format = "json"
+ schema = {
+ fields {
+ metric = "map<string, string>"
+ value = double
+ time = long
+ }
+ }
+ }
+}
+```
+
## Changelog
<ChangeLog />
diff --git a/docs/zh/connectors/sink/Kafka.md b/docs/zh/connectors/sink/Kafka.md
index d84e179b17..a7f22b9230 100644
--- a/docs/zh/connectors/sink/Kafka.md
+++ b/docs/zh/connectors/sink/Kafka.md
@@ -164,6 +164,49 @@ sink {
}
```
+### 使用 Kafka Headers
+
+本示例展示如何使用 `kafka_headers_fields` 将上游字段设置为 Kafka 消息头：
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ FakeSource {
+ parallelism = 1
+ plugin_output = "fake"
+ row.num = 16
+ schema = {
+ fields {
+ name = "string"
+ age = "int"
+ source = "string"
+ traceId = "string"
+ }
+ }
+ }
+}
+
+sink {
+ Kafka {
+ topic = "test_topic"
+ bootstrap.servers = "localhost:9092"
+ format = json
+ partition_key_fields = ["name"]
+ kafka_headers_fields = ["source", "traceId"]
+ semantics = EXACTLY_ONCE
+ kafka.config = {
+ acks = "all"
+ request.timeout.ms = 60000
+ buffer.memory = 33554432
+ }
+ }
+}
+```
+
### AWS MSK SASL/SCRAM
将以下 `${username}` 和 `${password}` 替换为 AWS MSK 中的配置值。
diff --git a/docs/zh/connectors/source/MongoDB-CDC.md
b/docs/zh/connectors/source/MongoDB-CDC.md
index f6e791e7e2..d1369d0acf 100644
--- a/docs/zh/connectors/source/MongoDB-CDC.md
+++ b/docs/zh/connectors/source/MongoDB-CDC.md
@@ -286,6 +286,197 @@ sink {
}
```
+### CDC数据写入另一个MongoDB
+
+您也可以将CDC事件路由到另一个MongoDB集合作为sink。下面的示例将源端 `mongo0` 的 `inventory.products`
镜像写入目标集群 `mongo1`：
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 5000
+}
+
+source {
+ MongoDB-CDC {
+ hosts = "mongo0:27017"
+ database = ["inventory"]
+ collection = ["inventory.products"]
+ schema = {
+ fields {
+ "_id" : string,
+ "name" : string,
+ "description" : string,
+ "weight" : string
+ }
+ }
+ }
+}
+
+sink {
+ MongoDB {
+ uri = "mongodb://mongo1:27017"
+ database = "inventory"
+ collection = "products_mirror"
+ }
+}
+```
+
+## 从指定时间戳启动
+
+如果希望跳过快照，从某个已知的时间点继续读取change stream，可以将 `startup.mode` 设置为 `timestamp`，并通过
`startup.timestamp` 提供epoch毫秒时间戳。这在维护窗口之后重放积压变更、或为新的sink初始化时跳过历史写入时非常有用：
+
+```hocon
+source {
+ MongoDB-CDC {
+ hosts = "mongo0:27017"
+ database = ["inventory"]
+ collection = ["inventory.products"]
+ startup.mode = "timestamp"
+ # 2026-08-01 00:00:00 UTC
+ startup.timestamp = 1785542400000
+ schema = {
+ fields {
+ "_id" : string,
+ "name" : string,
+ "description" : string,
+ "weight" : string
+ }
+ }
+ }
+}
+```
+
+## 使用SRV连接URI
+
+对于MongoDB Atlas或任何暴露 `mongodb+srv://` 连接字符串的部署，可以将URI直接传给 `hosts`
选项。URI中携带的认证信息、副本集名称等参数会原样传递给驱动，无需再额外配置 `connection.options`：
+
+```hocon
+source {
+ MongoDB-CDC {
+ hosts = "mongodb+srv://cluster0.example.net"
+ username = "stuser"
+ password = "stpw"
+ database = ["inventory"]
+ collection = ["inventory.products"]
+ schema = {
+ fields {
+ "_id" : string,
+ "name" : string,
+ "description" : string,
+ "weight" : string
+ }
+ }
+ }
+}
+```
+
+## 心跳与Resume Token维护
+
+如果某个集合长时间没有数据写入，change stream的resume token可能会过期（例如低流量集合）。将
`heartbeat.interval.ms` 设置为非零值，可以让连接器在没有数据时定期推进resume
token，从而在checkpoint恢复后保持change stream有效：
+
+```hocon
+source {
+ MongoDB-CDC {
+ hosts = "mongo0:27017"
+ database = ["inventory"]
+ collection = ["inventory.products"]
+ # 当没有数据写入时，每30秒发送一次心跳
+ heartbeat.interval.ms = 30000
+ schema = {
+ fields {
+ "_id" : string,
+ "name" : string,
+ "description" : string,
+ "weight" : string
+ }
+ }
+ }
+}
+```
+
+## 从多个MongoDB源读取
+
+单个 SeaTunnel 任务在 `source { ... }` 内只支持一个 source 块，因此同时从多个 MongoDB 集群拉取 CDC
流的可行方式是每个源提交一个独立任务，并写入同一个 sink 表。每个任务保留各自的并行度、schema 和重启 token，最终由 sink 表按主键去重。
+
+```hocon
+# 任务 A：从集群 mongo0 读取 CDC，写入 inventory_a.products_a
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 5000
+}
+
+source {
+ MongoDB-CDC {
+ hosts = "mongo0:27017"
+ database = ["inventory_a"]
+ collection = ["inventory_a.products_a"]
+ username = superuser
+ password = superpw
+ schema = {
+ fields {
+ "_id": string,
+ "name": string,
+ "price": int
+ }
+ }
+ }
+}
+
+sink {
+ jdbc {
+ url = "jdbc:mysql://mysql_e2e:3306/mongodb_cdc"
+ driver = "com.mysql.cj.jdbc.Driver"
+ username = "st_user"
+ password = "seatunnel"
+ generate_sink_sql = true
+ database = mongodb_cdc
+ table = "${table_name}"
+ primary_keys = ["_id"]
+ }
+}
+```
+
+```hocon
+# 任务 B：从集群 mongo1 读取 CDC，写入 inventory_b.products_b
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 5000
+}
+
+source {
+ MongoDB-CDC {
+ hosts = "mongo1:27017"
+ database = ["inventory_b"]
+ collection = ["inventory_b.products_b"]
+ username = superuser
+ password = superpw
+ schema = {
+ fields {
+ "_id": string,
+ "name": string,
+ "price": int
+ }
+ }
+ }
+}
+
+sink {
+ jdbc {
+ url = "jdbc:mysql://mysql_e2e:3306/mongodb_cdc"
+ driver = "com.mysql.cj.jdbc.Driver"
+ username = "st_user"
+ password = "seatunnel"
+ generate_sink_sql = true
+ database = mongodb_cdc
+ table = "${table_name}"
+ primary_keys = ["_id"]
+ }
+}
+```
+
## 多表同步
以下示例演示了如何读取多个 MongoDB 集合的 CDC 数据，并将每个集合写入对应的 MySQL 表。Sink 表名使用
`${table_name}`，因此 `inventory.products` 和 `inventory.orders` 会分别写入自己的目标表。
diff --git a/docs/zh/connectors/source/MySQL-CDC.md
b/docs/zh/connectors/source/MySQL-CDC.md
index 21f71cf990..c50393dfbe 100644
--- a/docs/zh/connectors/source/MySQL-CDC.md
+++ b/docs/zh/connectors/source/MySQL-CDC.md
@@ -412,6 +412,90 @@ sink {
}
```
+### 配置 Debezium 心跳
+
+对于低流量表，binlog 只有在发生行变更时才会推进。使用 Debezium 心跳让 binlog 位置持续向前滚动，便于下游 checkpoint
记录到新鲜偏移，并让复制延迟可观测。心跳表必须提前在 MySQL 服务端创建。
+
+```hocon
+source {
+ MySQL-CDC {
+ username = "st_user_source"
+ password = "mysqlpw"
+ table-names = ["mysql_cdc.mysql_cdc_e2e_source_table"]
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc"
+ debezium {
+ heartbeat.interval.ms = 100
+ heartbeat.action.query = "INSERT INTO mysql_cdc.heartbeat (ts) VALUES
(NOW())"
+ }
+ }
+}
+```
+
+### 定时刷新，无需等待 `batch_size`
+
+当 source 写入量非常低时，JDBC sink 可能在 checkpoint 触发前一直处于空闲状态。在 sink 中开启定时刷新，即使没有达到
`batch_size`，缓冲的数据也会被写入。
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 300000
+ sink.flush.interval = 500
+}
+
+source {
+ MySQL-CDC {
+ server-id = 5680-5690
+ username = "st_user_source"
+ password = "mysqlpw"
+ table-names = ["mysql_cdc.timer_flush_src"]
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc"
+ }
+}
+
+sink {
+ Jdbc {
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc"
+ driver = "com.mysql.cj.jdbc.Driver"
+ username = "st_user_sink"
+ password = "mysqlpw"
+ generate_sink_sql = true
+ database = mysql_cdc
+ table = timer_flush_sink
+ primary_keys = ["id"]
+ batch_size = 100000000
+ batch_interval_ms = 0
+ }
+}
+```
+
+`sink.flush.interval` 配置在 `env` 块中，无论 `batch_size` 是否达到，都会作用于 sink 流水线。
+
+### 读取没有主键的表
+
+对于没有物理主键的表，将 `exactly_once` 设为 `false`，并通过 `table-names-config.primaryKeys`
提供一列作为下游 upsert 所需的稳定行标识。
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 5000
+}
+
+source {
+ MySQL-CDC {
+ server-id = 5652
+ username = "st_user_source"
+ password = "mysqlpw"
+ table-names = ["mysql_cdc.mysql_cdc_e2e_source_table_no_primary_key"]
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc"
+ exactly_once = false
+ }
+}
+```
+
+如果没有可用的主键（无论配置的或物理的），connector 就无法安全地应用 UPDATE/DELETE
事件。仅在仅追加（append-only）场景或下游 sink 行为不依赖行标识时使用此模式。
+
### 从指定 Binlog 位置启动
当需要从明确的 binlog 文件和位置开始读取时，可以使用 `startup.mode = "specific"`。
diff --git a/docs/zh/connectors/source/Oracle-CDC.md
b/docs/zh/connectors/source/Oracle-CDC.md
index c39b83c354..4d8646a084 100644
--- a/docs/zh/connectors/source/Oracle-CDC.md
+++ b/docs/zh/connectors/source/Oracle-CDC.md
@@ -422,12 +422,11 @@ source {
### 配置 Debezium 心跳
-对于变更较少的表，可以配置 Debezium 心跳，让 Oracle LogMiner 位点定期向前推进。
+对于变更较少的表，Oracle LogMiner 的 SCN 只有在发生 redo log 变更时才会推进。使用 Debezium 心跳让 SCN
持续向前滚动，便于 checkpoint 定期记录偏移，并让复制延迟可观测。心跳表必须提前在 Oracle 服务端创建。
```hocon
source {
Oracle-CDC {
- plugin_output = "customers"
username = "system"
password = "top_secret"
database-names = ["ORCLCDB"]
@@ -436,13 +435,38 @@ source {
url = "jdbc:oracle:thin:@//oracle-host:1521/ORCLCDB"
debezium {
database.oracle.jdbc.timezoneAsRegion = "false"
- heartbeat.interval.ms = 1000
+ heartbeat.interval.ms = 100
heartbeat.action.query = "INSERT INTO DEBEZIUM.heartbeat (ts) VALUES
(SYSTIMESTAMP)"
}
}
}
```
+### 读取没有主键的表
+
+对于没有物理主键的表，将 `exactly_once` 设为 `false`，并通过 `table-names-config.primaryKeys`
提供一列作为下游 upsert 所需的稳定行标识。
+
+```hocon
+source {
+ Oracle-CDC {
+ username = "system"
+ password = "top_secret"
+ database-names = ["ORCLCDB"]
+ schema-names = ["DEBEZIUM"]
+ url = "jdbc:oracle:thin:@//oracle-host:1521/ORCLCDB"
+ table-names = ["ORCLCDB.DEBEZIUM.FULL_TYPES_NO_PRIMARY_KEY"]
+ table-names-config = [
+ {
+ table = "ORCLCDB.DEBEZIUM.FULL_TYPES_NO_PRIMARY_KEY"
+ primaryKeys = ["ID"]
+ }
+ ]
+ }
+}
+```
+
+没有可用的主键时，connector 无法安全地应用 UPDATE/DELETE 事件。仅在仅追加（append-only）场景下使用此模式。
+
### Schema change 事件过滤
当 `schema-changes.enabled = true` 时，可通过 `schema-changes.include` /
`schema-changes.exclude` 进一步
diff --git a/docs/zh/connectors/source/PostgreSQL-CDC.md
b/docs/zh/connectors/source/PostgreSQL-CDC.md
index de1eb73efb..602e0e060c 100644
--- a/docs/zh/connectors/source/PostgreSQL-CDC.md
+++ b/docs/zh/connectors/source/PostgreSQL-CDC.md
@@ -193,6 +193,92 @@ source {
}
```
+### 配置 Debezium 心跳
+
+对于低流量表，Postgres 逻辑解码槽的位置只有在 WAL 中发生行变更时才会推进。使用 Debezium 心跳让槽位持续推进，便于
checkpoint 定期记录偏移，并让复制延迟可观测。心跳表必须提前在 Postgres 服务端创建。
+
+```hocon
+source {
+ Postgres-CDC {
+ username = "postgres"
+ password = "postgres"
+ database-names = ["postgres_cdc"]
+ schema-names = ["inventory"]
+ table-names = ["postgres_cdc.inventory.postgres_cdc_table_1"]
+ url =
"jdbc:postgresql://postgres_cdc_e2e:5432/postgres_cdc?loggerLevel=OFF"
+ decoding.plugin.name = "decoderbufs"
+ slot.name = "seatunnel_postgres_cdc"
+ debezium {
+ heartbeat.interval.ms = 100
+ heartbeat.action.query = "INSERT INTO inventory.heartbeat (ts) VALUES
(NOW())"
+ }
+ }
+}
+```
+
+### 仅运行一次性快照
+
+当任务只需要执行初始快照并停止（不进入 WAL 流式读取）时，使用 `startup.mode =
"snapshot-only"`。该模式适合一次性数据回填。
+
+```hocon
+env {
+ execution.parallelism = 1
+ job.mode = "BATCH"
+ checkpoint.interval = 5000
+}
+
+source {
+ Postgres-CDC {
+ username = "postgres"
+ password = "postgres"
+ database-names = ["postgres_cdc"]
+ schema-names = ["inventory"]
+ table-names = ["postgres_cdc.inventory.postgres_cdc_table_1"]
+ url =
"jdbc:postgresql://postgres_cdc_e2e:5432/postgres_cdc?loggerLevel=OFF"
+ decoding.plugin.name = "decoderbufs"
+ slot.name = "seatunnel_postgres_cdc"
+ startup.mode = "snapshot-only"
+ }
+}
+
+sink {
+ Jdbc {
+ url =
"jdbc:postgresql://postgres_cdc_e2e:5432/postgres_cdc?loggerLevel=OFF"
+ driver = "org.postgresql.Driver"
+ username = "postgres"
+ password = "postgres"
+ generate_sink_sql = true
+ database = postgres_cdc
+ table = inventory.sink_postgres_cdc_table_1
+ primary_keys = ["id"]
+ }
+}
+```
+
+`snapshot-only` 模式下，connector 完全跳过 WAL 流式读取；如果快照读取需要独立的复制槽，请配置 `slot.name`。
+
+### 读取没有主键的表
+
+对于没有物理主键的表，将 `exactly_once` 设为 `false`，并通过 `table-names-config.primaryKeys`
提供一列作为下游 upsert 所需的稳定行标识。
+
+```hocon
+source {
+ Postgres-CDC {
+ username = "postgres"
+ password = "postgres"
+ database-names = ["postgres_cdc"]
+ schema-names = ["inventory"]
+ table-names = ["postgres_cdc.inventory.full_types_no_primary_key"]
+ url =
"jdbc:postgresql://postgres_cdc_e2e:5432/postgres_cdc?loggerLevel=OFF"
+ decoding.plugin.name = "decoderbufs"
+ exactly_once = false
+ slot.name = "seatunnel_postgres_cdc"
+ }
+}
+```
+
+没有可用的主键时，connector 无法安全地应用 UPDATE/DELETE 事件。仅在仅追加（append-only）场景下使用此模式。
+
## CDC 元数据字段
PostgreSQL CDC 会提供以下元数据字段，可配合 `Metadata` 转换使用：
