Connector couchbase migration

docs/en/connectors/sink/Couchbase.md

@@ -81,7 +81,6 @@ Couchbase stores JSON documents. The connector maps SeaTunnel types to JSON valu
 | primary-key           | `List<String>` | No       | -          | Field names used to build the document key (length-prefixed encoding: `<len>:<value>` components separated by `#`). A random UUID is used when not set. |
 | upsert-enable         | Boolean       | No       | `false`    | Enable upsert (insert-or-replace) mode. When `false`, duplicate keys will cause an error. |
 | buffer-flush.max-rows | Integer       | No       | `1000`     | Maximum rows to buffer before a batch write is triggered. Use `-1` to disable. |
-| buffer-flush.interval | Long          | No       | `30000`    | Maximum milliseconds between batch writes. Use `-1` to disable. |
 | retry.max             | Integer       | No       | `3`        | Maximum retry attempts on transient write failure. |
 | retry.interval        | Long          | No       | `1000`     | Base milliseconds for linear retry delay. Attempt `n` waits `retry.interval × n` ms. |
 
@@ -161,11 +160,31 @@ sink {
     primary-key            = ["user_id", "order_id"]
     upsert-enable          = true
     buffer-flush.max-rows  = 500
-    buffer-flush.interval  = 10000
     retry.max              = 5
     retry.interval         = 2000
   }
 }
 ```
 
+### Timer Flush
+
+The sink can flush its buffer on a timer so that buffered rows are written even when the upstream  
+flow is idle and fewer than `buffer-flush.max-rows` rows have been buffered. This timer is driven  
+by the engine, not by the connector, and is currently supported only by **SeaTunnel Zeta**.
+
+Enable it by setting `sink.flush.interval` (milliseconds) in the job `env` block:
+
+```hocon  
+env {  
+sink.flush.interval = 10000  
+}  
+```
+
+> On Spark and Flink there is no sub-checkpoint timer flush: `sink.flush.interval` is a Zeta engine  
+> primitive, and the Spark/Flink sink writer context does not implement it. On those engines the  
+> buffer is flushed when it reaches `buffer-flush.max-rows`, on checkpoint (`CouchbaseWriter`  
+> flushes in `prepareCommit()`), and when the writer is closed. For lower latency between  
+> checkpoints on Spark or Flink, tune `buffer-flush.max-rows` accordingly.
+
+
 <ChangeLog />

docs/zh/connectors/sink/Couchbase.md

@@ -76,7 +76,6 @@ sh bin/install-plugin.sh ${version}
 | primary-key            | `List<String>`  | 否       | -          | 用于构建文档键的字段名列表（长度前缀编码：`<长度>:<值>` 分量以 `#` 分隔）。未设置时使用随机 UUID。 |
 | upsert-enable          | Boolean        | 否       | `false`    | 是否启用 Upsert（插入或替换）模式。为 `false` 时，重复键将报错。 |
 | buffer-flush.max-rows  | Integer        | 否       | `1000`     | 触发批量写入的最大缓冲行数。设为 `-1` 禁用。 |
-| buffer-flush.interval  | Long           | 否       | `30000`    | 批量写入之间的最大间隔（毫秒）。设为 `-1` 禁用。 |
 | retry.max              | Integer        | 否       | `3`        | 写入失败时的最大重试次数。 |
 | retry.interval         | Long           | 否       | `1000`     | 线性退避基础间隔（毫秒）。第 n 次重试等待 `retry.interval × n` 毫秒。 |
 
