nzw921rx opened a new issue, #12016: URL: https://github.com/apache/seatunnel/issues/12016
### Search before asking - [x] I had searched in the [feature](https://github.com/apache/seatunnel/issues?q=is%3Aissue+label%3A%22Feature%22) 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: ```java 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. The generic engine timer-flush E2E coverage is located under: `seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base` ### 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 - [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
