This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12454-7c4f2d77209fd0c8edcc42aff105625f3d2a1406 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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();