@@ -154,11 +153,24 @@ sink {
     primary-key            = ["user_id", "order_id"]
     upsert-enable          = true
     buffer-flush.max-rows  = 500
-    buffer-flush.interval  = 10000
     retry.max              = 5
     retry.interval         = 2000
   }
 }
 ```
 
+### 定时刷新
+
+该连接器支持在空闲时按计时器刷新缓冲区，即使尚未达到 `buffer-flush.max-rows`，也能定期发送已缓冲的记录。  
+此定时器由**引擎**驱动，而非连接器本身，目前仅 **SeaTunnel Zeta** 支持。
+
+在 Job 的 `env` 块中设置 `sink.flush.interval`（毫秒）即可启用：
+
+```hocon  
+env {  
+  sink.flush.interval = 10000  
+}
+
+```
+
 <ChangeLog />

seatunnel-connectors-v2/connector-couchbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/couchbase/config/CouchbaseSinkOptions.java

@@ -38,19 +38,6 @@ public class CouchbaseSinkOptions extends CouchbaseConfig {
                             "The maximum number of buffered rows per batch write request."
                                     + " Use -1 to disable size-based flushing.");
 
-    /**
-     * Maximum time (ms) between two consecutive batch writes.
-     *
-     * <p>A value of {@code -1} disables interval-based flushing.
-     */
-    public static final Option<Long> BUFFER_FLUSH_INTERVAL =
-            Options.key("buffer-flush.interval")
-                    .longType()
-                    .defaultValue(30000L)
-                    .withDescription(
-                            "The maximum interval between batch write requests, in milliseconds."
-                                    + " Use -1 to disable interval-based flushing.");
-
     /** Number of retry attempts on transient write failures before giving up. */
     public static final Option<Integer> RETRY_MAX =
             Options.key("retry.max")

seatunnel-connectors-v2/connector-couchbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/couchbase/sink/CouchbaseSinkFactory.java

@@ -56,7 +56,6 @@ public OptionRule optionRule() {
                 .optional(
                         CouchbaseSinkOptions.SCOPE,
                         CouchbaseSinkOptions.BUFFER_FLUSH_MAX_ROWS,
-                        CouchbaseSinkOptions.BUFFER_FLUSH_INTERVAL,
                         CouchbaseSinkOptions.RETRY_MAX,
                         CouchbaseSinkOptions.RETRY_INTERVAL,
                         CouchbaseSinkOptions.UPSERT_ENABLE,
@@ -94,8 +93,6 @@ public TableSink createSink(TableSinkFactoryContext context) {
 
         config.getOptional(CouchbaseSinkOptions.BUFFER_FLUSH_MAX_ROWS)
                 .ifPresent(builder::withFlushSize);
-        config.getOptional(CouchbaseSinkOptions.BUFFER_FLUSH_INTERVAL)
-                .ifPresent(builder::withBatchIntervalMs);
         config.getOptional(CouchbaseSinkOptions.RETRY_MAX).ifPresent(builder::withRetryMax);
         config.getOptional(CouchbaseSinkOptions.RETRY_INTERVAL)
                 .ifPresent(builder::withRetryInterval);

seatunnel-connectors-v2/connector-couchbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/couchbase/sink/CouchbaseWriter.java

@@ -46,11 +46,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.UUID;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
 
 /**
  * Writes {@link SeaTunnelRow} records to a Couchbase collection.
@@ -59,8 +55,6 @@
  *
  * <ul>
  *   <li>The buffer reaches {@code buffer-flush.max-rows}, or
- *   <li>A periodic background timer fires every {@code buffer-flush.interval} milliseconds (the
- *       real max-latency guarantee, enforced even when no new rows arrive), or
  *   <li>A checkpoint or shutdown is triggered.
  * </ul>
  *
@@ -80,11 +74,6 @@
  * pre-assigned ids — no new UUIDs are generated on a second pass, so silent duplicate documents and
  * spurious {@link com.couchbase.client.core.error.DocumentExistsException} collisions on
  * already-committed rows are both eliminated.
- *
- * <p><b>Async-error propagation:</b> when the background timer flush fails the first exception is
- * latched in {@link #asyncFlushError}. Subsequent calls to {@link #write}, {@link #prepareCommit},
- * and {@link #close} check the latch and rethrow the failure so the task cannot continue silently
- * after a broken flush cycle.
  */
 @Slf4j
 public class CouchbaseWriter implements SinkWriter<SeaTunnelRow, Void, Void> {
@@ -99,25 +88,6 @@ public class CouchbaseWriter implements SinkWriter<SeaTunnelRow, Void, Void> {
     private final int maxRetries;
     private final long retryIntervalMs;
 
-    /**
-     * Background scheduler that fires a periodic flush at every {@code batchIntervalMs}. This is
-     * the mechanism that enforces the {@code buffer-flush.interval} contract even when no new rows
-     * arrive (low-throughput / idle streaming jobs). {@code null} when the interval is disabled
-     * ({@code batchIntervalMs == -1}).
-     */
-    private final ScheduledExecutorService flushScheduler;
-
-    private final ScheduledFuture<?> flushFuture;
-
-    /**
-     * Latches the first exception thrown by the background timer flush. Checked and rethrown by
-     * {@link #write}, {@link #prepareCommit}, and {@link #close} so that the task lifecycle methods
-     * surface the failure instead of silently continuing after a broken flush.
-     */
-    private final AtomicReference<Throwable> asyncFlushError = new AtomicReference<>();
-
-    // TODO: Reserve context for future parallelism/metrics use.
-    @SuppressWarnings("unused")
     private final SinkWriter.Context context;
 
     public CouchbaseWriter(
@@ -162,39 +132,16 @@ public CouchbaseWriter(
         this.cluster = connectedCluster;
         this.collection = resolvedCollection;
 
-        // Start a periodic background flush so that buffer-flush.interval is a true max-latency
-        // guarantee, even when no rows arrive (idle or low-throughput streaming jobs).
-        final long batchIntervalMs = options.getBatchIntervalMs();
-        if (batchIntervalMs > 0) {
-            this.flushScheduler =
-                    Executors.newSingleThreadScheduledExecutor(
-                            r -> {
-                                Thread t = new Thread(r, "couchbase-flush-timer");
-                                t.setDaemon(true);
-                                return t;
-                            });
-            this.flushFuture =
-                    flushScheduler.scheduleAtFixedRate(
-                            () -> {
-                                try {
-                                    doFlush();
-                                } catch (Exception e) {
-                                    // Latch only the first failure; subsequent timer ticks will
-                                    // observe the latch and skip attempting further flushes.
-                                    if (asyncFlushError.compareAndSet(null, e)) {
-                                        log.error(
-                                                "Periodic Couchbase flush failed — task will abort"
-                                                        + " on the next write or checkpoint",
-                                                e);
-                                    }
-                                }
-                            },
-                            batchIntervalMs,
-                            batchIntervalMs,
-                            TimeUnit.MILLISECONDS);
-        } else {
-            this.flushScheduler = null;
-            this.flushFuture = null;
+        // Opt in to engine-level timer flush. On Zeta the engine invokes this action on the normal
+        // Sink input-processing path when a FlushSignal arrives, so there is no connector-owned
+        // scheduler thread and no concurrency with write/checkpoint/close. On Spark and Flink the
+        // Context does not implement registerFlushAction (it keeps the interface's no-op default),
+        // so there is no periodic timer flush there; the buffer is flushed on
+        // buffer-flush.max-rows,
+        // on checkpoint, and on close(). The null-check is defensive for non-standard/test call
+        // sites that may not supply a context.
+        if (context != null) {
+            context.registerFlushAction(this::doFlush);
         }
     }
 
@@ -204,22 +151,17 @@ public CouchbaseWriter(
      * because CDC delete support is out of scope for this initial implementation.
      *
      * <p>The row's document id is assigned here, at buffer-add time, so that every subsequent flush
-     * attempt (including a {@link #close}-triggered re-flush of a buffer that a prior interrupted
-     * periodic flush left un-cleared) uses the same stable id. This prevents silent duplicate
-     * documents when no primary key is configured (UUID path) and prevents spurious {@link
+     * attempt (including a {@link #close}-triggered re-flush of a buffer that a prior flush left
+     * un-cleared) uses the same stable id. This prevents silent duplicate documents when no primary
+     * key is configured (UUID path) and prevents spurious {@link
      * com.couchbase.client.core.error.DocumentExistsException} collisions on already-committed rows
      * when a primary key is configured and insert mode is active.
      *
-     * <p>Checks {@link #asyncFlushError} before buffering and rethrows any latched background
-     * failure so the task cannot continue after a broken periodic flush.
-     *
-     * @param row the incoming row
-     * @throws CouchbaseConnectorException if the row kind is {@code DELETE} or a prior async flush
-     *     failed
+     * @param row the incoming row throws the CouchBaseConnectorException if the row kind is {@code
+     *     DELETE}
      */
     @Override
     public void write(SeaTunnelRow row) {
-        checkAsyncFlushError();
         if (row.getRowKind() == RowKind.UPDATE_BEFORE) {
             return;
         }
@@ -243,7 +185,6 @@ public void write(SeaTunnelRow row) {
 
     @Override
     public Optional<Void> prepareCommit() {
-        checkAsyncFlushError();
         doFlush();
         return Optional.empty();
     }
@@ -253,36 +194,8 @@ public void abortPrepare() {}
 
     @Override
     public void close() {
-        // Cancel the periodic flush timer first so it cannot race with the final flush below.
-        if (flushFuture != null) {
-            flushFuture.cancel(false);
-        }
-        if (flushScheduler != null) {
-            // shutdownNow() interrupts any in-flight periodic doFlush() that is sleeping in its
-            // retry backoff. We then awaitTermination() to ensure the interrupted timer thread
-            // has fully unwound — in particular that its catch/finally has set asyncFlushError
-            // if the flush failed — before we read the latch or attempt our own doFlush().
-            // Without awaitTermination() there is a window where checkAsyncFlushError() sees null
-            // even though the interrupted task's error handler has not yet executed.
-            flushScheduler.shutdownNow();
-            try {
-                if (!flushScheduler.awaitTermination(30, TimeUnit.SECONDS)) {
-                    log.warn(
-                            "Couchbase flush scheduler did not terminate within 30 s; "
-                                    + "proceeding with close anyway.");
-                }
-            } catch (InterruptedException ie) {
-                Thread.currentThread().interrupt();
-                log.warn("Interrupted while waiting for Couchbase flush scheduler to terminate.");
-            }
-        }
-        // checkAsyncFlushError() and doFlush() are both inside the try so that the finally
-        // block always disconnects the cluster even when a latched background failure exists.
-        // If doFlush() throws and cluster.disconnect() also throws, attach the disconnect failure
-        // as a suppressed exception so the data-relevant flush error is not masked.
         Throwable primaryThrowable = null;
         try {
-            checkAsyncFlushError();
             doFlush();
         } catch (Throwable t) {
             primaryThrowable = t;
@@ -317,26 +230,6 @@ public void close() {
     // ---------------------------------------------------------------------------
     // Internal helpers
     // ---------------------------------------------------------------------------
-
-    /**
-     * Throws a {@link CouchbaseConnectorException} wrapping the latched async error if one has been
-     * recorded by the background flush timer.
-     *
-     * <p>The latch is intentionally <em>sticky</em>: it is read with {@link AtomicReference#get()}
-     * rather than {@code getAndSet(null)} so that a fatal background failure remains visible to
-     * every subsequent caller ({@link #write}, {@link #prepareCommit}, {@link #close}) and cannot
-     * be silently consumed and lost.
-     */
-    private void checkAsyncFlushError() {
-        Throwable error = asyncFlushError.get();
-        if (error != null) {
-            throw new CouchbaseConnectorException(
-                    CouchbaseConnectorErrorCode.WRITE_RECORDS_FAILED,
-                    "A background Couchbase flush failed; task cannot continue safely",
-                    error);
-        }
-    }
-
     /** Converts a {@link SeaTunnelRow} to a {@link JsonObject} using the schema field names. */
     private JsonObject toJsonObject(SeaTunnelRow row) {
         JsonObject doc = JsonObject.create();

seatunnel-connectors-v2/connector-couchbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/couchbase/sink/CouchbaseWriterOptions.java

@@ -38,7 +38,6 @@ public class CouchbaseWriterOptions implements Serializable {
     private final String scope;
     private final String collection;
     private final int flushSize;
-    private final long batchIntervalMs;
     private final boolean upsertEnable;
     private final String[] primaryKey;
     private final int retryMax;
@@ -52,7 +51,6 @@ private CouchbaseWriterOptions(Builder builder) {
         this.scope = builder.scope;
         this.collection = builder.collection;
         this.flushSize = builder.flushSize;
-        this.batchIntervalMs = builder.batchIntervalMs;
         this.upsertEnable = builder.upsertEnable;
         this.primaryKey = builder.primaryKey;
         this.retryMax = builder.retryMax;
@@ -72,7 +70,6 @@ public static class Builder {
         private String scope = "_default";
         private String collection;
         private int flushSize = 1000;
-        private long batchIntervalMs = 30000L;
         private boolean upsertEnable = false;
         private String[] primaryKey = new String[0];
         private int retryMax = 3;
@@ -113,11 +110,6 @@ public Builder withFlushSize(int flushSize) {
             return this;
         }
 
-        public Builder withBatchIntervalMs(long batchIntervalMs) {
-            this.batchIntervalMs = batchIntervalMs;
-            return this;
-        }
-
         public Builder withUpsertEnable(boolean upsertEnable) {
             this.upsertEnable = upsertEnable;
             return this;