This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 c2bd543ffc Connector couchbase migration (#12076)
c2bd543ffc is described below
commit c2bd543ffccacd50f7d55425f6118f7771ba00a1
Author: Kaustubh Abhinand <[email protected]>
AuthorDate: Sat Sep 5 12:56:44 2026 +0000
Connector couchbase migration (#12076)
---
docs/en/connectors/sink/Couchbase.md | 23 +-
docs/zh/connectors/sink/Couchbase.md | 21 +-
.../couchbase/config/CouchbaseSinkOptions.java | 13 --
.../couchbase/sink/CouchbaseSinkFactory.java | 3 -
.../seatunnel/couchbase/sink/CouchbaseWriter.java | 137 ++---------
.../couchbase/sink/CouchbaseWriterOptions.java | 8 -
.../couchbase/sink/CouchbaseWriterCloseTest.java | 256 ---------------------
.../sink/CouchbaseWriterConstructorLeakTest.java | 1 -
.../couchbase/sink/CouchbaseWriterTest.java | 250 ++++++++++++++++++++
.../e2e/connector/couchbase/CouchbaseIT.java | 110 ++++++++-
.../test/resources/fake_source_to_couchbase.conf | 1 -
...f => fake_source_to_couchbase_timer_flush.conf} | 16 +-
12 files changed, 418 insertions(+), 421 deletions(-)
diff --git a/docs/en/connectors/sink/Couchbase.md
b/docs/en/connectors/sink/Couchbase.md
index e543fdff35..1304718d51 100644
--- a/docs/en/connectors/sink/Couchbase.md
+++ b/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 />
diff --git a/docs/zh/connectors/sink/Couchbase.md
b/docs/zh/connectors/sink/Couchbase.md
index dbf101070b..43883e9544 100644
--- a/docs/zh/connectors/sink/Couchbase.md
+++ b/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,29 @@ 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
+}
+
+```
+> 在 Spark 和 Flink 上没有检查点之间的定时刷新：`sink.flush.interval` 是 Zeta 引擎的能力，
+> Spark/Flink 的 Sink 写入器上下文并未实现它。在这两个引擎上，缓存会在达到
+> `buffer-flush.max-rows`、检查点时（`CouchbaseWriter` 在 `prepareCommit()` 中刷新）
+> 以及写入器关闭时被刷新。如需降低 Spark 或 Flink 上检查点之间的延迟，
+> 请相应调整 `buffer-flush.max-rows`。
+
<ChangeLog />
diff --git
a/seatunnel-connectors-v2/connector-couchbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/couchbase/config/CouchbaseSinkOptions.java
b/seatunnel-connectors-v2/connector-couchbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/couchbase/config/CouchbaseSinkOptions.java
index 6162ec3940..4d09a802c5 100644
---
a/seatunnel-connectors-v2/connector-couchbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/couchbase/config/CouchbaseSinkOptions.java
+++
b/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")
diff --git
a/seatunnel-connectors-v2/connector-couchbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/couchbase/sink/CouchbaseSinkFactory.java
b/seatunnel-connectors-v2/connector-couchbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/couchbase/sink/CouchbaseSinkFactory.java
index 2fff3c8663..630580311e 100644
---
a/seatunnel-connectors-v2/connector-couchbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/couchbase/sink/CouchbaseSinkFactory.java
+++
b/seatunnel-connectors-v2/connector-couchbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/couchbase/sink/CouchbaseSinkFactory.java
@@ -56,7 +56,6 @@ public class CouchbaseSinkFactory implements TableSinkFactory
{
.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 class CouchbaseSinkFactory implements TableSinkFactory
{
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);
diff --git
a/seatunnel-connectors-v2/connector-couchbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/couchbase/sink/CouchbaseWriter.java
b/seatunnel-connectors-v2/connector-couchbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/couchbase/sink/CouchbaseWriter.java
index 18c9f7e7c1..5da720e4ad 100644
---
a/seatunnel-connectors-v2/connector-couchbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/couchbase/sink/CouchbaseWriter.java
+++
b/seatunnel-connectors-v2/connector-couchbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/couchbase/sink/CouchbaseWriter.java
@@ -46,11 +46,7 @@ import java.util.Map;
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 @@ import java.util.concurrent.atomic.AtomicReference;
*
* <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 @@ import java.util.concurrent.atomic.AtomicReference;
* 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 class CouchbaseWriter implements
SinkWriter<SeaTunnelRow, Void, Void> {
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 class CouchbaseWriter implements
SinkWriter<SeaTunnelRow, Void, Void> {
* 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 class CouchbaseWriter implements
SinkWriter<SeaTunnelRow, Void, Void> {
@Override
public Optional<Void> prepareCommit() {
- checkAsyncFlushError();
doFlush();
return Optional.empty();
}
@@ -253,36 +194,8 @@ public class CouchbaseWriter implements
SinkWriter<SeaTunnelRow, Void, Void> {
@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 class CouchbaseWriter implements
SinkWriter<SeaTunnelRow, Void, Void> {
//
---------------------------------------------------------------------------
// 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();
diff --git
a/seatunnel-connectors-v2/connector-couchbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/couchbase/sink/CouchbaseWriterOptions.java
b/seatunnel-connectors-v2/connector-couchbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/couchbase/sink/CouchbaseWriterOptions.java
index 866c5fa7d3..324176d276 100644
---
a/seatunnel-connectors-v2/connector-couchbase/src/main/java/org/apache/seatunnel/connectors/seatunnel/couchbase/sink/CouchbaseWriterOptions.java
+++
b/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 @@ public class CouchbaseWriterOptions implements Serializable {
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 class CouchbaseWriterOptions implements Serializable {
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 class CouchbaseWriterOptions implements
Serializable {
return this;
}
- public Builder withBatchIntervalMs(long batchIntervalMs) {
- this.batchIntervalMs = batchIntervalMs;
- return this;
- }
-
public Builder withUpsertEnable(boolean upsertEnable) {
this.upsertEnable = upsertEnable;
return this;
diff --git
a/seatunnel-connectors-v2/connector-couchbase/src/test/java/org/apache/seatunnel/connectors/seatunnel/couchbase/sink/CouchbaseWriterCloseTest.java
b/seatunnel-connectors-v2/connector-couchbase/src/test/java/org/apache/seatunnel/connectors/seatunnel/couchbase/sink/CouchbaseWriterCloseTest.java
deleted file mode 100644
index aac404e279..0000000000
---
a/seatunnel-connectors-v2/connector-couchbase/src/test/java/org/apache/seatunnel/connectors/seatunnel/couchbase/sink/CouchbaseWriterCloseTest.java
+++ /dev/null
@@ -1,256 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.seatunnel.connectors.seatunnel.couchbase.sink;
-
-import org.apache.seatunnel.api.sink.SinkWriter;
-import org.apache.seatunnel.api.table.catalog.CatalogTable;
-import org.apache.seatunnel.api.table.catalog.TableIdentifier;
-import org.apache.seatunnel.api.table.catalog.TableSchema;
-
-import org.junit.jupiter.api.Test;
-import org.mockito.MockedStatic;
-import org.mockito.Mockito;
-
-import com.couchbase.client.java.Bucket;
-import com.couchbase.client.java.Cluster;
-import com.couchbase.client.java.Collection;
-import com.couchbase.client.java.Scope;
-
-import java.lang.reflect.Field;
-import java.util.concurrent.atomic.AtomicReference;
-
-import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.mockito.ArgumentMatchers.anyString;
-import static org.mockito.Mockito.atLeastOnce;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
-/**
- * Unit tests covering the two fixes applied to {@link
CouchbaseWriter#close()} and {@link
- * CouchbaseWriter}'s async-error latch:
- *
- * <ol>
- * <li><b>Blocker 1a</b> — {@code cluster.disconnect()} must be called even
when a latched
- * background flush error causes {@code checkAsyncFlushError()} to throw
inside {@code
- * close()}.
- * <li><b>Blocker 1b</b> — The {@code asyncFlushError} latch must be
<em>sticky</em>: reading it
- * via {@link AtomicReference#get()} must not clear it, so a second call
to {@code
- * checkAsyncFlushError()} still sees the same error.
- * </ol>
- *
- * <p>Uses {@code mockito-inline} to intercept the static {@link
Cluster#connect} factory.
- */
-class CouchbaseWriterCloseTest {
-
- //
-------------------------------------------------------------------------
- // Shared helpers (mirrors CouchbaseWriterConstructorLeakTest)
- //
-------------------------------------------------------------------------
-
- private static CouchbaseWriterOptions minimalOptions() {
- return CouchbaseWriterOptions.builder()
- .withConnectionString("couchbase://localhost")
- .withUsername("user")
- .withPassword("pass")
- .withBucket("test-bucket")
- .withScope("_default")
- .withCollection("test-collection")
- .withBatchIntervalMs(-1) // disable background scheduler
- .build();
- }
-
- private static CatalogTable minimalCatalogTable() {
- TableSchema schema = TableSchema.builder().build();
- return CatalogTable.of(
- TableIdentifier.of("catalog", "database", "table"),
- schema,
- java.util.Collections.emptyMap(),
- java.util.Collections.emptyList(),
- "");
- }
-
- private static SinkWriter.Context noopContext() {
- return mock(SinkWriter.Context.class);
- }
-
- private static Cluster buildMockClusterChain() {
- Collection collection = mock(Collection.class);
- Scope scope = mock(Scope.class);
- when(scope.collection(anyString())).thenReturn(collection);
- Bucket bucket = mock(Bucket.class);
- when(bucket.scope(anyString())).thenReturn(scope);
- Cluster cluster = mock(Cluster.class);
- when(cluster.bucket(anyString())).thenReturn(bucket);
- return cluster;
- }
-
- /**
- * Reflectively injects a {@link Throwable} into the {@code
asyncFlushError} latch of the given
- * writer, simulating a failure that was recorded by the background flush
timer.
- */
- private static void injectAsyncFlushError(CouchbaseWriter writer,
Throwable error)
- throws Exception {
- Field field =
CouchbaseWriter.class.getDeclaredField("asyncFlushError");
- field.setAccessible(true);
- @SuppressWarnings("unchecked")
- AtomicReference<Throwable> ref = (AtomicReference<Throwable>)
field.get(writer);
- ref.set(error);
- }
-
- //
-------------------------------------------------------------------------
- // Blocker 1a — disconnect() must execute even when the latch throws
- //
-------------------------------------------------------------------------
-
- /**
- * Verifies that {@code cluster.disconnect()} is called by the {@code
finally} block inside
- * {@code close()} even when a latched async flush error causes {@code
checkAsyncFlushError()}
- * to throw.
- *
- * <p>Before the fix, {@code checkAsyncFlushError()} was invoked
<em>before</em> entering the
- * {@code try/finally}, so a latched error caused the method to return
early and {@code
- * disconnect()} was never reached.
- */
- @Test
- void close_withLatchedAsyncError_stillDisconnectsCluster() throws
Exception {
- Cluster cluster = buildMockClusterChain();
-
- try (MockedStatic<Cluster> staticCluster =
Mockito.mockStatic(Cluster.class)) {
- staticCluster
- .when(() -> Cluster.connect(anyString(), anyString(),
anyString()))
- .thenReturn(cluster);
-
- CouchbaseWriter writer =
- new CouchbaseWriter(minimalOptions(),
minimalCatalogTable(), noopContext());
-
- // Simulate a background flush failure recorded by the timer
thread.
- injectAsyncFlushError(writer, new RuntimeException("simulated
timer flush failure"));
-
- // close() must throw because of the latched error — but
disconnect() must still run.
- assertThrows(
- Exception.class, writer::close, "close() must rethrow the
latched async error");
-
- // The critical assertion: disconnect() must have been called
exactly once
- // via the finally block, regardless of the async-error exception.
- verify(cluster, atLeastOnce()).disconnect();
- }
- }
-
- /**
- * Complementary happy-path: when no error is latched, {@code close()}
must complete without
- * throwing and {@code disconnect()} must still be called.
- */
- @Test
- void close_withNoLatchedError_disconnectsCluster() {
- Cluster cluster = buildMockClusterChain();
-
- try (MockedStatic<Cluster> staticCluster =
Mockito.mockStatic(Cluster.class)) {
- staticCluster
- .when(() -> Cluster.connect(anyString(), anyString(),
anyString()))
- .thenReturn(cluster);
-
- CouchbaseWriter writer =
- new CouchbaseWriter(minimalOptions(),
minimalCatalogTable(), noopContext());
-
- assertDoesNotThrow(writer::close, "close() must not throw when
there is no error");
-
- verify(cluster, atLeastOnce()).disconnect();
- }
- }
-
- //
-------------------------------------------------------------------------
- // Blocker 1b — asyncFlushError latch must be sticky (not consumed on
first read)
- //
-------------------------------------------------------------------------
-
- /**
- * Verifies that the {@code asyncFlushError} latch is <em>sticky</em>:
after a background error
- * is recorded, two successive calls to {@code checkAsyncFlushError()}
(via {@link
- * CouchbaseWriter#write} and then {@link CouchbaseWriter#prepareCommit})
must both throw.
- *
- * <p>Before the fix, the latch used {@code getAndSet(null)}, so only the
first caller saw the
- * error; the second call would silently return, allowing the task to
proceed after a fatal
- * flush failure.
- */
- @Test
- void asyncFlushError_latchIsSticky_secondCallerAlsoSeesError() throws
Exception {
- Cluster cluster = buildMockClusterChain();
-
- try (MockedStatic<Cluster> staticCluster =
Mockito.mockStatic(Cluster.class)) {
- staticCluster
- .when(() -> Cluster.connect(anyString(), anyString(),
anyString()))
- .thenReturn(cluster);
-
- CouchbaseWriter writer =
- new CouchbaseWriter(minimalOptions(),
minimalCatalogTable(), noopContext());
-
- injectAsyncFlushError(writer, new RuntimeException("background
error"));
-
- // First caller (write) must throw.
- assertThrows(
- Exception.class,
- () -> writer.write(new
org.apache.seatunnel.api.table.type.SeaTunnelRow(0)),
- "write() must throw when the latch is set");
-
- // Second caller (prepareCommit) must ALSO throw — the latch must
not have been cleared.
- assertThrows(
- Exception.class,
- writer::prepareCommit,
- "prepareCommit() must still throw: the async-error latch
must be sticky");
-
- // Cleanup: close() will also throw (latch still set) but
disconnect() must run.
- assertThrows(Exception.class, writer::close);
- verify(cluster, atLeastOnce()).disconnect();
- }
- }
-
- /**
- * Verifies that a single injected error remains visible through all three
lifecycle callers in
- * sequence: {@link CouchbaseWriter#write}, {@link
CouchbaseWriter#prepareCommit}, and {@link
- * CouchbaseWriter#close}.
- */
- @Test
- void asyncFlushError_latchIsSticky_allThreeLifecycleCallersThrow() throws
Exception {
- Cluster cluster = buildMockClusterChain();
-
- try (MockedStatic<Cluster> staticCluster =
Mockito.mockStatic(Cluster.class)) {
- staticCluster
- .when(() -> Cluster.connect(anyString(), anyString(),
anyString()))
- .thenReturn(cluster);
-
- CouchbaseWriter writer =
- new CouchbaseWriter(minimalOptions(),
minimalCatalogTable(), noopContext());
-
- injectAsyncFlushError(writer, new RuntimeException("persistent
background error"));
-
- assertThrows(
- Exception.class,
- () -> writer.write(new
org.apache.seatunnel.api.table.type.SeaTunnelRow(0)),
- "write() must throw");
- assertThrows(
- Exception.class,
- writer::prepareCommit,
- "prepareCommit() must throw (latch not cleared by write)");
- assertThrows(
- Exception.class,
- writer::close,
- "close() must throw (latch not cleared by prepareCommit)");
-
- verify(cluster, atLeastOnce()).disconnect();
- }
- }
-}
diff --git
a/seatunnel-connectors-v2/connector-couchbase/src/test/java/org/apache/seatunnel/connectors/seatunnel/couchbase/sink/CouchbaseWriterConstructorLeakTest.java
b/seatunnel-connectors-v2/connector-couchbase/src/test/java/org/apache/seatunnel/connectors/seatunnel/couchbase/sink/CouchbaseWriterConstructorLeakTest.java
index a943bda942..7d5ce27ef5 100644
---
a/seatunnel-connectors-v2/connector-couchbase/src/test/java/org/apache/seatunnel/connectors/seatunnel/couchbase/sink/CouchbaseWriterConstructorLeakTest.java
+++
b/seatunnel-connectors-v2/connector-couchbase/src/test/java/org/apache/seatunnel/connectors/seatunnel/couchbase/sink/CouchbaseWriterConstructorLeakTest.java
@@ -68,7 +68,6 @@ class CouchbaseWriterConstructorLeakTest {
.withBucket("test-bucket")
.withScope("_default")
.withCollection("test-collection")
- .withBatchIntervalMs(-1) // disable background scheduler
.build();
}
diff --git
a/seatunnel-connectors-v2/connector-couchbase/src/test/java/org/apache/seatunnel/connectors/seatunnel/couchbase/sink/CouchbaseWriterTest.java
b/seatunnel-connectors-v2/connector-couchbase/src/test/java/org/apache/seatunnel/connectors/seatunnel/couchbase/sink/CouchbaseWriterTest.java
new file mode 100644
index 0000000000..4ee11ae9a4
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-couchbase/src/test/java/org/apache/seatunnel/connectors/seatunnel/couchbase/sink/CouchbaseWriterTest.java
@@ -0,0 +1,250 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.couchbase.sink;
+
+import org.apache.seatunnel.api.sink.SinkWriter;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.type.RowKind;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.common.utils.function.RunnableWithException;
+import
org.apache.seatunnel.connectors.seatunnel.couchbase.exception.CouchbaseConnectorException;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import com.couchbase.client.java.Bucket;
+import com.couchbase.client.java.Cluster;
+import com.couchbase.client.java.Collection;
+import com.couchbase.client.java.Scope;
+import com.couchbase.client.java.kv.MutationResult;
+
+import java.util.Collections;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class CouchbaseWriterTest {
+
+ // ---- shared helpers -------------------------------------------------
+
+ private static CouchbaseWriterOptions minimalOptions() {
+ return CouchbaseWriterOptions.builder()
+ .withConnectionString("couchbase://localhost")
+ .withUsername("user")
+ .withPassword("pass")
+ .withBucket("test-bucket")
+ .withScope("_default")
+ .withCollection("test-collection")
+ .withUpsertEnable(true)
+ .withFlushSize(-1) // disable size-based flush unless a test
needs it
+ .build();
+ }
+
+ private static CatalogTable minimalCatalogTable() {
+ TableSchema schema = TableSchema.builder().build();
+ return CatalogTable.of(
+ TableIdentifier.of("catalog", "database", "table"),
+ schema,
+ Collections.emptyMap(),
+ Collections.emptyList(),
+ "");
+ }
+
+ private static Collection buildMockCollectionChain(Cluster cluster) {
+ Collection collection = mock(Collection.class);
+ Scope scope = mock(Scope.class);
+ when(scope.collection(anyString())).thenReturn(collection);
+ Bucket bucket = mock(Bucket.class);
+ when(bucket.scope(anyString())).thenReturn(scope);
+ when(cluster.bucket(anyString())).thenReturn(bucket);
+ return collection;
+ }
+
+ private static SeaTunnelRow insertRow() {
+ SeaTunnelRow row = new SeaTunnelRow(new Object[] {"v1"});
+ row.setRowKind(RowKind.INSERT);
+ return row;
+ }
+
+ // ---- 1. registration + flush on signal -------------------------------
+
+ @Test
+ void shouldRegisterFlushActionAndFlushBufferedRecordsOnSignal() throws
Exception {
+ Cluster cluster = mock(Cluster.class);
+ Collection collection = buildMockCollectionChain(cluster);
+ when(collection.upsert(anyString(),
any())).thenReturn(mock(MutationResult.class));
+
+ SinkWriter.Context context = mock(SinkWriter.Context.class);
+ ArgumentCaptor<RunnableWithException> actionCaptor =
+ ArgumentCaptor.forClass(RunnableWithException.class);
+
+ try (MockedStatic<Cluster> staticCluster =
Mockito.mockStatic(Cluster.class)) {
+ staticCluster
+ .when(() -> Cluster.connect(anyString(), anyString(),
anyString()))
+ .thenReturn(cluster);
+
+ CouchbaseWriter writer =
+ new CouchbaseWriter(minimalOptions(),
minimalCatalogTable(), context);
+ writer.write(insertRow());
+
+ verify(context,
times(1)).registerFlushAction(actionCaptor.capture());
+ // Buffered only: nothing written until the engine delivers the
flush signal.
+ verify(collection, never()).upsert(anyString(), any());
+
+ actionCaptor.getValue().run();
+
+ verify(collection, times(1)).upsert(anyString(), any());
+ }
+ }
+
+ // ---- 2. flush failure propagation -------------------------------------
+
+ @Test
+ void shouldPropagateFlushFailure() throws Exception {
+ Cluster cluster = mock(Cluster.class);
+ Collection collection = buildMockCollectionChain(cluster);
+ doThrow(new
RuntimeException("boom")).when(collection).upsert(anyString(), any());
+
+ SinkWriter.Context context = mock(SinkWriter.Context.class);
+ ArgumentCaptor<RunnableWithException> actionCaptor =
+ ArgumentCaptor.forClass(RunnableWithException.class);
+
+ try (MockedStatic<Cluster> staticCluster =
Mockito.mockStatic(Cluster.class)) {
+ staticCluster
+ .when(() -> Cluster.connect(anyString(), anyString(),
anyString()))
+ .thenReturn(cluster);
+
+ CouchbaseWriter writer =
+ new CouchbaseWriter(minimalOptions(),
minimalCatalogTable(), context);
+ writer.write(insertRow());
+
+ verify(context,
times(1)).registerFlushAction(actionCaptor.capture());
+ Assertions.assertThrows(
+ CouchbaseConnectorException.class, () ->
actionCaptor.getValue().run());
+ }
+ }
+
+ // ---- 3. fallback flush on close (Spark/Flink; no registerFlushAction) --
+
+ @Test
+ void shouldFlushOnCloseWhenEngineNeverInvokesFlushAction() throws
Exception {
+ Cluster cluster = mock(Cluster.class);
+ Collection collection = buildMockCollectionChain(cluster);
+ when(collection.upsert(anyString(),
any())).thenReturn(mock(MutationResult.class));
+
+ SinkWriter.Context context = mock(SinkWriter.Context.class);
+
+ try (MockedStatic<Cluster> staticCluster =
Mockito.mockStatic(Cluster.class)) {
+ staticCluster
+ .when(() -> Cluster.connect(anyString(), anyString(),
anyString()))
+ .thenReturn(cluster);
+
+ CouchbaseWriter writer =
+ new CouchbaseWriter(minimalOptions(),
minimalCatalogTable(), context);
+ writer.write(insertRow());
+
+ // Simulate Spark/Flink: registered action is never invoked.
+ verify(collection, never()).upsert(anyString(), any());
+
+ writer.close();
+
+ verify(collection, times(1)).upsert(anyString(), any());
+ verify(cluster, times(1)).disconnect();
+ }
+ }
+
+ // ---- 4. checkpoint flush (prepareCommit) -------------------------------
+
+ @Test
+ void shouldFlushOnPrepareCommitWhenEngineNeverInvokesFlushAction() throws
Exception {
+ Cluster cluster = mock(Cluster.class);
+ Collection collection = buildMockCollectionChain(cluster);
+ when(collection.upsert(anyString(),
any())).thenReturn(mock(MutationResult.class));
+
+ SinkWriter.Context context = mock(SinkWriter.Context.class);
+
+ try (MockedStatic<Cluster> staticCluster =
Mockito.mockStatic(Cluster.class)) {
+ staticCluster
+ .when(() -> Cluster.connect(anyString(), anyString(),
anyString()))
+ .thenReturn(cluster);
+
+ CouchbaseWriter writer =
+ new CouchbaseWriter(minimalOptions(),
minimalCatalogTable(), context);
+ writer.write(insertRow());
+
+ verify(collection, never()).upsert(anyString(), any());
+
+ Assertions.assertEquals(java.util.Optional.empty(),
writer.prepareCommit());
+
+ verify(collection, times(1)).upsert(anyString(), any());
+
+ // Buffer must be cleared: a second checkpoint with no new rows
sends nothing more.
+ Assertions.assertEquals(java.util.Optional.empty(),
writer.prepareCommit());
+ verify(collection, times(1)).upsert(anyString(), any());
+ }
+ }
+
+ // ---- 5. suppressed disconnect exception on close when flush fails ------
+
+ @Test
+ void closeShouldKeepFlushExceptionWhenDisconnectAlsoThrows() throws
Exception {
+ Cluster cluster = mock(Cluster.class);
+ Collection collection = buildMockCollectionChain(cluster);
+ doThrow(new
RuntimeException("boom")).when(collection).upsert(anyString(), any());
+ doThrow(new RuntimeException("disconnect
failed")).when(cluster).disconnect();
+
+ SinkWriter.Context context = mock(SinkWriter.Context.class);
+
+ try (MockedStatic<Cluster> staticCluster =
Mockito.mockStatic(Cluster.class)) {
+ staticCluster
+ .when(() -> Cluster.connect(anyString(), anyString(),
anyString()))
+ .thenReturn(cluster);
+
+ CouchbaseWriter writer =
+ new CouchbaseWriter(minimalOptions(),
minimalCatalogTable(), context);
+ writer.write(insertRow());
+
+ CouchbaseConnectorException thrown =
+ Assertions.assertThrows(CouchbaseConnectorException.class,
writer::close);
+
+ boolean disconnectSuppressed = false;
+ for (Throwable suppressed : thrown.getSuppressed()) {
+ if (suppressed.getMessage() != null
+ && suppressed.getMessage().contains("disconnect
failed")) {
+ disconnectSuppressed = true;
+ }
+ }
+ Assertions.assertTrue(
+ disconnectSuppressed,
+ "The disconnect failure should be suppressed on the
primary flush exception");
+
+ verify(cluster, times(1)).disconnect();
+ }
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-couchbase-e2e/src/test/java/org/apache/seatunnel/e2e/connector/couchbase/CouchbaseIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-couchbase-e2e/src/test/java/org/apache/seatunnel/e2e/connector/couchbase/CouchbaseIT.java
index b9be68fc96..20c7aeb8cd 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-couchbase-e2e/src/test/java/org/apache/seatunnel/e2e/connector/couchbase/CouchbaseIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-couchbase-e2e/src/test/java/org/apache/seatunnel/e2e/connector/couchbase/CouchbaseIT.java
@@ -19,8 +19,10 @@ package org.apache.seatunnel.e2e.connector.couchbase;
import org.apache.seatunnel.e2e.common.TestResource;
import org.apache.seatunnel.e2e.common.TestSuiteBase;
+import org.apache.seatunnel.e2e.common.container.EngineType;
import org.apache.seatunnel.e2e.common.container.TestContainer;
import org.apache.seatunnel.e2e.common.container.seatunnel.SeaTunnelContainer;
+import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer;
import org.awaitility.Awaitility;
import org.junit.jupiter.api.AfterAll;
@@ -63,6 +65,7 @@ public class CouchbaseIT extends TestSuiteBase implements
TestResource {
private static final String COUCHBASE_BUCKET = "test_bucket";
private static final String COUCHBASE_SCOPE = "_default";
private static final String COUCHBASE_COLLECTION = "test_collection";
+ private static final String COUCHBASE_COLLECTION_TIMER_FLUSH =
"test_collection_timer_flush";
/** Matches row.num in fake_source_to_couchbase.conf. */
private static final int EXPECTED_ROW_COUNT = 100;
@@ -125,6 +128,7 @@ public class CouchbaseIT extends TestSuiteBase implements
TestResource {
// reject requests for a short window after Cluster.connect() returns.
Wrapping the DDL
// itself in a retry loop is the safest approach — it eliminates the
startup timing race
// without relying on a fixed sleep.
+
String createCollectionDdl =
"CREATE COLLECTION `"
+ COUCHBASE_BUCKET
@@ -139,7 +143,6 @@ public class CouchbaseIT extends TestSuiteBase implements
TestResource {
.atMost(60, TimeUnit.SECONDS)
.untilAsserted(() -> cluster.query(createCollectionDdl));
- // Similarly retry CREATE PRIMARY INDEX until the collection is
visible to the query path.
String createIndexDdl =
"CREATE PRIMARY INDEX ON `"
+ COUCHBASE_BUCKET
@@ -154,9 +157,6 @@ public class CouchbaseIT extends TestSuiteBase implements
TestResource {
.atMost(60, TimeUnit.SECONDS)
.untilAsserted(() -> cluster.query(createIndexDdl));
- // Wait until the primary index transitions to 'online' before
returning.
- // The index build is asynchronous; without this guard the first N1QL
query in the
- // test body may hit the collection before the index is ready and
return no results.
String indexStatusQuery =
String.format(
"SELECT state FROM system:indexes WHERE keyspace_id =
'%s'"
@@ -178,6 +178,55 @@ public class CouchbaseIT extends TestSuiteBase implements
TestResource {
"Primary index not yet online");
});
+ // Similarly retry CREATE PRIMARY INDEX until the collection is
visible to the query path.
+ String createCollectionDdlTimerFlush =
+ "CREATE COLLECTION `"
+ + COUCHBASE_BUCKET
+ + "`.`"
+ + COUCHBASE_SCOPE
+ + "`.`"
+ + COUCHBASE_COLLECTION_TIMER_FLUSH
+ + "`";
+ Awaitility.given()
+ .ignoreExceptions()
+ .pollInterval(2, TimeUnit.SECONDS)
+ .atMost(60, TimeUnit.SECONDS)
+ .untilAsserted(() ->
cluster.query(createCollectionDdlTimerFlush));
+
+ String createIndexDdlTimerFlush =
+ "CREATE PRIMARY INDEX ON `"
+ + COUCHBASE_BUCKET
+ + "`.`"
+ + COUCHBASE_SCOPE
+ + "`.`"
+ + COUCHBASE_COLLECTION_TIMER_FLUSH
+ + "`";
+ Awaitility.given()
+ .ignoreExceptions()
+ .pollInterval(2, TimeUnit.SECONDS)
+ .atMost(60, TimeUnit.SECONDS)
+ .untilAsserted(() -> cluster.query(createIndexDdlTimerFlush));
+
+ String indexStatusQueryTimerFlush =
+ String.format(
+ "SELECT state FROM system:indexes WHERE keyspace_id =
'%s'"
+ + " AND `using` = 'gsi' AND is_primary = true"
+ + " AND `bucket_id` = '%s'",
+ COUCHBASE_COLLECTION_TIMER_FLUSH, COUCHBASE_BUCKET);
+ Awaitility.given()
+ .ignoreExceptions()
+ .pollInterval(1, TimeUnit.SECONDS)
+ .atMost(60, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ QueryResult r =
cluster.query(indexStatusQueryTimerFlush);
+ List<JsonObject> rows = r.rowsAs(JsonObject.class);
+ Assertions.assertFalse(rows.isEmpty(), "Primary
index not created yet");
+ Assertions.assertEquals(
+ "online",
+ rows.get(0).getString("state"),
+ "Primary index not yet online");
+ });
log.info("Couchbase cluster ready at {}",
couchbaseContainer.getConnectionString());
}
@@ -289,4 +338,57 @@ public class CouchbaseIT extends TestSuiteBase implements
TestResource {
doc.toMap());
});
}
+
+ @TestTemplate
+ @DisabledOnContainer(
+ value = {},
+ type = {EngineType.SPARK, EngineType.FLINK},
+ disabledReason =
+ "engine-level timer flush (sink.flush.interval) is only
supported on Zeta engine")
+ public void testCouchbaseSinkTimerFlush(TestContainer container) throws
Exception {
+ cluster.query(
+ String.format(
+ "DELETE FROM `%s`.`%s`.`%s`",
+ COUCHBASE_BUCKET, COUCHBASE_SCOPE,
COUCHBASE_COLLECTION_TIMER_FLUSH));
+
+ String jobId = String.valueOf(System.currentTimeMillis());
+ java.util.concurrent.CompletableFuture<Container.ExecResult> jobFuture
=
+ java.util.concurrent.CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return container.executeJob(
+
"/fake_source_to_couchbase_timer_flush.conf", jobId);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+
+ String countQuery =
+ String.format(
+ "SELECT COUNT(*) AS cnt FROM `%s`.`%s`.`%s`",
+ COUCHBASE_BUCKET, COUCHBASE_SCOPE,
COUCHBASE_COLLECTION_TIMER_FLUSH);
+
+ try {
+ Awaitility.given()
+ .ignoreExceptions()
+ .pollInterval(2, TimeUnit.SECONDS)
+ .atMost(60, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ Assertions.assertFalse(
+ jobFuture.isDone(),
+ "The streaming job must still be
running when timer flush publishes the buffered rows");
+ QueryResult result = cluster.query(countQuery);
+ List<JsonObject> rows =
result.rowsAs(JsonObject.class);
+ Assertions.assertFalse(
+ rows.isEmpty(), "COUNT query returned
no rows");
+ Assertions.assertEquals(10,
rows.get(0).getInt("cnt"));
+ });
+ } finally {
+ if (!jobFuture.isDone()) {
+ Container.ExecResult cancelResult = container.cancelJob(jobId);
+ Assertions.assertEquals(0, cancelResult.getExitCode(),
cancelResult.getStderr());
+ }
+ }
+ }
}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-couchbase-e2e/src/test/resources/fake_source_to_couchbase.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-couchbase-e2e/src/test/resources/fake_source_to_couchbase.conf
index e1986d88dd..9ec98fb6d3 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-couchbase-e2e/src/test/resources/fake_source_to_couchbase.conf
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-couchbase-e2e/src/test/resources/fake_source_to_couchbase.conf
@@ -55,6 +55,5 @@ sink {
primary-key = ["id"]
upsert-enable = true
buffer-flush.max-rows = 50
- buffer-flush.interval = 5000
}
}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-couchbase-e2e/src/test/resources/fake_source_to_couchbase.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-couchbase-e2e/src/test/resources/fake_source_to_couchbase_timer_flush.conf
similarity index 79%
copy from
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-couchbase-e2e/src/test/resources/fake_source_to_couchbase.conf
copy to
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-couchbase-e2e/src/test/resources/fake_source_to_couchbase_timer_flush.conf
index e1986d88dd..6ee7ac6fdf 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-couchbase-e2e/src/test/resources/fake_source_to_couchbase.conf
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-couchbase-e2e/src/test/resources/fake_source_to_couchbase_timer_flush.conf
@@ -17,16 +17,15 @@
env {
parallelism = 1
- job.mode = "BATCH"
+ job.mode = "STREAMING"
+ checkpoint.interval = 300000
+ sink.flush.interval = 3000
}
source {
FakeSource {
plugin_output = "fake"
- row.num = 100
- # auto.increment.enabled makes the primary-key field sequential (1..100) so
- # upsert-enable=true produces exactly 100 distinct document keys and the
- # COUNT(*) assertion in CouchbaseIT is fully deterministic.
+ row.num = 10
auto.increment.enabled = true
auto.increment.start = 1
schema = {
@@ -51,10 +50,9 @@ sink {
password = "password"
bucket = "test_bucket"
scope = "_default"
- collection = "test_collection"
+ collection = "test_collection_timer_flush"
primary-key = ["id"]
upsert-enable = true
- buffer-flush.max-rows = 50
- buffer-flush.interval = 5000
+ buffer-flush.max-rows = 100000
}
-}
+}
\ No newline at end of file
