[Feature][Connector-V2][Couchbase] Migrate CouchbaseWriter to engine-level timer flush #12016

Search before asking

• I had searched in the feature and found no similar feature requirement.

Description

CouchbaseWriter currently creates and manages its own ScheduledExecutorService to periodically flush buffered documents according to the connector-level buffer-flush.interval option.

This makes the connector responsible for an additional background thread and its lifecycle. It also introduces concurrency between scheduled flushes and the normal Sink paths, including write, prepareCommit, and close.

To propagate failures from the background thread, the Writer additionally maintains an AtomicReference<Throwable> and checks it from subsequent lifecycle calls.

Zeta now provides engine-level timer flush through FlushSignal and SinkWriter.Context#registerFlushAction. Couchbase Sink should adopt this mechanism so that timer flushes run on the normal Sink input-processing path instead of a connector-owned scheduler thread.

This is especially useful for low-throughput or temporarily idle streaming jobs: buffered documents can still be flushed before buffer-flush.max-rows is reached, while flush failures are propagated directly through the task execution path.

Proposed Changes

1. Register CouchbaseWriter#doFlush through the Writer context:

if (context != null) {
    context.registerFlushAction(this::doFlush);
}

2. Remove the connector-owned ScheduledExecutorService, ScheduledFuture, and scheduler startup/shutdown logic.
3. Remove the asynchronous error latch and related checks that become unnecessary after flush execution is serialized onto the Sink task path.
4. Use the engine-level sink.flush.interval option in the job env block for timer-based flushing.
5. Remove or deprecate the connector-level buffer-flush.interval option as appropriate, with a clear compatibility and migration note.
6. Preserve the existing size-based trigger (buffer-flush.max-rows), checkpoint flush in prepareCommit(), and final flush in close().
7. Ensure that timer-flush failures are propagated to the engine instead of being silently treated as successful flushes.
8. Update the English and Chinese Couchbase Sink documentation and mark timer flush as supported.
9. Document that sub-checkpoint engine timer flush is currently supported only by Zeta. On Flink and Spark, buffered records should continue to flush on the size threshold, checkpoint, and Writer close.

Test Requirements

Add focused tests covering the simplified Writer lifecycle and flush failure propagation.

Add or extend the Couchbase Zeta E2E test so that it:

• configures sink.flush.interval in the job env block;
• configures buffer-flush.max-rows above the number of generated records;
• produces a small number of records and then keeps the Source running but idle;
• prevents checkpoint or Writer shutdown from becoming the flush trigger before the assertion;
• verifies that the buffered documents reach Couchbase through the complete flow:

Source timer -> FlushSignal -> Sink -> CouchbaseWriter#doFlush

The test should fail if the engine timer-flush registration is removed.

Reference Implementation

The Prometheus Sink migration in #11778 follows the same pattern and can be used as a reference.

Usage Scenario

A long-running streaming job writes only a small number of Couchbase documents during each interval. The buffer does not reach buffer-flush.max-rows, and the upstream may remain idle for a long time.

Engine-level timer flush sends the buffered documents without waiting for another record, a checkpoint, or Writer shutdown, while avoiding concurrency and lifecycle problems caused by a connector-owned background thread.

Related issues

No response

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct