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 0278a6a747 [Fix][E2E] Fix task recovery hangs and PostgreSQL CDC LSN
handling (#12454)
0278a6a747 is described below
commit 0278a6a74746856488dd27713736f008e3115c11
Author: SEZ <[email protected]>
AuthorDate: Fri Oct 9 09:26:17 2026 +0000
[Fix][E2E] Fix task recovery hangs and PostgreSQL CDC LSN handling (#12454)
---
docs/en/connectors/source/PostgreSQL-CDC.md | 7 +
docs/en/engines/zeta/user-command.md | 5 +
docs/zh/connectors/source/PostgreSQL-CDC.md | 6 +
docs/zh/engines/zeta/user-command.md | 3 +
.../postgresql/PostgresOffsetContext.java | 54 +-
.../PostgresStreamingChangeEventSource.java | 552 +++++++++++++++++++++
.../postgresql/connection/WalPositionLocator.java | 230 +++++++++
.../reader/PostgresSourceFetchTaskContext.java | 8 +-
.../cdc/postgres/source/offset/LsnOffsetTest.java | 114 +++++
.../seatunnel/cdc/postgres/PostgresCDCIT.java | 41 +-
.../seatunnel/e2e/connector/paypal/PayPalIT.java | 4 +-
.../e2e/CheckpointCoordinatorFailoverIT.java | 2 +
.../e2e/transform/TestFilterRowKindIT.java | 1 +
...filter_row_kind_exclude_insert_multi_table.conf | 5 +-
.../engine/server/TaskExecutionService.java | 14 +-
.../engine/server/dag/physical/PhysicalVertex.java | 8 +
.../engine/server/TaskExecutionServiceTest.java | 140 ++++++
.../dag/physical/StateTransitionCleanupTest.java | 60 +++
18 files changed, 1224 insertions(+), 30 deletions(-)
diff --git a/docs/en/connectors/source/PostgreSQL-CDC.md
b/docs/en/connectors/source/PostgreSQL-CDC.md
index b5b45d83bf..ac6557a509 100644
--- a/docs/en/connectors/source/PostgreSQL-CDC.md
+++ b/docs/en/connectors/source/PostgreSQL-CDC.md
@@ -386,6 +386,13 @@ uses its `confirmed_flush_lsn` as the startup offset.
Unused replication slots hold WAL segments on disk, which can cause unbounded
WAL growth. When a
CDC job is permanently decommissioned, drop the unused replication slot
manually on PostgreSQL.
+### How are transaction boundaries handled during savepoint recovery?
+
+Recovery distinguishes a saved transaction commit from a processed row,
because a commit and the
+next transaction's first row can share the same LSN. Rows committed while the
job is stopped remain
+eligible for replay. Existing savepoints without the saved operation type
remain readable; recovery
+may replay a row at an ambiguous boundary to avoid losing it.
+
### Why does PostgreSQL CDC fall behind?
Replication lag can occur when the logical decoding plugin is slow or when the
WAL sender is under load. Monitor `pg_replication_slots` for
`confirmed_flush_lsn` drift. Ensure the CDC job consumes events continuously
and that network latency between SeaTunnel and PostgreSQL is low.
diff --git a/docs/en/engines/zeta/user-command.md
b/docs/en/engines/zeta/user-command.md
index 8ecd35b706..3ddb44734f 100644
--- a/docs/en/engines/zeta/user-command.md
+++ b/docs/en/engines/zeta/user-command.md
@@ -50,6 +50,11 @@ Usage: seatunnel.sh [options]
sh bin/seatunnel.sh --config $SEATUNNEL_HOME/config/v2.batch.config.template
```
+Without `--async`, the client waits for the final job result. If a job keeps
failing during
+task startup or automatic recovery, for example because a Paimon source lacks
`SELECT` permission, the job
+returns `FAILED` after exhausting the configured retries and the client exits
with a nonzero
+status.
+
The **--async** parameter allows the job to run in the background. When the
job is submitted, the client will exit.
```shell
diff --git a/docs/zh/connectors/source/PostgreSQL-CDC.md
b/docs/zh/connectors/source/PostgreSQL-CDC.md
index 732d6657f8..cc3459a631 100644
--- a/docs/zh/connectors/source/PostgreSQL-CDC.md
+++ b/docs/zh/connectors/source/PostgreSQL-CDC.md
@@ -376,6 +376,12 @@ SeaTunnel 在任务启动时会创建或复用 `slot.name` 指定的复制槽。
作为启动偏移量。未使用的复制槽会持续占用磁盘上的 WAL 段，导致 WAL 持续增长。当 CDC
任务永久下线时，应在 PostgreSQL 侧手动删除不再使用的复制槽。
+### 从保存点恢复时如何处理事务边界？
+
+恢复时会区分保存的事务提交记录与已处理的数据行，因为事务提交和下一事务的第一条数据行
+可能具有相同的 LSN。任务停止期间提交的数据仍可被重放。不包含操作类型的旧保存点仍可读取；
+为避免丢失数据，恢复时可能重放无法明确区分的边界记录。
+
### PostgreSQL CDC 为什么会滞后？
滞后可能由逻辑解码插件处理慢或 WAL sender 负载过高引起。可通过监控 `pg_replication_slots` 中的
`confirmed_flush_lsn` 漂移情况来排查。确保 CDC 任务持续消费事件，并保持 SeaTunnel 与 PostgreSQL
之间的网络低延迟。
diff --git a/docs/zh/engines/zeta/user-command.md
b/docs/zh/engines/zeta/user-command.md
index d5d9e917c0..87f66b5493 100644
--- a/docs/zh/engines/zeta/user-command.md
+++ b/docs/zh/engines/zeta/user-command.md
@@ -66,6 +66,9 @@ Usage: seatunnel.sh [options]
bin/seatunnel.sh --config $SEATUNNEL_HOME/config/v2.batch.config.template
```
+未使用 `--async` 时，客户端会等待作业的最终结果。如果作业在任务启动或自动恢复期间持续失败，例如
+Paimon source 缺少 `SELECT` 权限，达到配置的重试上限后，作业会返回 `FAILED`，客户端以非零状态退出。
+
**--async** 参数可以让作业在后台运行，当作业提交后，客户端会退出。
```shell
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/io/debezium/connector/postgresql/PostgresOffsetContext.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/io/debezium/connector/postgresql/PostgresOffsetContext.java
index bb7d1d4487..599180154b 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/io/debezium/connector/postgresql/PostgresOffsetContext.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/io/debezium/connector/postgresql/PostgresOffsetContext.java
@@ -27,6 +27,7 @@ import org.slf4j.LoggerFactory;
import io.debezium.connector.SnapshotRecord;
import io.debezium.connector.postgresql.connection.Lsn;
import io.debezium.connector.postgresql.connection.PostgresConnection;
+import
io.debezium.connector.postgresql.connection.ReplicationMessage.Operation;
import io.debezium.connector.postgresql.spi.OffsetState;
import
io.debezium.pipeline.source.snapshot.incremental.IncrementalSnapshotContext;
import
io.debezium.pipeline.source.snapshot.incremental.SignalBasedIncrementalSnapshotContext;
@@ -54,12 +55,14 @@ public class PostgresOffsetContext implements OffsetContext
{
public static final String LAST_COMPLETELY_PROCESSED_LSN_KEY = "lsn_proc";
public static final String LAST_COMMIT_LSN_KEY = "lsn_commit";
+ public static final String LAST_PROCESSED_MESSAGE_TYPE_KEY = "messageType";
private final Schema sourceInfoSchema;
private final SourceInfo sourceInfo;
private boolean lastSnapshotRecord;
private Lsn lastCompletelyProcessedLsn;
private Lsn lastCommitLsn;
+ private Operation lastProcessedMessageType;
private Lsn streamingStoppingLsn = null;
private final TransactionContext transactionContext;
private final IncrementalSnapshotContext<TableId>
incrementalSnapshotContext;
@@ -120,6 +123,9 @@ public class PostgresOffsetContext implements OffsetContext
{
if (lastCommitLsn != null) {
result.put(LAST_COMMIT_LSN_KEY, lastCommitLsn.asLong());
}
+ if (lastProcessedMessageType != null) {
+ result.put(LAST_PROCESSED_MESSAGE_TYPE_KEY,
lastProcessedMessageType.name());
+ }
return sourceInfo.isSnapshot()
? result
:
incrementalSnapshotContext.store(transactionContext.store(result));
@@ -163,7 +169,20 @@ public class PostgresOffsetContext implements
OffsetContext {
Long txId,
Long xmin,
TableId tableId) {
+ updateWalPosition(lsn, lastCompletelyProcessedLsn, commitTime, txId,
xmin, tableId, null);
+ }
+
+ /** Tracks the operation as well as its LSN so recovery can distinguish a
commit boundary. */
+ public void updateWalPosition(
+ Lsn lsn,
+ Lsn lastCompletelyProcessedLsn,
+ Instant commitTime,
+ Long txId,
+ Long xmin,
+ TableId tableId,
+ Operation messageType) {
this.lastCompletelyProcessedLsn = lastCompletelyProcessedLsn;
+ this.lastProcessedMessageType = messageType;
sourceInfo.update(lsn, commitTime, txId, xmin, tableId);
}
@@ -176,6 +195,9 @@ public class PostgresOffsetContext implements OffsetContext
{
public void updateCommitPosition(Lsn lsn, Lsn lastCompletelyProcessedLsn) {
this.lastCompletelyProcessedLsn = lastCompletelyProcessedLsn;
this.lastCommitLsn = lsn;
+ // Without transaction metadata, COMMIT bypasses updateWalPosition.
Its operation must
+ // still match lsn_proc in the heartbeat offset that a checkpoint can
persist.
+ this.lastProcessedMessageType = Operation.COMMIT;
sourceInfo.updateLastCommit(lsn);
}
@@ -199,6 +221,10 @@ public class PostgresOffsetContext implements
OffsetContext {
return lastCommitLsn;
}
+ Operation lastProcessedMessageType() {
+ return lastProcessedMessageType;
+ }
+
/**
* Returns the LSN that the streaming phase should stream events up to or
null if a stopping
* point is not set. If set during the streaming phase, any event with an
LSN less than the
@@ -264,17 +290,23 @@ public class PostgresOffsetContext implements
OffsetContext {
((Map<String, Object>) offset)
.getOrDefault(
SourceInfo.LAST_SNAPSHOT_RECORD_KEY, Boolean.FALSE);
- return new PostgresOffsetContext(
- connectorConfig,
- lsn,
- lastCompletelyProcessedLsn,
- lastCommitLsn,
- txId,
- useconds,
- snapshot,
- lastSnapshotRecord,
- TransactionContext.load(offset),
- SignalBasedIncrementalSnapshotContext.load(offset, false));
+ PostgresOffsetContext context =
+ new PostgresOffsetContext(
+ connectorConfig,
+ lsn,
+ lastCompletelyProcessedLsn,
+ lastCommitLsn,
+ txId,
+ useconds,
+ snapshot,
+ lastSnapshotRecord,
+ TransactionContext.load(offset),
+ SignalBasedIncrementalSnapshotContext.load(offset,
false));
+ Object messageType = offset.get(LAST_PROCESSED_MESSAGE_TYPE_KEY);
+ if (messageType != null) {
+ context.lastProcessedMessageType =
Operation.valueOf(messageType.toString());
+ }
+ return context;
}
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/io/debezium/connector/postgresql/PostgresStreamingChangeEventSource.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/io/debezium/connector/postgresql/PostgresStreamingChangeEventSource.java
new file mode 100644
index 0000000000..e60e8cc195
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/io/debezium/connector/postgresql/PostgresStreamingChangeEventSource.java
@@ -0,0 +1,552 @@
+/*
+ * 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 io.debezium.connector.postgresql;
+
+import org.apache.kafka.connect.errors.ConnectException;
+
+import org.postgresql.core.BaseConnection;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.debezium.DebeziumException;
+import io.debezium.connector.postgresql.connection.LogicalDecodingMessage;
+import io.debezium.connector.postgresql.connection.Lsn;
+import io.debezium.connector.postgresql.connection.PostgresConnection;
+import io.debezium.connector.postgresql.connection.ReplicationConnection;
+import
io.debezium.connector.postgresql.connection.ReplicationMessage.Operation;
+import io.debezium.connector.postgresql.connection.ReplicationStream;
+import io.debezium.connector.postgresql.connection.WalPositionLocator;
+import io.debezium.connector.postgresql.spi.Snapshotter;
+import io.debezium.heartbeat.Heartbeat;
+import io.debezium.pipeline.ErrorHandler;
+import io.debezium.pipeline.source.spi.StreamingChangeEventSource;
+import io.debezium.relational.TableId;
+import io.debezium.util.Clock;
+import io.debezium.util.DelayStrategy;
+import io.debezium.util.ElapsedTimeStrategy;
+import io.debezium.util.Threads;
+
+import java.sql.SQLException;
+import java.util.Map;
+import java.util.Objects;
+import java.util.OptionalLong;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Debezium 1.9.8 streaming implementation with the DBZ-5915/DBZ-6204 resume
operation passed
+ * through the offset context. Other streaming and offset-flush behavior is
unchanged.
+ *
+ * <p>Copyright Debezium Authors. Adapted from Debezium 1.9.8.Final.
+ *
+ * @author Horia Chiorean ([email protected]), Jiri Pechanec
+ */
+public class PostgresStreamingChangeEventSource
+ implements StreamingChangeEventSource<PostgresPartition,
PostgresOffsetContext> {
+
+ private static final String KEEP_ALIVE_THREAD_NAME = "keep-alive";
+
+ /**
+ * Number of received events without sending anything to Kafka which will
trigger a "WAL backlog
+ * growing" warning.
+ */
+ private static final int GROWING_WAL_WARNING_LOG_INTERVAL = 10_000;
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(PostgresStreamingChangeEventSource.class);
+
+ // PGOUTPUT decoder sends the messages with larger time gaps than other
decoders
+ // We thus try to read the message multiple times before we make poll pause
+ private static final int THROTTLE_NO_MESSAGE_BEFORE_PAUSE = 5;
+
+ private final PostgresConnection connection;
+ private final PostgresEventDispatcher<TableId> dispatcher;
+ private final ErrorHandler errorHandler;
+ private final Clock clock;
+ private final PostgresSchema schema;
+ private final PostgresConnectorConfig connectorConfig;
+ private final PostgresTaskContext taskContext;
+ private final ReplicationConnection replicationConnection;
+ private final AtomicReference<ReplicationStream> replicationStream = new
AtomicReference<>();
+ private final Snapshotter snapshotter;
+ private final DelayStrategy pauseNoMessage;
+ private final ElapsedTimeStrategy connectionProbeTimer;
+
+ // Offset committing is an asynchronous operation.
+ // When connector is restarted we cannot be sure about timing of recovery,
offset committing
+ // etc.
+ // as this is driven by Kafka Connect. This might be a root cause of
DBZ-5163.
+ // This flag will ensure that LSN is flushed only if we are really in
message processing mode.
+ private volatile boolean lsnFlushingAllowed = false;
+
+ /**
+ * The minimum of (number of event received since the last event sent to
Kafka, number of event
+ * received since last WAL growing warning issued).
+ */
+ private long numberOfEventsSinceLastEventSentOrWalGrowingWarning = 0;
+
+ private Lsn lastCompletelyProcessedLsn;
+
+ public PostgresStreamingChangeEventSource(
+ PostgresConnectorConfig connectorConfig,
+ Snapshotter snapshotter,
+ PostgresConnection connection,
+ PostgresEventDispatcher<TableId> dispatcher,
+ ErrorHandler errorHandler,
+ Clock clock,
+ PostgresSchema schema,
+ PostgresTaskContext taskContext,
+ ReplicationConnection replicationConnection) {
+ this.connectorConfig = connectorConfig;
+ this.connection = connection;
+ this.dispatcher = dispatcher;
+ this.errorHandler = errorHandler;
+ this.clock = clock;
+ this.schema = schema;
+ pauseNoMessage =
+
DelayStrategy.constant(taskContext.getConfig().getPollInterval().toMillis());
+ this.taskContext = taskContext;
+ this.snapshotter = snapshotter;
+ this.replicationConnection = replicationConnection;
+ this.connectionProbeTimer =
+ ElapsedTimeStrategy.constant(
+ Clock.system(),
connectorConfig.statusUpdateInterval());
+ }
+
+ @Override
+ public void init() {
+ // refresh the schema so we have a latest view of the DB tables
+ try {
+ taskContext.refreshSchema(connection, true);
+ } catch (SQLException e) {
+ throw new DebeziumException("Error while executing initial schema
load", e);
+ }
+ }
+
+ @Override
+ public void execute(
+ ChangeEventSourceContext context,
+ PostgresPartition partition,
+ PostgresOffsetContext offsetContext)
+ throws InterruptedException {
+ if (!snapshotter.shouldStream()) {
+ LOGGER.info("Streaming is not enabled in correct configuration");
+ return;
+ }
+
+ lsnFlushingAllowed = false;
+
+ // replication slot could exist at the time of starting Debezium so we
will stream from the
+ // position in the slot
+ // instead of the last position in the database
+ boolean hasStartLsnStoredInContext = offsetContext != null;
+
+ if (!hasStartLsnStoredInContext) {
+ offsetContext =
+ PostgresOffsetContext.initialContext(connectorConfig,
connection, clock);
+ }
+
+ try {
+ final WalPositionLocator walPosition;
+
+ if (hasStartLsnStoredInContext) {
+ // start streaming from the last recorded position in the
offset
+ final Lsn lsn =
+ offsetContext.lastCompletelyProcessedLsn() != null
+ ? offsetContext.lastCompletelyProcessedLsn()
+ : offsetContext.lsn();
+ LOGGER.info("Retrieved latest position from stored offset
'{}'", lsn);
+ walPosition =
+ new WalPositionLocator(
+ offsetContext.lastCommitLsn(),
+ lsn,
+ offsetContext.lastProcessedMessageType());
+ replicationStream.compareAndSet(
+ null, replicationConnection.startStreaming(lsn,
walPosition));
+ } else {
+ LOGGER.info(
+ "No previous LSN found in Kafka, streaming from the
latest xlogpos or flushed LSN...");
+ walPosition = new WalPositionLocator();
+ replicationStream.compareAndSet(
+ null,
replicationConnection.startStreaming(walPosition));
+ }
+ // for large dbs, the refresh of schema can take too much time
+ // such that the connection times out. We must enable keep
+ // alive to ensure that it doesn't time out
+ ReplicationStream stream = this.replicationStream.get();
+ stream.startKeepAlive(
+ Threads.newSingleThreadExecutor(
+ PostgresConnector.class,
+ connectorConfig.getLogicalName(),
+ KEEP_ALIVE_THREAD_NAME));
+
+ init();
+
+ // If we need to do a pre-snapshot streaming catch up, we should
allow the snapshot
+ // transaction to persist
+ // but normally we want to start streaming without any open
transactions.
+ if (!isInPreSnapshotCatchUpStreaming(offsetContext)) {
+ connection.commit();
+ }
+
+ this.lastCompletelyProcessedLsn =
replicationStream.get().startLsn();
+
+ if (walPosition.searchingEnabled()) {
+ searchWalPosition(context, stream, walPosition);
+ try {
+ if (!isInPreSnapshotCatchUpStreaming(offsetContext)) {
+ connection.commit();
+ }
+ } catch (Exception e) {
+ LOGGER.info("Commit failed while preparing for reconnect",
e);
+ }
+ walPosition.enableFiltering();
+ stream.stopKeepAlive();
+ replicationConnection.reconnect();
+ replicationStream.set(
+ replicationConnection.startStreaming(
+ walPosition.getLastEventStoredLsn(),
walPosition));
+ stream = this.replicationStream.get();
+ stream.startKeepAlive(
+ Threads.newSingleThreadExecutor(
+ PostgresConnector.class,
+ connectorConfig.getLogicalName(),
+ KEEP_ALIVE_THREAD_NAME));
+ }
+ processMessages(context, partition, offsetContext, stream);
+ } catch (Throwable e) {
+ errorHandler.setProducerThrowable(e);
+ } finally {
+ if (replicationConnection != null) {
+ LOGGER.debug("stopping streaming...");
+ // stop the keep alive thread, this also shuts down the
+ // executor pool
+ ReplicationStream stream = replicationStream.get();
+ if (stream != null) {
+ stream.stopKeepAlive();
+ }
+ // TODO author=Horia Chiorean date=08/11/2016
description=Ideally we'd close the
+ // stream, but it's not reliable atm (see javadoc)
+ // replicationStream.close();
+ // close the connection - this should also disconnect the
current stream even if
+ // it's blocking
+ try {
+ if (!isInPreSnapshotCatchUpStreaming(offsetContext)) {
+ connection.commit();
+ }
+ replicationConnection.close();
+ } catch (Exception e) {
+ LOGGER.debug("Exception while closing the connection", e);
+ }
+ replicationStream.set(null);
+ }
+ }
+ }
+
+ private void processMessages(
+ ChangeEventSourceContext context,
+ PostgresPartition partition,
+ PostgresOffsetContext offsetContext,
+ final ReplicationStream stream)
+ throws SQLException, InterruptedException {
+ LOGGER.info("Processing messages");
+ int noMessageIterations = 0;
+ while (context.isRunning()
+ && (offsetContext.getStreamingStoppingLsn() == null
+ || (lastCompletelyProcessedLsn.compareTo(
+
offsetContext.getStreamingStoppingLsn())
+ < 0))) {
+
+ boolean receivedMessage =
+ stream.readPending(
+ message -> {
+ final Lsn lsn = stream.lastReceivedLsn();
+
+ if (message.isLastEventForLsn()) {
+ lastCompletelyProcessedLsn = lsn;
+ }
+
+ // Tx BEGIN/END event
+ if (message.isTransactionalMessage()) {
+ if
(!connectorConfig.shouldProvideTransactionMetadata()) {
+ LOGGER.trace("Received transactional
message {}", message);
+ // Don't skip on BEGIN message as it
would flush LSN for the
+ // whole transaction
+ // too early
+ if (message.getOperation() ==
Operation.COMMIT) {
+ commitMessage(partition,
offsetContext, lsn);
+ }
+ return;
+ }
+
+ offsetContext.updateWalPosition(
+ lsn,
+ lastCompletelyProcessedLsn,
+ message.getCommitTime(),
+ toLong(message.getTransactionId()),
+
taskContext.getSlotXmin(connection),
+ null,
+ message.getOperation());
+ if (message.getOperation() ==
Operation.BEGIN) {
+
dispatcher.dispatchTransactionStartedEvent(
+ partition,
+
toString(message.getTransactionId()),
+ offsetContext);
+ } else if (message.getOperation() ==
Operation.COMMIT) {
+ commitMessage(partition,
offsetContext, lsn);
+
dispatcher.dispatchTransactionCommittedEvent(
+ partition, offsetContext);
+ }
+ maybeWarnAboutGrowingWalBacklog(true);
+ } else if (message.getOperation() ==
Operation.MESSAGE) {
+ offsetContext.updateWalPosition(
+ lsn,
+ lastCompletelyProcessedLsn,
+ message.getCommitTime(),
+ toLong(message.getTransactionId()),
+
taskContext.getSlotXmin(connection),
+ null,
+ message.getOperation());
+
+ // non-transactional message that will not
be followed by a
+ // COMMIT message
+ if (message.isLastEventForLsn()) {
+ commitMessage(partition,
offsetContext, lsn);
+ }
+
+ dispatcher.dispatchLogicalDecodingMessage(
+ partition,
+ offsetContext,
+
clock.currentTimeAsInstant().toEpochMilli(),
+ (LogicalDecodingMessage) message);
+
+ maybeWarnAboutGrowingWalBacklog(true);
+ }
+ // DML event
+ else {
+ TableId tableId = null;
+ if (message.getOperation() !=
Operation.NOOP) {
+ tableId =
PostgresSchema.parse(message.getTable());
+ Objects.requireNonNull(tableId);
+ }
+
+ offsetContext.updateWalPosition(
+ lsn,
+ lastCompletelyProcessedLsn,
+ message.getCommitTime(),
+ toLong(message.getTransactionId()),
+
taskContext.getSlotXmin(connection),
+ tableId,
+ message.getOperation());
+
+ boolean dispatched =
+ message.getOperation() !=
Operation.NOOP
+ &&
dispatcher.dispatchDataChangeEvent(
+ partition,
+ tableId,
+ new
PostgresChangeRecordEmitter(
+ partition,
+
offsetContext,
+ clock,
+
connectorConfig,
+ schema,
+ connection,
+ tableId,
+ message));
+
+
maybeWarnAboutGrowingWalBacklog(dispatched);
+ }
+ });
+
+ probeConnectionIfNeeded();
+
+ if (receivedMessage) {
+ noMessageIterations = 0;
+ lsnFlushingAllowed = true;
+ } else {
+ if (offsetContext.hasCompletelyProcessedPosition()) {
+ dispatcher.dispatchHeartbeatEvent(partition,
offsetContext);
+ }
+ noMessageIterations++;
+ if (noMessageIterations >= THROTTLE_NO_MESSAGE_BEFORE_PAUSE) {
+ noMessageIterations = 0;
+ pauseNoMessage.sleepWhen(true);
+ }
+ }
+ if (!isInPreSnapshotCatchUpStreaming(offsetContext)) {
+ // During catch up streaming, the streaming phase needs to
hold a transaction open
+ // so that
+ // the phase can stream event up to a specific lsn and the
snapshot that occurs
+ // after the catch up
+ // streaming will not lose the current view of data. Since we
need to hold the
+ // transaction open
+ // for the snapshot, this block must not commit during catch
up streaming.
+ connection.commit();
+ }
+ }
+ }
+
+ private void searchWalPosition(
+ ChangeEventSourceContext context,
+ final ReplicationStream stream,
+ final WalPositionLocator walPosition)
+ throws SQLException, InterruptedException {
+ AtomicReference<Lsn> resumeLsn = new AtomicReference<>();
+ int noMessageIterations = 0;
+
+ LOGGER.info("Searching for WAL resume position");
+ while (context.isRunning() && resumeLsn.get() == null) {
+
+ boolean receivedMessage =
+ stream.readPending(
+ message -> {
+ final Lsn lsn = stream.lastReceivedLsn();
+ resumeLsn.set(walPosition.resumeFromLsn(lsn,
message).orElse(null));
+ });
+
+ if (receivedMessage) {
+ noMessageIterations = 0;
+ } else {
+ noMessageIterations++;
+ if (noMessageIterations >= THROTTLE_NO_MESSAGE_BEFORE_PAUSE) {
+ noMessageIterations = 0;
+ pauseNoMessage.sleepWhen(true);
+ }
+ }
+
+ probeConnectionIfNeeded();
+ }
+ LOGGER.info("WAL resume position '{}' discovered", resumeLsn.get());
+ }
+
+ private void probeConnectionIfNeeded() throws SQLException {
+ if (connectionProbeTimer.hasElapsed()) {
+ connection.prepareQuery("SELECT 1");
+ connection.commit();
+ }
+ }
+
+ private void commitMessage(
+ PostgresPartition partition, PostgresOffsetContext offsetContext,
final Lsn lsn)
+ throws SQLException, InterruptedException {
+ lastCompletelyProcessedLsn = lsn;
+ offsetContext.updateCommitPosition(lsn, lastCompletelyProcessedLsn);
+ maybeWarnAboutGrowingWalBacklog(false);
+ dispatcher.dispatchHeartbeatEvent(partition, offsetContext);
+ }
+
+ /**
+ * If we receive change events but all of them get filtered out, we cannot
commit any new offset
+ * with Apache Kafka. This in turn means no LSN is ever acknowledged with
the replication slot,
+ * causing any ever growing WAL backlog.
+ *
+ * <p>This situation typically occurs if there are changes on the database
server, (e.g. in an
+ * excluded database), but none of them is in table.include.list. To
prevent this, heartbeats
+ * can be used, as they will allow us to commit offsets also when not
propagating any "real"
+ * change event.
+ *
+ * <p>The purpose of this method is to detect this situation and log a
warning every {@link
+ * #GROWING_WAL_WARNING_LOG_INTERVAL} filtered events.
+ *
+ * @param dispatched Whether an event was sent to the broker or not
+ */
+ private void maybeWarnAboutGrowingWalBacklog(boolean dispatched) {
+ if (dispatched) {
+ numberOfEventsSinceLastEventSentOrWalGrowingWarning = 0;
+ } else {
+ numberOfEventsSinceLastEventSentOrWalGrowingWarning++;
+ }
+
+ if (numberOfEventsSinceLastEventSentOrWalGrowingWarning >
GROWING_WAL_WARNING_LOG_INTERVAL
+ && !dispatcher.heartbeatsEnabled()) {
+ LOGGER.warn(
+ "Received {} events which were all filtered out, so no
offset could be committed. "
+ + "This prevents the replication slot from
acknowledging the processed WAL offsets, "
+ + "causing a growing backlog of non-removeable WAL
segments on the database server. "
+ + "Consider to either adjust your filter
configuration or enable heartbeat events "
+ + "(via the {} option) to avoid this situation.",
+ numberOfEventsSinceLastEventSentOrWalGrowingWarning,
+ Heartbeat.HEARTBEAT_INTERVAL_PROPERTY_NAME);
+
+ numberOfEventsSinceLastEventSentOrWalGrowingWarning = 0;
+ }
+ }
+
+ @Override
+ public void commitOffset(Map<String, ?> offset) {
+ try {
+ ReplicationStream replicationStream = this.replicationStream.get();
+ final Lsn commitLsn =
+ Lsn.valueOf((Long)
offset.get(PostgresOffsetContext.LAST_COMMIT_LSN_KEY));
+ final Lsn changeLsn =
+ Lsn.valueOf(
+ (Long)
+ offset.get(
+ PostgresOffsetContext
+
.LAST_COMPLETELY_PROCESSED_LSN_KEY));
+ final Lsn lsn = (commitLsn != null) ? commitLsn : changeLsn;
+
+ if (replicationStream != null && lsn != null) {
+ if (!lsnFlushingAllowed) {
+ LOGGER.info(
+ "Received offset commit request on '{}', but
ignoring it. LSN flushing is not allowed yet",
+ lsn);
+ return;
+ }
+
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("Flushing LSN to server: {}", lsn);
+ }
+ // tell the server the point up to which we've processed data,
so it can be free to
+ // recycle WAL segments
+ replicationStream.flushLsn(lsn);
+ } else {
+ LOGGER.debug("Streaming has already stopped, ignoring commit
callback...");
+ }
+ } catch (SQLException e) {
+ throw new ConnectException(e);
+ }
+ }
+
+ /**
+ * Returns whether the current streaming phase is running a catch up
streaming phase that runs
+ * before a snapshot. This is useful for transaction management.
+ *
+ * <p>During pre-snapshot catch up streaming, we open the snapshot
transaction early and hold
+ * the transaction open throughout the pre snapshot catch up streaming
phase so that we know
+ * where to stop streaming and can start the snapshot phase at a
consistent location. This is
+ * opposed the regular streaming, where we do not a lingering open
transaction.
+ *
+ * @return true if the current streaming phase is performing catch up
streaming
+ */
+ private boolean isInPreSnapshotCatchUpStreaming(PostgresOffsetContext
offsetContext) {
+ return offsetContext.getStreamingStoppingLsn() != null;
+ }
+
+ private Long toLong(OptionalLong l) {
+ return l.isPresent() ? Long.valueOf(l.getAsLong()) : null;
+ }
+
+ private String toString(OptionalLong l) {
+ return l.isPresent() ? String.valueOf(l.getAsLong()) : null;
+ }
+
+ @FunctionalInterface
+ public static interface PgConnectionSupplier {
+ BaseConnection get() throws SQLException;
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/io/debezium/connector/postgresql/connection/WalPositionLocator.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/io/debezium/connector/postgresql/connection/WalPositionLocator.java
new file mode 100644
index 0000000000..a4518bc830
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/io/debezium/connector/postgresql/connection/WalPositionLocator.java
@@ -0,0 +1,230 @@
+/*
+ * 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 io.debezium.connector.postgresql.connection;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.debezium.DebeziumException;
+import
io.debezium.connector.postgresql.connection.ReplicationMessage.Operation;
+
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
+
+/**
+ * This class is responsible for finding out a LSN from which Debezium should
resume streaming after
+ * connector restarts. The LSNs are not guaranteed to be ordered in the WAL.
LSN of commits are
+ * ordered and LSNs inside a transaction are ordered. It is thus necessary to
find out the beginning
+ * of the unprocessed transaction and the LSN from which the streaming should
start. Backports
+ * DBZ-5915 and DBZ-6204 to the bundled Debezium 1.9.8 implementation.
+ *
+ * <p>Copyright Debezium Authors. Adapted from Debezium 1.9.8.Final.
+ *
+ * @author Jiri Pechanec
+ */
+public class WalPositionLocator {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(WalPositionLocator.class);
+
+ private final Lsn lastCommitStoredLsn;
+ private final Lsn lastEventStoredLsn;
+ private final Operation lastProcessedMessageType;
+ private Lsn txStartLsn = null;
+ private Lsn lsnAfterLastEventStoredLsn = null;
+ private Lsn firstLsnReceived = null;
+ private boolean passMessages = true;
+ private Lsn startStreamingLsn = null;
+ private boolean storeLsnAfterLastEventStoredLsn = false;
+ private Set<Lsn> lsnSeen = new HashSet<>(1_000);
+
+ public WalPositionLocator(Lsn lastCommitStoredLsn, Lsn lastEventStoredLsn)
{
+ this(lastCommitStoredLsn, lastEventStoredLsn, null);
+ }
+
+ public WalPositionLocator(
+ Lsn lastCommitStoredLsn, Lsn lastEventStoredLsn, Operation
lastProcessedMessageType) {
+ this.lastCommitStoredLsn = lastCommitStoredLsn;
+ this.lastEventStoredLsn = lastEventStoredLsn;
+ this.lastProcessedMessageType = lastProcessedMessageType;
+
+ LOGGER.info(
+ "Looking for WAL restart position for last commit LSN '{}' and
last change LSN '{}'",
+ lastCommitStoredLsn,
+ lastEventStoredLsn);
+ }
+
+ public WalPositionLocator() {
+ this.lastCommitStoredLsn = null;
+ this.lastEventStoredLsn = null;
+ this.lastProcessedMessageType = null;
+
+ LOGGER.info("WAL position will not be searched");
+ }
+
+ /**
+ * @return the first LSN from which processing should be started or empty
if the position has
+ * not been found yet
+ */
+ public Optional<Lsn> resumeFromLsn(Lsn currentLsn, ReplicationMessage
message) {
+ LOGGER.trace("Processing LSN '{}', operation '{}'", currentLsn,
message.getOperation());
+
+ lsnSeen.add(currentLsn);
+
+ if (firstLsnReceived == null) {
+ firstLsnReceived = currentLsn;
+ LOGGER.info("First LSN '{}' received", firstLsnReceived);
+ }
+ if (storeLsnAfterLastEventStoredLsn) {
+ // Event that immediately follows the last event seen
+ // We can resume streaming from it
+ if (currentLsn.equals(lastEventStoredLsn)) {
+ // BEGIN and first message after change have the same LSN
+ if (txStartLsn != null
+ && (lastProcessedMessageType == null
+ || lastProcessedMessageType == Operation.BEGIN
+ || lastProcessedMessageType ==
Operation.COMMIT)) {
+ // The previous COMMIT can share its LSN with this BEGIN
and INSERT.
+ // Replay this transaction unless the stored operation was
already a row.
+ startStreamingLsn = txStartLsn;
+ return Optional.of(startStreamingLsn);
+ }
+ return Optional.empty();
+ }
+ lsnAfterLastEventStoredLsn = currentLsn;
+ storeLsnAfterLastEventStoredLsn = false;
+ LOGGER.info(
+ "LSN after last stored change LSN '{}' received",
lsnAfterLastEventStoredLsn);
+ startStreamingLsn = lsnAfterLastEventStoredLsn;
+ return Optional.of(startStreamingLsn);
+ }
+ if (currentLsn.equals(lastEventStoredLsn)) {
+ storeLsnAfterLastEventStoredLsn = true;
+ }
+
+ if (lastCommitStoredLsn == null) {
+ startStreamingLsn = firstLsnReceived;
+ return Optional.of(startStreamingLsn);
+ }
+
+ switch (message.getOperation()) {
+ case BEGIN:
+ txStartLsn = currentLsn;
+ break;
+ case COMMIT:
+ if (currentLsn.compareTo(lastCommitStoredLsn) > 0) {
+ LOGGER.info(
+ "Received COMMIT LSN '{}' larger than than last
stored commit LSN '{}'",
+ currentLsn,
+ lastCommitStoredLsn);
+ if (lsnAfterLastEventStoredLsn != null) {
+ startStreamingLsn = lsnAfterLastEventStoredLsn;
+ LOGGER.info(
+ "Will restart from LSN '{}' that follows the
lastest stored",
+ startStreamingLsn);
+ return Optional.of(startStreamingLsn);
+ } else if (txStartLsn != null) {
+ // Transaction that has not been processed and at the
same
+ // time the event following the last seen was not met
so
+ // replaying from the start of transaction
+ startStreamingLsn = txStartLsn;
+ LOGGER.info(
+ "Will restart from LSN '{}' that is start of
the first unprocessed transaction",
+ startStreamingLsn);
+ return Optional.of(startStreamingLsn);
+ } else {
+ // Transaction that has not been processed and at the
same
+ // time the event following the last seen was not met
and
+ // transaction start event has not been met so
replaying
+ // from the very first event
+ startStreamingLsn = firstLsnReceived;
+ LOGGER.info(
+ "Will restart from LSN '{}' that is the first
LSN available",
+ startStreamingLsn);
+ return Optional.of(startStreamingLsn);
+ }
+ }
+ break;
+ default:
+ break;
+ }
+ return Optional.empty();
+ }
+
+ /**
+ * Decides whether the message with given LSN should be removed or not
based on previously
+ * located LSN point.
+ *
+ * @param lsn
+ * @return true if the message should be skipped, false otherwise
+ */
+ public boolean skipMessage(Lsn lsn) {
+ if (passMessages) {
+ return false;
+ }
+ if (startStreamingLsn == null || startStreamingLsn.equals(lsn)) {
+ LOGGER.info("Message with LSN '{}' arrived, switching off the
filtering", lsn);
+ passMessages = true;
+ lsnSeen = new HashSet<>(); // Empty the Map as it might be large
and is no longer needed
+ return false;
+ }
+ if (lsn.isValid() && !lsnSeen.contains(lsn)) {
+ throw new DebeziumException(
+ String.format(
+ "Message with LSN '%s' not present among LSNs seen
in the location phase '%s'. This is unexpected and can lead to an infinite loop
or a data loss.",
+ lsn, lsnSeen));
+ }
+ LOGGER.debug("Message with LSN '{}' filtered", lsn);
+ return true;
+ }
+
+ /** Enables filtering of message LSNs based on calculated position. */
+ public void enableFiltering() {
+ passMessages = false;
+ }
+
+ /** @return true if searching of WAL position should be executed */
+ public boolean searchingEnabled() {
+ return lastEventStoredLsn != null;
+ }
+
+ public Lsn getLastEventStoredLsn() {
+ return lastEventStoredLsn;
+ }
+
+ @Override
+ public String toString() {
+ return "WalPositionLocator [lastCommitStoredLsn="
+ + lastCommitStoredLsn
+ + ", lastEventStoredLsn="
+ + lastEventStoredLsn
+ + ", txStartLsn="
+ + txStartLsn
+ + ", lsnAfterLastEventStoredLsn="
+ + lsnAfterLastEventStoredLsn
+ + ", firstLsnReceived="
+ + firstLsnReceived
+ + ", passMessages="
+ + passMessages
+ + ", startStreamingLsn="
+ + startStreamingLsn
+ + ", storeLsnAfterLastEventStoredLsn="
+ + storeLsnAfterLastEventStoredLsn
+ + "]";
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/reader/PostgresSourceFetchTaskContext.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/reader/PostgresSourceFetchTaskContext.java
index 71b707d4db..468550b006 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/reader/PostgresSourceFetchTaskContext.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/reader/PostgresSourceFetchTaskContext.java
@@ -460,12 +460,16 @@ public class PostgresSourceFetchTaskContext extends
JdbcSourceFetchTaskContext {
Map<String, String> offsetStrMap =
Objects.requireNonNull(offset, "offset is null for the
sourceSplitBase")
.getOffset();
- // all the keys happen to be long type for
PostgresOffsetContext.Loader.load
+ // Numeric offsets are stored as strings in split state; the resume
operation is an enum.
Map<String, Object> offsetMap = new HashMap<>();
for (String key : offsetStrMap.keySet()) {
String value = offsetStrMap.get(key);
if (value != null) {
- offsetMap.put(key, Long.parseLong(value));
+ offsetMap.put(
+ key,
+
PostgresOffsetContext.LAST_PROCESSED_MESSAGE_TYPE_KEY.equals(key)
+ ? value
+ : Long.parseLong(value));
}
}
return loader.load(offsetMap);
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/offset/LsnOffsetTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/offset/LsnOffsetTest.java
index f0ca36e6da..d1c2b21c77 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/offset/LsnOffsetTest.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/offset/LsnOffsetTest.java
@@ -20,12 +20,23 @@ package
org.apache.seatunnel.connectors.seatunnel.cdc.postgres.source.offset;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import io.debezium.config.Configuration;
+import io.debezium.connector.postgresql.PostgresConnectorConfig;
import io.debezium.connector.postgresql.PostgresOffsetContext;
import io.debezium.connector.postgresql.SourceInfo;
import io.debezium.connector.postgresql.connection.Lsn;
+import io.debezium.connector.postgresql.connection.ReplicationMessage;
+import
io.debezium.connector.postgresql.connection.ReplicationMessage.Operation;
+import io.debezium.connector.postgresql.connection.WalPositionLocator;
+import io.debezium.relational.TableId;
+import java.time.Instant;
import java.util.HashMap;
import java.util.Map;
+import java.util.Optional;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
class LsnOffsetTest {
@@ -65,4 +76,107 @@ class LsnOffsetTest {
Lsn lsnCommit = offset.getLsnCommit();
Assertions.assertEquals(Lsn.valueOf(12345L), lsnCommit);
}
+
+ @Test
+ void testRestoreCommitDoesNotSkipInsertAtSameLsn() {
+ assertFirstInsertReplayed(Operation.COMMIT);
+ }
+
+ @Test
+ void testRestoreBeginDoesNotSkipFirstInsert() {
+ assertFirstInsertReplayed(Operation.BEGIN);
+ }
+
+ @Test
+ void testLegacyOffsetWithoutOperationReplaysAmbiguousBoundary() {
+ assertFirstInsertReplayed(null);
+ }
+
+ @Test
+ void testRestoreProcessedInsertResumesAtNextRecord() {
+ Lsn stored = Lsn.valueOf(123L);
+ Lsn next = Lsn.valueOf(124L);
+ WalPositionLocator locator =
+ new WalPositionLocator(Lsn.valueOf(100L), stored,
Operation.INSERT);
+ Assertions.assertFalse(locator.resumeFromLsn(stored,
message(Operation.BEGIN)).isPresent());
+ Assertions.assertFalse(
+ locator.resumeFromLsn(stored,
message(Operation.INSERT)).isPresent());
+ Assertions.assertEquals(
+ Optional.of(next), locator.resumeFromLsn(next,
message(Operation.UPDATE)));
+ locator.enableFiltering();
+ Assertions.assertTrue(locator.skipMessage(stored));
+ Assertions.assertFalse(locator.skipMessage(next));
+ }
+
+ @Test
+ void testOffsetOperationSurvivesCheckpointAndChangesAtCommit() {
+ PostgresOffsetContext.Loader loader = offsetLoader();
+ PostgresOffsetContext context = loader.load(legacyOffset());
+ Assertions.assertFalse(
+ context.getOffset()
+
.containsKey(PostgresOffsetContext.LAST_PROCESSED_MESSAGE_TYPE_KEY));
+ context.updateWalPosition(
+ Lsn.valueOf(123L),
+ Lsn.valueOf(123L),
+ Instant.EPOCH,
+ 7L,
+ null,
+ TableId.parse("postgres.public.test"),
+ Operation.INSERT);
+ Assertions.assertEquals(
+ "INSERT",
+ loader.load(context.getOffset())
+ .getOffset()
+
.get(PostgresOffsetContext.LAST_PROCESSED_MESSAGE_TYPE_KEY));
+
+ // The default non-transactional-metadata path updates the commit
independently.
+ context.updateCommitPosition(Lsn.valueOf(124L), Lsn.valueOf(124L));
+ PostgresOffsetContext restored = loader.load(context.getOffset());
+ Assertions.assertEquals(
+ "COMMIT",
+
restored.getOffset().get(PostgresOffsetContext.LAST_PROCESSED_MESSAGE_TYPE_KEY));
+ Assertions.assertEquals(
+ 124L,
+
restored.getOffset().get(PostgresOffsetContext.LAST_COMPLETELY_PROCESSED_LSN_KEY));
+ }
+
+ private void assertFirstInsertReplayed(Operation previousOperation) {
+ Lsn boundary = Lsn.valueOf(123L);
+ WalPositionLocator locator =
+ previousOperation == null
+ ? new WalPositionLocator(boundary, boundary)
+ : new WalPositionLocator(boundary, boundary,
previousOperation);
+ Assertions.assertFalse(
+ locator.resumeFromLsn(boundary,
message(Operation.BEGIN)).isPresent());
+ Assertions.assertEquals(
+ Optional.of(boundary), locator.resumeFromLsn(boundary,
message(Operation.INSERT)));
+ locator.enableFiltering();
+ Assertions.assertFalse(locator.skipMessage(boundary));
+ }
+
+ private ReplicationMessage message(Operation operation) {
+ ReplicationMessage message = mock(ReplicationMessage.class);
+ when(message.getOperation()).thenReturn(operation);
+ return message;
+ }
+
+ private PostgresOffsetContext.Loader offsetLoader() {
+ return new PostgresOffsetContext.Loader(
+ new PostgresConnectorConfig(
+ Configuration.create()
+ .with("database.server.name", "offset-test")
+ .with("database.hostname", "localhost")
+ .with("database.user", "test")
+ .with("database.dbname", "postgres")
+ .build()));
+ }
+
+ private Map<String, Object> legacyOffset() {
+ Map<String, Object> offset = new HashMap<>();
+ offset.put(SourceInfo.LSN_KEY, 123L);
+ offset.put(SourceInfo.TIMESTAMP_USEC_KEY, 0L);
+ offset.put(PostgresOffsetContext.LAST_COMPLETELY_PROCESSED_LSN_KEY,
123L);
+ offset.put(PostgresOffsetContext.LAST_COMMIT_LSN_KEY, 100L);
+ return offset;
+ }
}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/PostgresCDCIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/PostgresCDCIT.java
index beac962e97..4123adbac8 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/PostgresCDCIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/PostgresCDCIT.java
@@ -484,16 +484,10 @@ public class PostgresCDCIT extends TestSuiteBase
implements TestResource {
}
Assertions.assertEquals(1, seedRows.size());
});
- String postSeedLsn = getCurrentWalLsn();
- await().atMost(30000, TimeUnit.MILLISECONDS)
- .untilAsserted(
- () ->
- Assertions.assertTrue(
- isLsnGreaterThanOrEqual(
-
getReplicationSlotCommittedLsn(
- committedSlotName),
- postSeedLsn)));
- container.cancelJob(String.valueOf(seedJobId));
+ // Use a completed checkpoint as the committed-offset seed. The
global WAL also
+ // contains filtered sink writes and is not itself a source
checkpoint boundary.
+ Assertions.assertEquals(
+ 0,
container.savepointJob(String.valueOf(seedJobId)).getExitCode());
seedJob.get(30, TimeUnit.SECONDS);
await().atMost(30000, TimeUnit.MILLISECONDS)
.untilAsserted(
@@ -589,6 +583,17 @@ public class PostgresCDCIT extends TestSuiteBase
implements TestResource {
+ " where id = 14"));
});
+ // Drain the earlier writes before saving. With a quiet WAL, the
next transaction's
+ // BEGIN/INSERT can share the saved COMMIT LSN (the DBZ-6204
recovery boundary).
+ String quietBoundaryLsn = getCurrentWalLsn();
+ await().atMost(60000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertTrue(
+ isLsnGreaterThanOrEqual(
+
getReplicationSlotCommittedLsn(
+ committedSlotName),
+ quietBoundaryLsn)));
Assertions.assertEquals(
0,
container.savepointJob(String.valueOf(committedOffsetJobId)).getExitCode());
committedOffsetJob.get(30, TimeUnit.SECONDS);
@@ -601,6 +606,8 @@ public class PostgresCDCIT extends TestSuiteBase implements
TestResource {
Assertions.assertFalse(
isReplicationSlotActive(committedSlotName)));
+ // Changes committed while stopped must be replayed even at the
saved commit LSN.
+ insertSourceTableRow(POSTGRESQL_SCHEMA, SOURCE_TABLE_1, 15);
committedOffsetJob =
CompletableFuture.runAsync(
() -> {
@@ -618,9 +625,8 @@ public class PostgresCDCIT extends TestSuiteBase implements
TestResource {
// Restoring the checkpoint and reconnecting the existing
replication slot can take
// longer on shared GitHub runners than the initial CDC startup.
waitForReplicationSlotActive(committedSlotName);
- // Insert only after the restored replication connection is active
so this CDC record
- // is not written before the restored slot can consume it.
- insertSourceTableRow(POSTGRESQL_SCHEMA, SOURCE_TABLE_1, 15);
+ // Also verify that live changes continue after recovery.
+ insertSourceTableRow(POSTGRESQL_SCHEMA, SOURCE_TABLE_1, 16);
await().atMost(RESTORE_ASSERT_TIMEOUT_MILLIS,
TimeUnit.MILLISECONDS)
.untilAsserted(
() -> {
@@ -634,6 +640,15 @@ public class PostgresCDCIT extends TestSuiteBase
implements TestResource {
+ SINK_TABLE_1
+ " where id =
15")
.size());
+ Assertions.assertEquals(
+ 1,
+ query(
+ "select * from "
+ +
POSTGRESQL_SCHEMA
+ + "."
+ + SINK_TABLE_1
+ + " where id =
16")
+ .size());
});
} finally {
try {
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paypal-e2e/src/test/java/org/apache/seatunnel/e2e/connector/paypal/PayPalIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paypal-e2e/src/test/java/org/apache/seatunnel/e2e/connector/paypal/PayPalIT.java
index 95b27bb4ce..e324cf0916 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paypal-e2e/src/test/java/org/apache/seatunnel/e2e/connector/paypal/PayPalIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paypal-e2e/src/test/java/org/apache/seatunnel/e2e/connector/paypal/PayPalIT.java
@@ -75,6 +75,7 @@ public class PayPalIT extends TestSuiteBase implements
TestResource {
}
}
+ /** Verifies that a bodyless OAuth retry and token expiry refresh preserve
all records. */
@TestTemplate
public void readsAllRecordsThroughOAuthExpiryRefresh(TestContainer
container) throws Exception {
client.reset();
@@ -90,6 +91,7 @@ public class PayPalIT extends TestSuiteBase implements
TestResource {
"mock-client:mock-secret"
.getBytes(StandardCharsets.UTF_8)))
.withBody("grant_type=client_credentials");
+ client.when(oauth,
Times.exactly(1)).respond(HttpResponse.response().withStatusCode(503));
client.when(oauth, Times.exactly(1))
.respond(
HttpResponse.response()
@@ -117,7 +119,7 @@ public class PayPalIT extends TestSuiteBase implements
TestResource {
.withBody(report(2, row("T0006"))));
Container.ExecResult result =
container.executeJob("/paypal_to_assert.conf");
Assertions.assertEquals(0, result.getExitCode(), result.getStderr());
- client.verify(oauth, VerificationTimes.exactly(2));
+ client.verify(oauth, VerificationTimes.exactly(3));
client.verify(first, VerificationTimes.exactly(1));
client.verify(last, VerificationTimes.exactly(1));
client.verify(
diff --git
a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/CheckpointCoordinatorFailoverIT.java
b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/CheckpointCoordinatorFailoverIT.java
index d2256b3305..3cf9cd0980 100644
---
a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/CheckpointCoordinatorFailoverIT.java
+++
b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/CheckpointCoordinatorFailoverIT.java
@@ -371,6 +371,8 @@ public class CheckpointCoordinatorFailoverIT {
observedPipelineId.get(), ckIdBefore.get(),
ckIdAfter.get()));
clientJobProxy.cancelJob();
+ // Recovery must also resolve cancellation through the job
completion future.
+ Assertions.assertEquals(JobStatus.CANCELED,
clientJobProxy.waitForJobComplete());
} finally {
if (engineClient != null) {
engineClient.close();
diff --git
a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/java/org/apache/seatunnel/e2e/transform/TestFilterRowKindIT.java
b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/java/org/apache/seatunnel/e2e/transform/TestFilterRowKindIT.java
index d923d7137b..ee6f92aacc 100644
---
a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/java/org/apache/seatunnel/e2e/transform/TestFilterRowKindIT.java
+++
b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/java/org/apache/seatunnel/e2e/transform/TestFilterRowKindIT.java
@@ -45,6 +45,7 @@ public class TestFilterRowKindIT extends TestSuiteBase {
Assertions.assertEquals(0, execResult4.getExitCode());
}
+ /** Verifies exact per-table totals with one assertion writer across
engine parallelisms. */
@TestTemplate
public void testFilterRowKindMultiTable(TestContainer container)
throws IOException, InterruptedException {
diff --git
a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/resources/filter_row_kind_exclude_insert_multi_table.conf
b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/resources/filter_row_kind_exclude_insert_multi_table.conf
index ad6f50177b..ce2d726b20 100644
---
a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/resources/filter_row_kind_exclude_insert_multi_table.conf
+++
b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-1/src/test/resources/filter_row_kind_exclude_insert_multi_table.conf
@@ -103,6 +103,9 @@ transform {
sink {
Assert {
+ # Check totals for each table after collecting rows from every transform
partition.
+ # See apache/seatunnel#12216 for aggregation across multiple assertion
writers.
+ parallelism = 1
rules =
{
tables_configs = [
@@ -148,4 +151,4 @@ sink {
]
}
}
-}
\ No newline at end of file
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/TaskExecutionService.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/TaskExecutionService.java
index e2eb7f0fff..d72723cddc 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/TaskExecutionService.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/TaskExecutionService.java
@@ -404,7 +404,7 @@ public class TaskExecutionService implements
DynamicMetricsProvider {
MDCExecutorService mdcExecutorService =
MDCTracer.tracing(executorService);
CountDownLatch startedLatch = new CountDownLatch(tasks.size());
- taskGroupExecutionTracker.blockingFutures =
+ List<Future<?>> blockingFutures =
tasks.stream()
.map(
t ->
@@ -425,7 +425,17 @@ public class TaskExecutionService implements
DynamicMetricsProvider {
// on cancellation there is a race where the executor might not have
started
// the worker yet. This would result in taskletDone() never being
called for
// a worker.
- uncheckRun(startedLatch::await);
+ try {
+ uncheckRun(startedLatch::await);
+ } finally {
+ // Early failure cleanup must not cancel a worker before run()
releases startedLatch.
+ // Publish the futures after startup, then honor any failure or
cancellation received
+ // while waiting so already-running tasks are still interrupted.
+ taskGroupExecutionTracker.blockingFutures = blockingFutures;
+ if (taskGroupExecutionTracker.executionCompletedExceptionally()) {
+ taskGroupExecutionTracker.cancelAllTask();
+ }
+ }
}
/**
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java
index 273bd587cb..c2a73ef9c1 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java
@@ -406,9 +406,17 @@ public class PhysicalVertex {
}
}
+ /**
+ * Requests cancellation and publishes any terminal result received while
state processing was
+ * stopped for pipeline restore.
+ */
public synchronized void cancel() {
if (!getExecutionState().isEndState()) {
updateTaskState(ExecutionState.CANCELING);
+ } else if (!taskFuture.isDone()) {
+ // A late terminal callback may arrive before this vertex resumes.
Its new completion
+ // future still needs to be resolved without changing the terminal
state.
+ stateProcess();
}
}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/TaskExecutionServiceTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/TaskExecutionServiceTest.java
index e95b23f586..2ce8e0c022 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/TaskExecutionServiceTest.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/TaskExecutionServiceTest.java
@@ -46,6 +46,8 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.RepeatedTest;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.Mockito;
import com.hazelcast.flakeidgen.FlakeIdGenerator;
@@ -66,9 +68,17 @@ import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
import static java.util.Collections.emptySet;
import static
org.apache.seatunnel.engine.server.execution.ExecutionState.CANCELED;
@@ -105,6 +115,136 @@ public class TaskExecutionServiceTest extends
AbstractSeaTunnelServerTest {
failure -> {});
}
+ /**
+ * A task can fail or be cancelled while another blocking worker has not
yet been scheduled.
+ * Cleanup must let that worker release the startup latch so deployment
and the final result
+ * both return.
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testBlockingDeploymentCompletesAfterEarlyFailureOrCancellation(
+ boolean cancelDuringStartup) throws Exception {
+ TaskExecutionService service = server.getTaskExecutionService();
+ ExecutorService originalExecutor = getField(service,
"executorService");
+ CountDownLatch failFirstTask = new CountDownLatch(1);
+ CountDownLatch firstTaskClosed = new CountDownLatch(1);
+ CountDownLatch startSecondWorker = new CountDownLatch(1);
+ CountDownLatch workersSubmitted = new CountDownLatch(2);
+ AtomicInteger workerNumber = new AtomicInteger();
+ AtomicReference<Thread> deploymentThread = new AtomicReference<>();
+ ExecutorService workers =
+ new ThreadPoolExecutor(
+ 0,
+ Integer.MAX_VALUE,
+ 60L,
+ TimeUnit.SECONDS,
+ new SynchronousQueue<>(),
+ runnable -> {
+ int number = workerNumber.incrementAndGet();
+ Thread thread =
+ new Thread(
+ () -> {
+ try {
+ if (number == 2) {
+
startSecondWorker.await();
+ }
+ runnable.run();
+ } catch (InterruptedException
e) {
+
Thread.currentThread().interrupt();
+ }
+ },
+ "delayed-blocking-worker-" +
number);
+ thread.setDaemon(true);
+ return thread;
+ }) {
+ @Override
+ public void execute(Runnable command) {
+ super.execute(command);
+ workersSubmitted.countDown();
+ }
+ };
+ ExecutorService deployer = Executors.newSingleThreadExecutor();
+ TaskGroupLocation location = newTaskGroupLocation();
+ String failureMessage = "Source permission denied during
initialization";
+ Task failingTask =
+ new TestTask(new AtomicBoolean(true), 0, false) {
+ @Override
+ public void init() throws Exception {
+ failFirstTask.await();
+ throw new IllegalStateException(failureMessage);
+ }
+
+ @Override
+ public void close() {
+ firstTaskClosed.countDown();
+ }
+ };
+ Task delayedTask = new TestTask(new AtomicBoolean(true), 0, false);
+ TaskGroup group =
+ new TaskGroupDefaultImpl(
+ location,
+ "early-startup-failure",
+ Lists.newArrayList(failingTask, delayedTask));
+ ConcurrentHashMap<Long, ClassLoader> classLoaders = new
ConcurrentHashMap<>();
+ for (Task task : group.getTasks()) {
+ classLoaders.put(task.getTaskID(),
Thread.currentThread().getContextClassLoader());
+ }
+
+ ReflectionUtils.setField(service, "executorService", workers);
+ Future<PassiveCompletableFuture<TaskExecutionState>> deployment = null;
+ try {
+ Future<PassiveCompletableFuture<TaskExecutionState>>
submittedDeployment =
+ deployer.submit(
+ () -> {
+ deploymentThread.set(Thread.currentThread());
+ return service.deployLocalTask(
+ FLAKE_ID_GENERATOR.newId(),
+ group,
+ classLoaders,
+ new ConcurrentHashMap<>(),
+ () -> {},
+ failure -> {});
+ });
+ deployment = submittedDeployment;
+ assertTrue(workersSubmitted.await(10, TimeUnit.SECONDS));
+ // Both executor submissions have returned, so an unfinished
deployment can only
+ // be waiting for the delayed worker's startup signal, not
submitting a worker.
+ await().atMost(10, TimeUnit.SECONDS)
+ .until(
+ () -> {
+ Thread thread = deploymentThread.get();
+ return !submittedDeployment.isDone()
+ && thread.getState() ==
Thread.State.WAITING;
+ });
+ if (cancelDuringStartup) {
+ service.cancelTaskGroup(location);
+ } else {
+ failFirstTask.countDown();
+ assertTrue(firstTaskClosed.await(10, TimeUnit.SECONDS));
+ }
+ startSecondWorker.countDown();
+
+ TaskExecutionState result =
+ deployment.get(10, TimeUnit.SECONDS).get(10,
TimeUnit.SECONDS);
+ assertEquals(cancelDuringStartup ? CANCELED : FAILED,
result.getExecutionState());
+ if (!cancelDuringStartup) {
+ assertTrue(result.getThrowableMsg().contains(failureMessage));
+ }
+ } finally {
+ failFirstTask.countDown();
+ startSecondWorker.countDown();
+ if (deployment != null) {
+ deployment.cancel(true);
+ }
+ service.cancelTaskGroup(location);
+ workers.shutdownNow();
+ deployer.shutdownNow();
+ workers.awaitTermination(10, TimeUnit.SECONDS);
+ deployer.awaitTermination(10, TimeUnit.SECONDS);
+ ReflectionUtils.setField(service, "executorService",
originalExecutor);
+ }
+ }
+
@Test
public void testCancel() {
TaskExecutionService taskExecutionService =
server.getTaskExecutionService();
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/dag/physical/StateTransitionCleanupTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/dag/physical/StateTransitionCleanupTest.java
index 7a1a339a0e..2bcf9a53db 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/dag/physical/StateTransitionCleanupTest.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/dag/physical/StateTransitionCleanupTest.java
@@ -23,15 +23,19 @@ import org.apache.seatunnel.engine.common.Constant;
import org.apache.seatunnel.engine.common.config.EngineConfig;
import org.apache.seatunnel.engine.common.config.JobConfig;
import org.apache.seatunnel.engine.common.job.JobStatus;
+import org.apache.seatunnel.engine.common.utils.PassiveCompletableFuture;
import org.apache.seatunnel.engine.core.dag.logical.LogicalDag;
import org.apache.seatunnel.engine.core.job.JobImmutableInformation;
import org.apache.seatunnel.engine.core.job.PipelineStatus;
import org.apache.seatunnel.engine.server.AbstractSeaTunnelServerTest;
import org.apache.seatunnel.engine.server.TestUtils;
import org.apache.seatunnel.engine.server.execution.ExecutionState;
+import org.apache.seatunnel.engine.server.execution.TaskExecutionState;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junitpioneer.jupiter.SetEnvironmentVariable;
import com.hazelcast.map.IMap;
@@ -46,6 +50,62 @@ import static
org.apache.seatunnel.engine.core.classloader.DefaultClassLoaderSer
@SetEnvironmentVariable(key = SKIP_CHECK_JAR, value = "true")
class StateTransitionCleanupTest extends AbstractSeaTunnelServerTest {
+ /**
+ * A terminal callback can arrive after reset while the vertex is stopped.
Resuming pipeline
+ * cancellation must complete the new future even though no further task
transition is needed.
+ */
+ @ParameterizedTest
+ @CsvSource({
+ "FAILED, true",
+ "CANCELED, true",
+ "FINISHED, true",
+ "FAILED, false",
+ "CANCELED, false",
+ "FINISHED, false"
+ })
+ void testRestoredCancellationCompletesLateTerminalState(
+ ExecutionState terminalState, boolean callbackBeforeInit) throws
Exception {
+ long jobId =
instance.getFlakeIdGenerator(Constant.SEATUNNEL_ID_GENERATOR_NAME).newId();
+ PlanWithStateMaps planWithStateMaps = createPhysicalPlan(jobId);
+ PhysicalVertex vertex =
+ planWithStateMaps
+ .physicalPlan
+ .getPipelineList()
+ .get(0)
+ .getPhysicalVertexList()
+ .get(0);
+
+ vertex.startPhysicalVertex();
+ vertex.updateTaskState(ExecutionState.CANCELED);
+ vertex.reset();
+ String error =
+ terminalState == ExecutionState.FAILED ? "Paimon SELECT
permission denied" : null;
+ TaskExecutionState lateResult =
+ new TaskExecutionState(vertex.getTaskGroupLocation(),
terminalState, error);
+ if (callbackBeforeInit) {
+ vertex.updateStateByExecutionService(lateResult);
+ }
+ PassiveCompletableFuture<TaskExecutionState> restoredFuture =
vertex.initStateFuture();
+
+ if (!callbackBeforeInit) {
+ vertex.updateStateByExecutionService(lateResult);
+ }
+ Assertions.assertFalse(restoredFuture.isDone());
+
+ vertex.startPhysicalVertex();
+ vertex.cancel();
+
+ Assertions.assertTrue(
+ restoredFuture.isDone(),
+ "Resumed cancellation must publish the terminal task result");
+ TaskExecutionState result = restoredFuture.get();
+ Assertions.assertEquals(terminalState, result.getExecutionState());
+ Assertions.assertEquals(error, result.getThrowableMsg());
+ Assertions.assertEquals(
+ terminalState,
+
planWithStateMaps.runningJobState.get(vertex.getTaskGroupLocation()));
+ }
+
@Test
void testPhysicalVertexIgnoresLateTransitionWhenTaskStateAlreadyTerminal()
throws Exception {
long jobId =
instance.getFlakeIdGenerator(Constant.SEATUNNEL_ID_GENERATOR_NAME).newId();
