[Fix][Connector-V2] Close JDBC source read transactions after each split

docs/en/connectors/sink/Assert.md

@@ -17,8 +17,11 @@ Assert is a sink connector used to validate pipeline output. It checks row count
 ## Key Features
 
 - [ ] [exactly-once](../../introduction/concepts/connector-v2-features.md)
-- [ ] [timer flush](../../introduction/concepts/connector-v2-features.md)
+- [ ] [cdc](../../introduction/concepts/connector-v2-features.md)
+- [x] [batch](../../introduction/concepts/connector-v2-features.md)
+- [x] [stream](../../introduction/concepts/connector-v2-features.md)
 - [x] [support multiple table write](../../introduction/concepts/connector-v2-features.md)
+- [ ] [timer flush](../../introduction/concepts/connector-v2-features.md)
 
 ## Options
 
@@ -135,6 +138,16 @@ Sink plugin common parameters, please refer to [Sink Common Options](../common-o
 - `tables_configs` is used for multi-table jobs. The `table_path` value must match the table path carried by the upstream source.
 - `equals_to` compares the actual field value with the configured expected value. For complex values such as array, map, and row, use the same HOCON value shape as the source data.
 
+:::tip
+
+The Assert sink is a terminal sink — it has no external system to write to. Use it to validate intermediate results without needing a downstream database. The connector does not interpret `UPDATE` or `DELETE` row kinds as CDC operations; every received row is asserted against the configured rules. If the configured row range, field value, or catalog metadata check fails, the job fails with the matching error message.
+
+:::
+
+## Streaming Validation
+
+Assert works in both `BATCH` and `STREAMING` job modes. Field rules (`NOT_NULL`, `MIN_LENGTH`, `MAX_LENGTH`, etc.) are checked on every row as it arrives at the sink writer. Row count rules (`MIN_ROW` / `MAX_ROW`) are evaluated **exactly once** when the sink writer closes (at job shutdown, savepoint, or failure), against the cumulative row count observed by that writer instance since it was created — not per checkpoint window, and not reset between checkpoints. If you need per-checkpoint-window row-count validation, that requires a source code change (out of scope for a docs update).
+
 ## Example
 
 ### Simple
@@ -631,6 +644,67 @@ sink {
 
 ```
 
+### Stream Validation With Checkpoint Window
+
+The example below shows a streaming job that ends with the row count satisfying the cumulative `MIN_ROW` / `MAX_ROW` window (`50 ≤ total rows ≤ 5000`). The check runs once at writer close against the cumulative count, not per checkpoint window.
+
+```hocon
+env {
+  parallelism = 1
+  job.mode = "STREAMING"
+  checkpoint.interval = 60000
+}
+
+source {
+  FakeSource {
+    row.num = 1000
+    schema {
+      fields {
+        name = string
+        age = int
+      }
+    }
+    plugin_output = "stream_data"
+  }
+}
+
+sink {
+  Assert {
+    plugin_input = "stream_data"
+    rules =
+      {
+        row_rules = [
+          {
+            rule_type = MIN_ROW
+            rule_value = 50
+          },
+          {
+            rule_type = MAX_ROW
+            rule_value = 5000
+          }
+        ],
+        field_rules = [{
+          field_name = age
+          field_type = int
+          field_value = [
+            {
+              rule_type = NOT_NULL
+            },
+            {
+              rule_type = MIN
+              rule_value = 0
+            },
+            {
+              rule_type = MAX
+              rule_value = 150
+            }
+          ]
+        }]
+      }
+  }
+}
+```
+
 ## Changelog
 
 <ChangeLog />

docs/en/connectors/sink/GoogleBigtable.md

@@ -4,15 +4,21 @@ import ChangeLog from '../changelog/connector-google-bigtable.md';
 
 > Google Bigtable sink connector
 
+## Support Those Engines
+
+> SeaTunnel Zeta<br/>
+
 ## Description
 
 Writes data to Google Cloud Bigtable using the native Bigtable Data v2 Java client.
 
-## Key features
+## Key Features
 
 - [ ] [exactly-once](../../introduction/concepts/connector-v2-features.md)
 - [x] [batch](../../introduction/concepts/connector-v2-features.md)
+- [ ] [cdc](../../introduction/concepts/connector-v2-features.md)
 - [x] [support multiple table write](../../introduction/concepts/connector-v2-features.md)
+- [ ] [timer flush](../../introduction/concepts/connector-v2-features.md)
 
 ## Options
 
@@ -30,7 +36,7 @@ Writes data to Google Cloud Bigtable using the native Bigtable Data v2 Java clie
 | batch_mutation_size | int     | no       | 100           |
 | schema_save_mode    | enum    | no       | RECREATE_SCHEMA |
 | data_save_mode      | enum    | no       | APPEND_DATA   |
-| multi_table_sink_replica | int | no       | -             |
+| multi_table_sink_replica | int | no       | 1             |
 | common-options      |         | no       | -             |
 
 ### project_id [string]
@@ -43,13 +49,13 @@ Bigtable instance ID. Example: `"my-bigtable-instance"`
 
 ### table [string]
 
-The Bigtable table name to write to. Example: `"my-table"`
+The Bigtable table name to write to. Example: `"my-table"`. The connector does not create the Bigtable table; create it (with all required column families) before running the job.
 
 ### rowkey_column [list]
 
 Column names used to compose the Bigtable row key. Example: `["id"]` or `["tenant", "id"]`.
 
-When multiple columns are specified they are joined with `rowkey_delimiter`.
+When multiple columns are specified they are joined with `rowkey_delimiter`. With a single row-key column, a null or empty value fails the job with `WRITE_FAILED`. With multiple row-key columns, a null value in any non-last column silently becomes an empty segment in the composed row key (joined by `rowkey_delimiter`); only when the entire composed key is empty does the job fail.
 
 ### column_family [config]
 
@@ -70,6 +76,8 @@ column_family {
 }
 ```
 
+Field names that do not appear in the map fall back to the `all_columns` family, or to the default family `cf` if `all_columns` is not configured.
+
 ### credentials_path [string]
 
 Path to the Google Cloud service account JSON key file.
@@ -109,13 +117,13 @@ Data save mode. Only `APPEND_DATA` is supported now.
 
 ### multi_table_sink_replica [int]
 
-The number of sink replicas used for multi-table writing. For details, see [Sink Common Options](../common-options/sink-common-options.md).
+The number of sink replicas used for multi-table writing. For details, see [Sink Common Options](../common-options/sink-common-options.md). `multi_table_sink_replica` increases the number of parallel writer replicas within a single sink instance; the target Bigtable table is fixed by the `table` option and is not derived per upstream table.
 
 ### common options
 
 Sink plugin common parameters, please refer to [Sink Common Options](../common-options/sink-common-options.md) for details.
 
-## Data Types
+## Data Type Mapping
 
 All SeaTunnel types are supported:
 
@@ -137,15 +145,20 @@ All SeaTunnel types are supported:
 
 :::tip
 
-Bigtable does not have relational columns. The sink writes every non-row-key field as a Bigtable cell. The target column family is selected by `column_family`; the Bigtable qualifier is the SeaTunnel field name.
+Bigtable does not have relational columns. The sink writes every non-row-key field as a Bigtable cell. The target column family is selected by `column_family`; the Bigtable qualifier is the SeaTunnel field name. The sink treats every upstream row as an unconditional cell mutation, so `UPDATE` / `DELETE` row kinds are not interpreted as CDC operations and overwrite the previous cell under the same `(row key, column family, qualifier)` triple.
 
 :::
 
-## Example
+## Task Example
 
 ### Basic — Application Default Credentials
 
 ```hocon
+env {
+  parallelism = 1
+  job.mode = "BATCH"
+}
+
 sink {
   GoogleBigtable {
     project_id  = "my-gcp-project"
@@ -162,6 +175,11 @@ sink {
 ### Service Account Key File
 
 ```hocon
+env {
+  parallelism = 1
+  job.mode = "BATCH"
+}
+
 sink {
   GoogleBigtable {
     project_id       = "my-gcp-project"
@@ -181,6 +199,11 @@ sink {
 ### Multiple Column Families
 
 ```hocon
+env {
+  parallelism = 1
+  job.mode = "BATCH"
+}
+
 sink {
   GoogleBigtable {
     project_id  = "my-gcp-project"
@@ -200,6 +223,11 @@ sink {
 ### Use a version column and empty null values
 
 ```hocon
+env {
+  parallelism = 1
+  job.mode = "BATCH"
+}
+
 sink {
   GoogleBigtable {
     project_id       = "my-gcp-project"
@@ -217,6 +245,52 @@ sink {
 }
 ```
 
+### Streaming write with checkpoint flush
+
+In streaming mode, the writer flushes the in-memory mutation buffer at every checkpoint. The current `batch_mutation_size` still controls the in-task buffer; checkpoint frequency only affects how often already buffered mutations are sent to Bigtable.
+
+```hocon
+env {
+  parallelism = 2
+  job.mode = "STREAMING"
+  checkpoint.interval = 30000
+}
+
+source {
+  FakeSource {
+    row.num = 1000
+    schema {
+      fields {
+        tenant_id  = string
+        event_id   = string
+        event_ts   = bigint
+        event_type = string
+        payload    = string
+      }
+    }
+    plugin_output = "events_stream"
+  }
+}
+
+sink {
+  GoogleBigtable {
+    plugin_input = "events_stream"
+    project_id   = "my-gcp-project"
+    instance_id  = "my-bigtable-instance"
+    table        = "events"
+    credentials_path = "/secrets/sa-key.json"
+    rowkey_column    = ["tenant_id", "event_id"]
+    rowkey_delimiter = "#"
+    version_column   = "event_ts"
+    column_family {
+      all_columns = "data"
+      event_type  = "meta"
+    }
+    batch_mutation_size = 200
+  }
+}
+```
+
 ## Changelog
 
 <ChangeLog />

docs/en/connectors/sink/GoogleFirestore.md

@@ -25,8 +25,9 @@ indexes in Google Cloud before running queries that need them.
 - [ ] [exactly-once](../../introduction/concepts/connector-v2-features.md)
 - [ ] [cdc](../../introduction/concepts/connector-v2-features.md)
 - [x] [batch](../../introduction/concepts/connector-v2-features.md)
-- [ ] [stream](../../introduction/concepts/connector-v2-features.md)
+- [x] [stream](../../introduction/concepts/connector-v2-features.md)
 - [ ] [support multiple table write](../../introduction/concepts/connector-v2-features.md)
+- [ ] [timer flush](../../introduction/concepts/connector-v2-features.md)
 
 ## Supported DataSource Info
 
@@ -99,14 +100,16 @@ Sink plugin common parameters, please refer to [Sink Common Options](../common-o
 ## Notes
 
 - The connector currently provides a sink only. There is no GoogleFirestore source connector.
-- Each sink block writes to one configured collection. It does not switch collections automatically for multi-table input.
+- Each sink block writes to one configured collection. It does not switch collections automatically for multi-table input; use one sink block per Firestore collection.
 - Firestore document IDs are generated automatically. Use another connector or transform before this sink if you need deterministic document IDs.
-- The sink does not interpret `UPDATE` or `DELETE` row kinds as CDC operations.
+- The sink does not interpret `UPDATE` or `DELETE` row kinds as CDC operations — every row triggers a Firestore `add` call that produces a new document.
 - Do not put raw service account JSON directly in `credentials`; encode it with Base64 first.
-- Field names in the upstream SeaTunnel schema become Firestore document field
-  names.
+- Field names in the upstream SeaTunnel schema become Firestore document field names.
+- The connector works in both `BATCH` and `STREAMING` job modes. In the current implementation `FirestoreSinkWriter.write()` calls the Firestore client's `add(...)` once per row and does not buffer or batch rows, so there is no in-memory write buffer to flush at checkpoint boundaries; checkpoint completion does not imply that all previously written rows have reached Firestore.
 
-## Example
+## Task Example
+
+### Batch write of typed rows
 
 ```hocon
 env {
@@ -153,6 +156,39 @@ sink {
 }
 ```
 
+### Streaming write with checkpoint interval
+
+```hocon
+env {
+  parallelism = 1
+  job.mode = "STREAMING"
+  checkpoint.interval = 30000
+}
+
+source {
+  FakeSource {
+    row.num = 100
+    schema = {
+      fields {
+        c_string = string
+        c_int = int
+        c_timestamp = timestamp
+      }
+    }
+    plugin_output = "firestore_stream"
+  }
+}
+
+sink {
+  GoogleFirestore {
+    plugin_input = "firestore_stream"
+    project_id = "my-gcp-project"
+    collection = "events"
+    credentials = "base64-service-account-json"
+  }
+}
+```
+
 ## Changelog
 
 <ChangeLog />