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-11569-84e43a95b469b2021b0590e0bbac703bcc9cdcf3 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 7a58d1005b864ce787ded4f0d64c6aab1135393b Author: Daniel <[email protected]> AuthorDate: Mon Sep 7 05:57:12 2026 +0000 [Fix][Connector-V2] Propagate JDBC XA commit failures (#11569) Co-authored-by: DanielLeens <[email protected]> Co-authored-by: davidzollo <[email protected]> Co-authored-by: davidzollo <[email protected]> Co-authored-by: Claude Sonnet 5 <[email protected]> --- docs/en/connectors/sink/Jdbc.md | 10 + .../introduction/concepts/incompatible-changes.md | 6 + docs/zh/connectors/sink/Jdbc.md | 8 + .../introduction/concepts/incompatible-changes.md | 6 + .../seatunnel/jdbc/internal/xa/XaFacade.java | 7 +- .../jdbc/internal/xa/XaFacadeImplAutoLoad.java | 14 +- .../seatunnel/jdbc/internal/xa/XaGroupOps.java | 13 +- .../seatunnel/jdbc/internal/xa/XaGroupOpsImpl.java | 9 +- .../jdbc/sink/JdbcSinkAggregatedCommitter.java | 278 ++++++++++++- .../jdbc/internal/xa/XaFacadeImplAutoLoadTest.java | 145 +++++++ .../jdbc/internal/xa/XaGroupOpsImplTest.java | 122 ++++++ .../jdbc/sink/JdbcSinkAggregatedCommitterTest.java | 440 +++++++++++++++++++++ .../jdbc/internal/xa/XaGroupOpsImplIT.java | 75 +++- 13 files changed, 1097 insertions(+), 36 deletions(-) diff --git a/docs/en/connectors/sink/Jdbc.md b/docs/en/connectors/sink/Jdbc.md index d3181e8a41..00172e891a 100644 --- a/docs/en/connectors/sink/Jdbc.md +++ b/docs/en/connectors/sink/Jdbc.md @@ -336,6 +336,16 @@ please refer to appendix for other data sources The number of retries for transaction commit failures +In exactly-once XA mode, this retry budget is consumed within a single aggregated-commit or +restore invocation. During restore, SeaTunnel replays the still-prepared suffix starting from the +first checkpoint XID that remains in the XA recovery scan. Missing XIDs before that boundary are +treated as already resolved only after the still-prepared suffix commits successfully. If none of +the checkpoint XIDs remain in the recovery scan, SeaTunnel treats the whole batch as already +resolved and does not replay it. SeaTunnel waits one second between synchronous retry rounds so +transient resource-manager outages can consume the retry budget over time instead of burning every +attempt immediately. If a gap appears after the first recovered checkpoint XID, +recovery still fails closed instead of inferring a successful commit. + ### transaction_timeout_sec [int] The timeout after the transaction is opened, the default is -1 (never timeout). Note that setting the timeout may affect diff --git a/docs/en/introduction/concepts/incompatible-changes.md b/docs/en/introduction/concepts/incompatible-changes.md index 50561c2ebb..79ce8ea56c 100644 --- a/docs/en/introduction/concepts/incompatible-changes.md +++ b/docs/en/introduction/concepts/incompatible-changes.md @@ -19,6 +19,12 @@ You need to check this document before you upgrade to related version. ### JDBC Connector +- **Breaking Change: JDBC XA restore now uses recovery-order evidence and fail-closed gaps** + - **Affected component**: `seatunnel-connectors-v2/connector-jdbc` sink exactly-once XA path + - **Description**: SeaTunnel now consumes `max_commit_attempts` within a single aggregated-commit or restore invocation, and restore replays only the still-prepared suffix starting from the first checkpoint XID that remains in the XA recovery scan. Missing XIDs before that boundary are treated as already resolved only after the suffix commits successfully. If none of the checkpoint XIDs remain in the recovery scan, SeaTunnel treats the whole batch as already resolved and skips replay. [...] + - **Impact**: Jobs that previously relied on restore inferring success from a missing XA branch may now fail during recovery when the XA recovery scan still contains later checkpoint XIDs but shows a gap after them. Operators may also observe that `max_commit_attempts` is exhausted within one restore/commit invocation rather than across repeated task restarts. + - **Migration Guide**: Before upgrading, inspect the resource manager for dangling prepared XA transactions (for example `XA RECOVER` on MySQL or `pg_prepared_xacts` on PostgreSQL). If recovery fails closed because a later checkpoint XID still exists but a following one is missing, investigate whether the missing XID was rolled back, expired, or cleaned up externally before retrying the job. XA recovery cannot distinguish a SeaTunnel-committed XID from one rolled back or removed by an [...] + - **Breaking Change: Mapping of timezone-aware timestamp columns to `TIMESTAMP_TZ` type** - **Affected component**: `seatunnel-connectors-v2/connector-jdbc`, `seatunnel-connectors-v2/connector-iceberg`, `seatunnel-connectors-v2/connector-cdc-base`, `seatunnel-connectors-v2/connector-cdc-tidb`, `seatunnel-connectors-v2/connector-starrocks`, `seatunnel-connectors-v2/connector-hudi`, `seatunnel-connectors-v2/connector-snowflake` (via JDBC dialect) - **Description**: Previously, JDBC sources mapped both timezone-naive (e.g., MySQL `DATETIME`) and timezone-aware (e.g., MySQL `TIMESTAMP`) timestamp columns to SeaTunnel's internal `TIMESTAMP` type. Now, timezone-aware columns like MySQL `TIMESTAMP`, PostgreSQL `timestamptz`, Oracle `TIMESTAMP WITH LOCAL TIME ZONE`, SQL Server `datetimeoffset`, Snowflake `TIMESTAMP_LTZ/TZ`, and others are explicitly mapped to `TIMESTAMP_TZ`. This ensures that timezone semantics are accurately preserv [...] diff --git a/docs/zh/connectors/sink/Jdbc.md b/docs/zh/connectors/sink/Jdbc.md index 1db79f01f0..f5ac84c757 100644 --- a/docs/zh/connectors/sink/Jdbc.md +++ b/docs/zh/connectors/sink/Jdbc.md @@ -335,6 +335,14 @@ JDBC `executeBatch` 失败后的重试次数。Exactly-once 模式要求设置 事务提交失败的最大重试次数 +在 exactly-once 的 XA 模式下，这个重试预算会在单次 aggregated-commit 或 restore 调用内消耗完。 +恢复时，SeaTunnel 会从 XA recovery scan 中第一个仍然处于 prepared 状态的 checkpoint XID 开始， +严格回放其后的事务后缀。位于该边界之前、且在 recovery scan 中缺失的 XID，只有在后缀严格提交 +成功之后才会被视为已经解析完成。如果 recovery scan 中一个 checkpoint XID 都不存在，SeaTunnel +会把整个批次视为已经完成，不再重复回放。SeaTunnel 会在同步重试轮次之间固定等待 1 秒，让瞬时 +的资源管理器不可用可以真正消耗这份重试预算，而不是在瞬间烧完所有尝试次数。只有在第一个 recovered checkpoint XID 之后又出现缺 +口时，恢复才会直接 fail-closed，而不是把结果推断成已经成功提交。 + ### transaction_timeout_sec [int] 在事务开启后的超时时间，默认值为-1（即永不超时）。请注意，设置超时时间可能会影响到精确一次（exactly-once）的语义 diff --git a/docs/zh/introduction/concepts/incompatible-changes.md b/docs/zh/introduction/concepts/incompatible-changes.md index 887647be93..9e16201b6f 100644 --- a/docs/zh/introduction/concepts/incompatible-changes.md +++ b/docs/zh/introduction/concepts/incompatible-changes.md @@ -13,6 +13,12 @@ ### JDBC Connector +- **破坏性变更：JDBC XA restore 改为基于 recovery 顺序证据并对缺口 fail-closed** + - **影响范围**：`seatunnel-connectors-v2/connector-jdbc` sink 的 exactly-once XA 路径 + - **变更说明**：SeaTunnel 现在会在单次 aggregated-commit 或 restore 调用内消耗完 `max_commit_attempts`。恢复时，只会从 XA recovery scan 中第一个仍然存在的 checkpoint XID 开始，严格回放其后的 prepared 事务后缀。位于该边界之前、且在 recovery scan 中缺失的 XID，只有在后缀严格提交成功之后才会被视为已经完成；如果 recovery scan 中一个 checkpoint XID 都不存在，SeaTunnel 会把整个批次视为已经完成并跳过回放；只有在第一个 recovered checkpoint XID 之后又出现缺失 XID 时，restore 才会直接 fail-closed，而不是仅凭 `XAER_NOTA` 这类“事务不存在”结果去推断已经提交成功。 + - **影响**：以前依赖“XA 分支缺失即视为成功”的任务，在升级后如果 recovery scan 里仍然能看到后续 checkpoint XID、但中间出现缺口，可能会在恢复阶段收到明确的 XA restore 错误。另外，`max_commit_attempts` 现在会在一次 restore/commit 调用内耗尽，而不是分散到多次任务重启中。 + - **迁移指南**：升级前请先检查资源管理器中是否还残留 prepared XA 事务，例如 MySQL 可使用 `XA RECOVER`，PostgreSQL 可检查 `pg_prepared_xacts`。如果升级后 restore 因为“后面仍有 checkpoint XID，但中间出现缺失 XID”而进入 fail-closed，请重点确认缺失 XID 是否被回滚、超时过期，或被外部清理，再决定后续恢复操作。XA recovery 无法区分 SeaTunnel 已提交的 XID 与被外部清理者回滚或删除的 XID。因此，位于 recovered 后缀之前的缺失 XID，或全部 XID 缺失的批次，会被推断为已经完成；在相关作业可能恢复时，不要对 SeaTunnel 所属的 prepared XA 分支执行外部清理，任何清理操作都应与作业恢复流程协调。 + - **破坏性变更：带时区的时间戳列映射为 `TIMESTAMP_TZ` 类型** - **影响范围**：`seatunnel-connectors-v2/connector-jdbc`、`seatunnel-connectors-v2/connector-iceberg`、`seatunnel-connectors-v2/connector-cdc-base`、`seatunnel-connectors-v2/connector-cdc-tidb`、`seatunnel-connectors-v2/connector-starrocks`、`seatunnel-connectors-v2/connector-hudi`、`seatunnel-connectors-v2/connector-snowflake`（通过 JDBC 方言） - **变更说明**：以前，JDBC Source 将无时区（如 MySQL `DATETIME`）和带时区（如 MySQL `TIMESTAMP`）的时间戳列都映射为 SeaTunnel 内部的 `TIMESTAMP` 类型。现在，带时区的列（如 MySQL `TIMESTAMP`、PostgreSQL `timestamptz`、Oracle `TIMESTAMP WITH LOCAL TIME ZONE`、SQL Server `datetimeoffset`、Snowflake `TIMESTAMP_LTZ/TZ` 等）被显式映射为 `TIMESTAMP_TZ`。这确保了在写入 Iceberg 等格式时，时区语义得到准确保留（在 Iceberg 中 `TIMESTAMP` 存为无时区的 `timestamp`，`TIMESTAMP_TZ` 存为带时区的 `timestamptz`）。 diff --git a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/xa/XaFacade.java b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/xa/XaFacade.java index e47ca7d4a9..acd709db8f 100644 --- a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/xa/XaFacade.java +++ b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/xa/XaFacade.java @@ -97,11 +97,12 @@ public interface XaFacade extends JdbcConnectionProvider, Serializable, AutoClos } /** - * Indicates a transient or unknown failure from the resource manager (see {@link - * XAException#XA_RBTRANSIENT XA_RBTRANSIENT}, {@link XAException#XAER_RMFAIL XAER_RMFAIL}). + * Indicates that an XA operation had no effect and can be retried, or that its outcome is + * unknown because the resource manager is unavailable (see {@link XAException#XA_RETRY + * XA_RETRY}, {@link XAException#XAER_RMFAIL XAER_RMFAIL}). */ class TransientXaException extends RuntimeException { - TransientXaException(XAException cause) { + public TransientXaException(XAException cause) { super(cause); } } diff --git a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/xa/XaFacadeImplAutoLoad.java b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/xa/XaFacadeImplAutoLoad.java index 5dedc0e475..2842613b30 100644 --- a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/xa/XaFacadeImplAutoLoad.java +++ b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/xa/XaFacadeImplAutoLoad.java @@ -56,6 +56,7 @@ import static javax.transaction.xa.XAException.XA_HEURRB; import static javax.transaction.xa.XAException.XA_RBBASE; import static javax.transaction.xa.XAException.XA_RBTIMEOUT; import static javax.transaction.xa.XAException.XA_RBTRANSIENT; +import static javax.transaction.xa.XAException.XA_RETRY; import static javax.transaction.xa.XAResource.TMENDRSCAN; import static javax.transaction.xa.XAResource.TMNOFLAGS; import static javax.transaction.xa.XAResource.TMSTARTRSCAN; @@ -72,7 +73,7 @@ public class XaFacadeImplAutoLoad implements XaFacade { private static final Logger LOG = LoggerFactory.getLogger(XaFacadeImplAutoLoad.class); private static final Set<Integer> TRANSIENT_ERR_CODES = - new HashSet<>(Arrays.asList(XA_RBTRANSIENT, XAER_RMFAIL)); + new HashSet<>(Arrays.asList(XA_RETRY, XAER_RMFAIL)); private static final Set<Integer> HEUR_ERR_CODES = new HashSet<>(Arrays.asList(XA_HEURRB, XA_HEURCOM, XA_HEURHAZ, XA_HEURMIX)); private static final int MAX_RECOVER_CALLS = 100; @@ -297,19 +298,22 @@ public class XaFacadeImplAutoLoad implements XaFacade { } } + /** + * Preserves retryable XA failures as {@link TransientXaException}; all other failures use the + * connector error contract. + */ private static RuntimeException wrapException(String action, Optional<Xid> xid, Exception ex) { if (ex instanceof XAException) { XAException xa = (XAException) ex; if (TRANSIENT_ERR_CODES.contains(xa.errorCode)) { - throw new JdbcConnectorException( - JdbcConnectorErrorCode.XA_OPERATION_FAILED, new TransientXaException(xa)); + return new TransientXaException(xa); } else { - throw new JdbcConnectorException( + return new JdbcConnectorException( JdbcConnectorErrorCode.XA_OPERATION_FAILED, formatErrorMessage(action, xid, of(xa.errorCode), xa.getMessage())); } } else { - throw new JdbcConnectorException( + return new JdbcConnectorException( JdbcConnectorErrorCode.XA_OPERATION_FAILED, formatErrorMessage(action, xid, empty(), ex.getMessage()), ex); diff --git a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/xa/XaGroupOps.java b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/xa/XaGroupOps.java index 655d9fe989..e254e27764 100644 --- a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/xa/XaGroupOps.java +++ b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/xa/XaGroupOps.java @@ -29,8 +29,17 @@ import java.util.List; public interface XaGroupOps extends Serializable { - // Commit a batch of transactions - public GroupXaOperationResult<XidInfo> commit( + /** + * Commits a batch of prepared transactions. + * + * @param xids prepared transactions to commit + * @param allowOutOfOrderCommits whether transactions after a failed transaction may be + * committed + * @param maxCommitAttempts maximum number of transient commit attempts + * @return commit result containing transactions that need another attempt; implementations must + * eventually return an empty retry list or throw once the retry budget is exhausted + */ + GroupXaOperationResult<XidInfo> commit( List<XidInfo> xids, boolean allowOutOfOrderCommits, int maxCommitAttempts); void rollback(List<XidInfo> xids); diff --git a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/xa/XaGroupOpsImpl.java b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/xa/XaGroupOpsImpl.java index 88a48a65e0..77acbaf4f8 100644 --- a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/xa/XaGroupOpsImpl.java +++ b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/xa/XaGroupOpsImpl.java @@ -67,12 +67,9 @@ public class XaGroupOpsImpl implements XaGroupOps { } } result.getForRetry().addAll(xids); - // TODO At present, it is impossible to distinguish whether - // the repeated Commit failure caused by restore (exception should not be thrown) or - // the failure of normal process Commit (exception should be thrown). - // So currently the exception is not thrown. - - // result.throwIfAnyFailed("commit"); + // A permanent or unknown commit failure must fail the checkpoint instead of being reported + // as a successful commit. + result.throwIfAnyFailed("commit"); throwIfAnyReachedMaxAttempts(result, maxCommitAttempts); result.getTransientFailure() .ifPresent( diff --git a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkAggregatedCommitter.java b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkAggregatedCommitter.java index c4abb10cb8..6cbefba761 100644 --- a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkAggregatedCommitter.java +++ b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkAggregatedCommitter.java @@ -30,15 +30,25 @@ import org.apache.seatunnel.connectors.seatunnel.jdbc.state.XidInfo; import lombok.extern.slf4j.Slf4j; +import javax.transaction.xa.Xid; + import java.io.IOException; import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.HashSet; import java.util.List; -import java.util.stream.Collectors; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.TimeUnit; @Slf4j public class JdbcSinkAggregatedCommitter implements SinkAggregatedCommitter<XidInfo, JdbcAggregatedCommitInfo> { + private static final long XA_RETRY_BACKOFF_MILLIS = TimeUnit.SECONDS.toMillis(1); + private XaFacade xaFacade; private XaGroupOps xaGroupOps; private final JdbcSinkConfig jdbcSinkConfig; @@ -70,22 +80,79 @@ public class JdbcSinkAggregatedCommitter @Override public List<JdbcAggregatedCommitInfo> commit( List<JdbcAggregatedCommitInfo> aggregatedCommitInfos) throws IOException { + return commitPreparedTransactions(aggregatedCommitInfos); + } + + /** + * Reconciles checkpoint XIDs with the resource manager using commit-order evidence. Checkpoint + * XIDs from the first still-prepared transaction onward must all be present in the recovery + * scan and are replayed strictly. An all-absent batch is treated as already resolved, while an + * absent prefix before a still-prepared suffix is treated as already resolved only after that + * suffix commits successfully. The scan is refreshed for each batch because an external + * resource-manager actor, such as a DBA or RM cleanup process, can resolve an in-doubt + * transaction between restored batches. This is not a second Zeta committer; the refresh + * narrows, but cannot eliminate, the gap before the following XA commit call. + */ + @Override + public List<JdbcAggregatedCommitInfo> restoreCommit( + List<JdbcAggregatedCommitInfo> aggregatedCommitInfos) throws IOException { + tryOpen(); + for (JdbcAggregatedCommitInfo aggregatedCommitInfo : aggregatedCommitInfos) { + // Refresh RM evidence for every batch because a DBA or RM cleanup process can resolve + // an in-doubt transaction while earlier restored batches are being replayed. + replayRecoveredCheckpoint( + aggregatedCommitInfo.getXidInfoList(), recoverCheckpointTransactions()); + } + return Collections.emptyList(); + } + + /** + * Commits prepared transactions and completes bounded transient retries in the same invocation. + * + * @param aggregatedCommitInfos prepared transactions grouped for commit + * @return an empty list after all prepared transactions have been committed + * @throws IOException when the committer cannot open its XA resource + */ + private List<JdbcAggregatedCommitInfo> commitPreparedTransactions( + List<JdbcAggregatedCommitInfo> aggregatedCommitInfos) throws IOException { tryOpen(); - return aggregatedCommitInfos.stream() - .map( - aggregatedCommitInfo -> { - log.info("commit xid: " + aggregatedCommitInfo.getXidInfoList()); - GroupXaOperationResult<XidInfo> result = - xaGroupOps.commit( - new ArrayList<>(aggregatedCommitInfo.getXidInfoList()), - false, - jdbcSinkConfig - .getJdbcConnectionConfig() - .getMaxCommitAttempts()); - return new JdbcAggregatedCommitInfo(result.getForRetry()); - }) - .filter(ainfo -> !ainfo.getXidInfoList().isEmpty()) - .collect(Collectors.toList()); + for (JdbcAggregatedCommitInfo aggregatedCommitInfo : aggregatedCommitInfos) { + log.info("commit xid: " + aggregatedCommitInfo.getXidInfoList()); + commitXidInfos(aggregatedCommitInfo.getXidInfoList()); + } + return Collections.emptyList(); + } + + /** + * Commits one XID list and exhausts bounded transient retries while the retry-attempt state is + * still available in memory. + * + * @param xidInfos prepared transactions to commit + */ + private void commitXidInfos(List<XidInfo> xidInfos) { + List<XidInfo> pending = new ArrayList<>(xidInfos); + int maxCommitAttempts = jdbcSinkConfig.getJdbcConnectionConfig().getMaxCommitAttempts(); + int remainingRounds = Math.max(1, maxCommitAttempts); + while (!pending.isEmpty()) { + if (remainingRounds-- == 0) { + throw new JdbcConnectorException( + CommonErrorCodeDeprecated.WRITER_OPERATION_FAILED, + String.format( + "XA commit retry loop did not terminate within %d rounds for transactions: %s", + Math.max(1, maxCommitAttempts), pending)); + } + GroupXaOperationResult<XidInfo> result = + xaGroupOps.commit(pending, false, maxCommitAttempts); + // Zeta does not persist the returned committables before restarting a failed + // checkpoint, so complete bounded retries in this invocation. + pending = new ArrayList<>(result.getForRetry()); + if (!pending.isEmpty() && remainingRounds > 0) { + backoffBeforeRetry( + "commit", + Math.max(1, maxCommitAttempts) - remainingRounds + 1, + Math.max(1, maxCommitAttempts)); + } + } } @Override @@ -114,4 +181,183 @@ public class JdbcSinkAggregatedCommitter e); } } + + /** + * Normalizes driver-specific {@link Xid} implementations into canonical values for recovery + * reconciliation. + * + * @param recoveredXids transactions returned by the resource manager + * @return canonical values for transactions returned by the resource manager + */ + private Set<XidKey> normalizeXids(Collection<Xid> recoveredXids) { + Set<XidKey> normalized = new HashSet<>(); + for (Xid xid : recoveredXids) { + normalized.add(XidKey.from(xid)); + } + return normalized; + } + + /** + * Replays one restored checkpoint using the recovery scan as evidence for which transactions + * can still be committed safely. + * + * @param checkpointXids checkpoint-owned prepared transactions in original commit order + * @param recoveredXids canonical values returned by the resource manager recovery scan + */ + private void replayRecoveredCheckpoint( + List<XidInfo> checkpointXids, Set<XidKey> recoveredXids) { + if (checkpointXids.isEmpty()) { + return; + } + int firstRecoveredIndex = findFirstRecoveredIndex(checkpointXids, recoveredXids); + if (firstRecoveredIndex < 0) { + log.warn( + "Skipping checkpoint batch because none of its transactions remain in the XA recovery scan; treating it as already resolved: {}", + checkpointXids); + return; + } + List<XidInfo> stillPrepared = new ArrayList<>(); + for (int i = firstRecoveredIndex; i < checkpointXids.size(); i++) { + XidInfo xidInfo = checkpointXids.get(i); + if (!containsEquivalentXid(recoveredXids, xidInfo.getXid())) { + throw new JdbcConnectorException( + CommonErrorCodeDeprecated.WRITER_OPERATION_FAILED, + String.format( + "checkpoint transaction %s is absent from the XA recovery scan after still-prepared transactions in the same commit batch: %s", + xidInfo.getXid(), checkpointXids)); + } + stillPrepared.add(xidInfo); + } + commitXidInfos(stillPrepared); + if (firstRecoveredIndex > 0) { + List<XidInfo> alreadyResolved = checkpointXids.subList(0, firstRecoveredIndex); + log.warn( + "Skipping {} checkpoint transactions that are absent from the XA recovery scan but precede the first still-prepared transaction: {}", + alreadyResolved.size(), + alreadyResolved); + } + } + + /** + * Retries the recovery scan with the same bounded budget as XA commit so transient RM outages + * do not fail restore immediately. + * + * @return canonical values returned by the recovery scan + */ + private Set<XidKey> recoverCheckpointTransactions() { + int maxCommitAttempts = jdbcSinkConfig.getJdbcConnectionConfig().getMaxCommitAttempts(); + XaFacade.TransientXaException lastTransientFailure = null; + for (int attempt = 1; attempt <= Math.max(1, maxCommitAttempts); attempt++) { + try { + return normalizeXids(xaFacade.recover()); + } catch (XaFacade.TransientXaException e) { + lastTransientFailure = e; + log.warn( + "Transient XA recovery-scan failure on attempt {}/{}", + attempt, + Math.max(1, maxCommitAttempts), + e); + if (attempt < Math.max(1, maxCommitAttempts)) { + backoffBeforeRetry( + "recovery scan", attempt + 1, Math.max(1, maxCommitAttempts)); + } + } + } + throw new JdbcConnectorException( + CommonErrorCodeDeprecated.WRITER_OPERATION_FAILED, + String.format( + "unable to complete the XA recovery scan within %d attempts", + Math.max(1, maxCommitAttempts)), + lastTransientFailure); + } + + /** + * Finds the first checkpoint transaction that is still prepared in the resource manager. + * + * @param checkpointXids checkpoint-owned prepared transactions in original commit order + * @param recoveredXids canonical values returned by the recovery scan + * @return the first still-prepared checkpoint transaction index, or {@code -1} if none remain + */ + private int findFirstRecoveredIndex(List<XidInfo> checkpointXids, Set<XidKey> recoveredXids) { + for (int i = 0; i < checkpointXids.size(); i++) { + if (containsEquivalentXid(recoveredXids, checkpointXids.get(i).getXid())) { + return i; + } + } + return -1; + } + + /** + * Checks whether the recovery scan contains the checkpoint transaction by canonical XID value. + * + * @param recoveredXids canonical recovery-scan values + * @param checkpointXid transaction restored from checkpoint state + * @return whether the recovery scan contains the checkpoint transaction + */ + private boolean containsEquivalentXid(Set<XidKey> recoveredXids, Xid checkpointXid) { + return recoveredXids.contains(XidKey.from(checkpointXid)); + } + + /** + * Adds a bounded pause between synchronous retry rounds so transient RM outages do not consume + * the whole retry budget immediately. + * + * @param operation operation being retried + * @param nextAttempt 1-based retry attempt number that will run after the pause + * @param maxAttempts bounded retry budget + */ + private void backoffBeforeRetry(String operation, int nextAttempt, int maxAttempts) { + try { + Thread.sleep(XA_RETRY_BACKOFF_MILLIS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new JdbcConnectorException( + CommonErrorCodeDeprecated.WRITER_OPERATION_FAILED, + String.format( + "Interrupted while waiting to retry XA %s on attempt %d/%d", + operation, nextAttempt, maxAttempts), + e); + } + } + + /** Canonical XID value used to compare driver-specific {@link Xid} implementations. */ + private static final class XidKey { + private final int formatId; + private final byte[] globalTransactionId; + private final byte[] branchQualifier; + + private XidKey(int formatId, byte[] globalTransactionId, byte[] branchQualifier) { + this.formatId = formatId; + this.globalTransactionId = + Arrays.copyOf(globalTransactionId, globalTransactionId.length); + this.branchQualifier = Arrays.copyOf(branchQualifier, branchQualifier.length); + } + + private static XidKey from(Xid xid) { + return new XidKey( + xid.getFormatId(), xid.getGlobalTransactionId(), xid.getBranchQualifier()); + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof XidKey)) { + return false; + } + XidKey xidKey = (XidKey) o; + return formatId == xidKey.formatId + && Arrays.equals(globalTransactionId, xidKey.globalTransactionId) + && Arrays.equals(branchQualifier, xidKey.branchQualifier); + } + + @Override + public int hashCode() { + int result = Objects.hash(formatId); + result = 31 * result + Arrays.hashCode(globalTransactionId); + result = 31 * result + Arrays.hashCode(branchQualifier); + return result; + } + } } diff --git a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/xa/XaFacadeImplAutoLoadTest.java b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/xa/XaFacadeImplAutoLoadTest.java new file mode 100644 index 0000000000..c6322d3f2a --- /dev/null +++ b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/xa/XaFacadeImplAutoLoadTest.java @@ -0,0 +1,145 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.jdbc.internal.xa; + +import org.apache.seatunnel.connectors.seatunnel.jdbc.exception.JdbcConnectorException; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import javax.transaction.xa.XAException; +import javax.transaction.xa.XAResource; +import javax.transaction.xa.Xid; + +import java.lang.reflect.Field; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +/** + * Tests XA error classification and explicit recovery outcomes without requiring a database + * resource manager. + */ +class XaFacadeImplAutoLoadTest { + + /** + * Verifies that resource-manager availability errors remain retryable instead of being wrapped + * as permanent connector failures. + */ + @Test + void testCommitPreservesTransientXaException() throws Exception { + XAResource xaResource = mock(XAResource.class); + XAException transientFailure = new XAException(XAException.XAER_RMFAIL); + doThrow(transientFailure).when(xaResource).commit(any(), eq(false)); + XaFacadeImplAutoLoad xaFacade = createOpenFacade(xaResource); + + XaFacade.TransientXaException exception = + Assertions.assertThrows( + XaFacade.TransientXaException.class, + () -> xaFacade.commit(createXid(), false)); + + Assertions.assertSame(transientFailure, exception.getCause()); + } + + /** + * Verifies that XA_RETRY remains retryable because the resource manager reports that the + * operation had no effect. + */ + @Test + void testCommitRetriesWhenResourceManagerRequestsRetry() throws Exception { + XAResource xaResource = mock(XAResource.class); + XAException retry = new XAException(XAException.XA_RETRY); + doThrow(retry).when(xaResource).commit(any(), eq(false)); + XaFacadeImplAutoLoad xaFacade = createOpenFacade(xaResource); + + XaFacade.TransientXaException exception = + Assertions.assertThrows( + XaFacade.TransientXaException.class, + () -> xaFacade.commit(createXid(), false)); + + Assertions.assertSame(retry, exception.getCause()); + } + + /** + * Verifies that an explicit heuristic-commit result is accepted as durable completion evidence. + */ + @Test + void testCommitAcceptsHeuristicCommitResult() throws Exception { + XAResource xaResource = mock(XAResource.class); + XAException heuristicallyCommitted = new XAException(XAException.XA_HEURCOM); + doThrow(heuristicallyCommitted).when(xaResource).commit(any(), eq(false)); + XaFacadeImplAutoLoad xaFacade = createOpenFacade(xaResource); + + Assertions.assertDoesNotThrow(() -> xaFacade.commit(createXid(), false)); + + verify(xaResource).forget(any()); + } + + /** + * Verifies that a transaction reported as rolled back is a permanent failure and is never + * retried as though its commit outcome were unknown. + */ + @Test + void testCommitDoesNotRetryRolledBackTransaction() throws Exception { + XAResource xaResource = mock(XAResource.class); + XAException rolledBack = new XAException(XAException.XA_RBTRANSIENT); + doThrow(rolledBack).when(xaResource).commit(any(), eq(false)); + XaFacadeImplAutoLoad xaFacade = createOpenFacade(xaResource); + + Assertions.assertThrows( + JdbcConnectorException.class, () -> xaFacade.commit(createXid(), false)); + } + + /** + * Verifies that XAER_NOTA is ignored only when recovery is repeating a commit whose response + * may have been lost. + */ + @Test + void testCommitIgnoresUnknownTransactionOnlyWhenRequested() throws Exception { + XAResource xaResource = mock(XAResource.class); + XAException unknownTransaction = new XAException(XAException.XAER_NOTA); + doThrow(unknownTransaction).when(xaResource).commit(any(), eq(false)); + XaFacadeImplAutoLoad xaFacade = createOpenFacade(xaResource); + Xid xid = createXid(); + + Assertions.assertDoesNotThrow(() -> xaFacade.commit(xid, true)); + Assertions.assertThrows(JdbcConnectorException.class, () -> xaFacade.commit(xid, false)); + } + + /** + * Creates a facade with an injected XA resource so error handling can be tested without a + * database. + */ + private XaFacadeImplAutoLoad createOpenFacade(XAResource xaResource) throws Exception { + XaFacadeImplAutoLoad xaFacade = new XaFacadeImplAutoLoad(null); + Field field = XaFacadeImplAutoLoad.class.getDeclaredField("xaResource"); + field.setAccessible(true); + field.set(xaFacade, xaResource); + return xaFacade; + } + + /** + * Creates a stable transaction identifier whose value is shared across the commit test cases. + */ + private Xid createXid() { + return new XidImpl(1, new byte[] {1}, new byte[] {1}); + } +} diff --git a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/xa/XaGroupOpsImplTest.java b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/xa/XaGroupOpsImplTest.java new file mode 100644 index 0000000000..64a0ee4a8e --- /dev/null +++ b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/xa/XaGroupOpsImplTest.java @@ -0,0 +1,122 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.jdbc.internal.xa; + +import org.apache.seatunnel.connectors.seatunnel.jdbc.exception.JdbcConnectorException; +import org.apache.seatunnel.connectors.seatunnel.jdbc.state.XidInfo; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import javax.transaction.xa.XAException; +import javax.transaction.xa.Xid; + +import java.util.ArrayList; +import java.util.Collections; + +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; + +/** + * Tests grouped XA commit failure propagation and retry accounting. + * + * <p>Each case verifies strict resource-manager commit handling for prepared transaction batches. + */ +class XaGroupOpsImplTest { + + /** + * Verifies that a permanent commit failure propagates instead of being reported as a successful + * checkpoint. + */ + @Test + void testCommitPropagatesPermanentFailure() { + XaFacade xaFacade = mock(XaFacade.class); + Xid xid = createXid(); + RuntimeException commitFailure = new RuntimeException("permanent commit failure"); + doThrow(commitFailure).when(xaFacade).commit(xid, false); + XaGroupOps xaGroupOps = new XaGroupOpsImpl(xaFacade); + + JdbcConnectorException exception = + Assertions.assertThrows( + JdbcConnectorException.class, + () -> + xaGroupOps.commit( + new ArrayList<>( + Collections.singletonList(new XidInfo(xid, 0))), + false, + 3)); + + Assertions.assertSame(commitFailure, exception.getCause()); + } + + /** + * Verifies that a transient failure is returned for retry with an incremented attempt count. + */ + @Test + void testCommitReturnsTransientFailureForRetry() { + XaFacade xaFacade = mock(XaFacade.class); + Xid xid = createXid(); + XAException transientCause = new XAException(XAException.XAER_RMFAIL); + doThrow(new XaFacade.TransientXaException(transientCause)) + .when(xaFacade) + .commit(xid, false); + XaGroupOps xaGroupOps = new XaGroupOpsImpl(xaFacade); + + GroupXaOperationResult<XidInfo> result = + xaGroupOps.commit( + new ArrayList<>(Collections.singletonList(new XidInfo(xid, 0))), false, 3); + + Assertions.assertEquals(1, result.getForRetry().size()); + Assertions.assertEquals(1, result.getForRetry().get(0).getAttempts()); + Assertions.assertSame(xid, result.getForRetry().get(0).getXid()); + } + + /** + * Verifies that retry exhaustion fails the commit instead of silently discarding the + * transaction. + */ + @Test + void testCommitFailsWhenTransientRetryLimitIsReached() { + XaFacade xaFacade = mock(XaFacade.class); + Xid xid = createXid(); + XAException transientCause = new XAException(XAException.XAER_RMFAIL); + doThrow(new XaFacade.TransientXaException(transientCause)) + .when(xaFacade) + .commit(xid, false); + XaGroupOps xaGroupOps = new XaGroupOpsImpl(xaFacade); + + JdbcConnectorException exception = + Assertions.assertThrows( + JdbcConnectorException.class, + () -> + xaGroupOps.commit( + new ArrayList<>( + Collections.singletonList(new XidInfo(xid, 2))), + false, + 3)); + + Assertions.assertTrue(exception.getMessage().contains("reached max number")); + } + + /** + * Creates a stable transaction identifier whose value is shared across grouped commit tests. + */ + private Xid createXid() { + return new XidImpl(1, new byte[] {1}, new byte[] {1}); + } +} diff --git a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkAggregatedCommitterTest.java b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkAggregatedCommitterTest.java new file mode 100644 index 0000000000..50310ddf33 --- /dev/null +++ b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkAggregatedCommitterTest.java @@ -0,0 +1,440 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.jdbc.sink; + +import org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcConnectionConfig; +import org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcSinkConfig; +import org.apache.seatunnel.connectors.seatunnel.jdbc.exception.JdbcConnectorException; +import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.xa.GroupXaOperationResult; +import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.xa.XaFacade; +import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.xa.XaGroupOps; +import org.apache.seatunnel.connectors.seatunnel.jdbc.state.JdbcAggregatedCommitInfo; +import org.apache.seatunnel.connectors.seatunnel.jdbc.state.XidInfo; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import javax.transaction.xa.XAException; +import javax.transaction.xa.Xid; + +import java.lang.reflect.Field; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.argThat; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Tests normal and recovery commit semantics, including RM reconciliation, for the JDBC aggregated + * committer. + */ +class JdbcSinkAggregatedCommitterTest { + + /** + * Verifies that recovery matches driver-specific XIDs by value and never commits an unrelated + * recovered transaction. + */ + @Test + void testRestoreCommitReconcilesRecoveredTransactionsByValue() throws Exception { + JdbcConnectionConfig connectionConfig = + JdbcConnectionConfig.builder().maxCommitAttempts(3).build(); + JdbcSinkConfig sinkConfig = + JdbcSinkConfig.builder().jdbcConnectionConfig(connectionConfig).build(); + JdbcSinkAggregatedCommitter committer = new JdbcSinkAggregatedCommitter(sinkConfig); + XaFacade xaFacade = mock(XaFacade.class); + XaGroupOps xaGroupOps = mock(XaGroupOps.class); + when(xaFacade.isOpen()).thenReturn(true); + Xid checkpointXid = createXid(1, new byte[] {1}, new byte[] {2}); + Xid recoveredXid = createXid(1, new byte[] {1}, new byte[] {2}); + Xid unrelatedXid = createXid(2, new byte[] {3}, new byte[] {4}); + when(xaFacade.recover()).thenReturn(Arrays.asList(recoveredXid, unrelatedXid)); + when(xaGroupOps.commit(anyList(), eq(false), eq(3))) + .thenReturn(new GroupXaOperationResult<>()); + setPrivateField(committer, "xaFacade", xaFacade); + setPrivateField(committer, "xaGroupOps", xaGroupOps); + JdbcAggregatedCommitInfo commitInfo = + new JdbcAggregatedCommitInfo( + Collections.singletonList(new XidInfo(checkpointXid, 0))); + + committer.restoreCommit(Collections.singletonList(commitInfo)); + + verify(xaFacade).recover(); + verify(xaGroupOps) + .commit( + argThat(xids -> xids.size() == 1 && xids.get(0).getXid() == checkpointXid), + eq(false), + eq(3)); + } + + /** + * Verifies that each restored batch uses a fresh recovery scan instead of reusing stale + * resource-manager evidence from an earlier batch. + */ + @Test + void testRestoreCommitRefreshesRecoveryScanForEachBatch() throws Exception { + JdbcSinkAggregatedCommitter committer = createCommitter(); + XaFacade xaFacade = mock(XaFacade.class); + XaGroupOps xaGroupOps = mock(XaGroupOps.class); + Xid firstCheckpointXid = createXid(1, new byte[] {1}, new byte[] {1}); + Xid secondCheckpointXid = createXid(2, new byte[] {2}, new byte[] {2}); + Xid firstRecoveredXid = createXid(1, new byte[] {1}, new byte[] {1}); + Xid secondRecoveredXid = createXid(2, new byte[] {2}, new byte[] {2}); + when(xaFacade.isOpen()).thenReturn(true); + when(xaFacade.recover()) + .thenReturn(Arrays.asList(firstRecoveredXid, secondRecoveredXid)) + .thenReturn(Collections.emptyList()); + when(xaGroupOps.commit(anyList(), eq(false), eq(3))) + .thenReturn(new GroupXaOperationResult<>()); + setPrivateField(committer, "xaFacade", xaFacade); + setPrivateField(committer, "xaGroupOps", xaGroupOps); + JdbcAggregatedCommitInfo firstCommitInfo = + new JdbcAggregatedCommitInfo( + Collections.singletonList(new XidInfo(firstCheckpointXid, 0))); + JdbcAggregatedCommitInfo secondCommitInfo = + new JdbcAggregatedCommitInfo( + Collections.singletonList(new XidInfo(secondCheckpointXid, 0))); + + committer.restoreCommit(Arrays.asList(firstCommitInfo, secondCommitInfo)); + + verify(xaFacade, times(2)).recover(); + verify(xaGroupOps) + .commit( + argThat( + xids -> + xids.size() == 1 + && xids.get(0).getXid() == firstCheckpointXid), + eq(false), + eq(3)); + verify(xaGroupOps, never()) + .commit( + argThat( + xids -> + xids.stream() + .anyMatch( + xidInfo -> + xidInfo.getXid() + == secondCheckpointXid)), + eq(false), + eq(3)); + } + + /** + * Verifies that a missing prefix is skipped only after the still-prepared suffix commits + * strictly. + */ + @Test + void testRestoreCommitSkipsMissingPrefixAfterRecoveredSuffixCommits() throws Exception { + JdbcSinkAggregatedCommitter committer = createCommitter(); + XaFacade xaFacade = mock(XaFacade.class); + XaGroupOps xaGroupOps = mock(XaGroupOps.class); + Xid alreadyResolvedXid = createXid(1, new byte[] {1}, new byte[] {1}); + Xid stillPreparedXid = createXid(2, new byte[] {2}, new byte[] {2}); + Xid recoveredStillPreparedXid = createXid(2, new byte[] {2}, new byte[] {2}); + when(xaFacade.isOpen()).thenReturn(true); + when(xaFacade.recover()).thenReturn(Collections.singletonList(recoveredStillPreparedXid)); + when(xaGroupOps.commit(anyList(), eq(false), eq(3))) + .thenReturn(new GroupXaOperationResult<>()); + setPrivateField(committer, "xaFacade", xaFacade); + setPrivateField(committer, "xaGroupOps", xaGroupOps); + JdbcAggregatedCommitInfo commitInfo = + new JdbcAggregatedCommitInfo( + Arrays.asList( + new XidInfo(alreadyResolvedXid, 0), + new XidInfo(stillPreparedXid, 0))); + + Assertions.assertDoesNotThrow( + () -> committer.restoreCommit(Collections.singletonList(commitInfo))); + + verify(xaGroupOps) + .commit( + argThat( + xids -> + xids.size() == 1 + && xids.get(0).getXid() == stillPreparedXid), + eq(false), + eq(3)); + verify(xaGroupOps, never()) + .commit( + argThat( + xids -> + xids.stream() + .anyMatch( + xidInfo -> + xidInfo.getXid() + == alreadyResolvedXid)), + eq(false), + eq(3)); + } + + /** + * Verifies that restore treats an all-absent checkpoint batch as already resolved instead of + * replaying or failing it again. + */ + @Test + void testRestoreCommitSkipsAlreadyResolvedBatchWhenRecoveryScanHasNoCheckpointXid() + throws Exception { + JdbcSinkAggregatedCommitter committer = createCommitter(); + XaFacade xaFacade = mock(XaFacade.class); + XaGroupOps xaGroupOps = mock(XaGroupOps.class); + when(xaFacade.isOpen()).thenReturn(true); + when(xaFacade.recover()).thenReturn(Collections.emptyList()); + setPrivateField(committer, "xaFacade", xaFacade); + setPrivateField(committer, "xaGroupOps", xaGroupOps); + JdbcAggregatedCommitInfo commitInfo = + new JdbcAggregatedCommitInfo( + Arrays.asList( + new XidInfo(createXid(1, new byte[] {1}, new byte[] {2}), 0), + new XidInfo(createXid(2, new byte[] {3}, new byte[] {4}), 0))); + + Assertions.assertDoesNotThrow( + () -> committer.restoreCommit(Collections.singletonList(commitInfo))); + + verify(xaGroupOps, never()).commit(anyList(), eq(false), eq(3)); + } + + /** + * Verifies that restore reaches the still-prepared XID after an earlier XID was already + * resolved by a previous failed attempt. + */ + @Test + void testRestoreCommitReportsRecoveredFailureBeforeAlreadyResolvedReplay() throws Exception { + JdbcSinkAggregatedCommitter committer = createCommitter(); + XaFacade xaFacade = mock(XaFacade.class); + XaGroupOps xaGroupOps = mock(XaGroupOps.class); + RuntimeException stillPreparedFailure = new RuntimeException("still prepared failed"); + Xid alreadyResolvedXid = createXid(1, new byte[] {1}, new byte[] {1}); + Xid stillPreparedXid = createXid(2, new byte[] {2}, new byte[] {2}); + Xid recoveredStillPreparedXid = createXid(2, new byte[] {2}, new byte[] {2}); + when(xaFacade.isOpen()).thenReturn(true); + when(xaFacade.recover()).thenReturn(Collections.singletonList(recoveredStillPreparedXid)); + when(xaGroupOps.commit( + argThat( + xids -> + xids.size() == 1 + && xids.get(0).getXid() == stillPreparedXid), + eq(false), + eq(3))) + .thenThrow(stillPreparedFailure); + setPrivateField(committer, "xaFacade", xaFacade); + setPrivateField(committer, "xaGroupOps", xaGroupOps); + JdbcAggregatedCommitInfo commitInfo = + new JdbcAggregatedCommitInfo( + Arrays.asList( + new XidInfo(alreadyResolvedXid, 0), + new XidInfo(stillPreparedXid, 0))); + + RuntimeException exception = + Assertions.assertThrows( + RuntimeException.class, + () -> committer.restoreCommit(Collections.singletonList(commitInfo))); + + Assertions.assertSame(stillPreparedFailure, exception); + verify(xaGroupOps) + .commit( + argThat( + xids -> + xids.size() == 1 + && xids.get(0).getXid() == stillPreparedXid), + eq(false), + eq(3)); + verify(xaGroupOps, never()) + .commit( + argThat( + xids -> + xids.stream() + .anyMatch( + xidInfo -> + xidInfo.getXid() + == alreadyResolvedXid)), + eq(false), + eq(3)); + } + + /** + * Verifies that a missing XID after the first still-prepared transaction fails closed because + * the batch order cannot explain it as an earlier successful commit. + */ + @Test + void testRestoreCommitFailsClosedOnMissingGapAfterRecoveredTransaction() throws Exception { + JdbcSinkAggregatedCommitter committer = createCommitter(); + XaFacade xaFacade = mock(XaFacade.class); + XaGroupOps xaGroupOps = mock(XaGroupOps.class); + Xid stillPreparedXid = createXid(1, new byte[] {1}, new byte[] {1}); + Xid missingTailXid = createXid(2, new byte[] {2}, new byte[] {2}); + Xid recoveredStillPreparedXid = createXid(1, new byte[] {1}, new byte[] {1}); + when(xaFacade.isOpen()).thenReturn(true); + when(xaFacade.recover()).thenReturn(Collections.singletonList(recoveredStillPreparedXid)); + setPrivateField(committer, "xaFacade", xaFacade); + setPrivateField(committer, "xaGroupOps", xaGroupOps); + JdbcAggregatedCommitInfo commitInfo = + new JdbcAggregatedCommitInfo( + Arrays.asList( + new XidInfo(stillPreparedXid, 0), new XidInfo(missingTailXid, 0))); + + JdbcConnectorException exception = + Assertions.assertThrows( + JdbcConnectorException.class, + () -> committer.restoreCommit(Collections.singletonList(commitInfo))); + + Assertions.assertTrue( + exception + .getMessage() + .contains( + "is absent from the XA recovery scan after still-prepared transactions")); + verify(xaGroupOps, never()).commit(anyList(), eq(false), eq(3)); + } + + /** + * Verifies that restore retries a transient recovery-scan failure before replaying the + * checkpoint transactions. + */ + @Test + void testRestoreCommitRetriesTransientRecoveryScanFailure() throws Exception { + JdbcSinkAggregatedCommitter committer = createCommitter(); + XaFacade xaFacade = mock(XaFacade.class); + XaGroupOps xaGroupOps = mock(XaGroupOps.class); + XAException transientCause = new XAException(XAException.XAER_RMFAIL); + Xid checkpointXid = createXid(1, new byte[] {1}, new byte[] {1}); + Xid recoveredXid = createXid(1, new byte[] {1}, new byte[] {1}); + when(xaFacade.isOpen()).thenReturn(true); + when(xaFacade.recover()) + .thenThrow(new XaFacade.TransientXaException(transientCause)) + .thenReturn(Collections.singletonList(recoveredXid)); + when(xaGroupOps.commit(anyList(), eq(false), eq(3))) + .thenReturn(new GroupXaOperationResult<>()); + setPrivateField(committer, "xaFacade", xaFacade); + setPrivateField(committer, "xaGroupOps", xaGroupOps); + JdbcAggregatedCommitInfo commitInfo = + new JdbcAggregatedCommitInfo( + Collections.singletonList(new XidInfo(checkpointXid, 0))); + + Assertions.assertDoesNotThrow( + () -> committer.restoreCommit(Collections.singletonList(commitInfo))); + + verify(xaFacade, times(2)).recover(); + verify(xaGroupOps).commit(anyList(), eq(false), eq(3)); + } + + /** + * Verifies that Zeta retries are completed while the incremented attempt state is available. + */ + @Test + void testCommitCompletesBoundedRetriesWithinInvocation() throws Exception { + JdbcConnectionConfig connectionConfig = + JdbcConnectionConfig.builder().maxCommitAttempts(3).build(); + JdbcSinkConfig sinkConfig = + JdbcSinkConfig.builder().jdbcConnectionConfig(connectionConfig).build(); + JdbcSinkAggregatedCommitter committer = new JdbcSinkAggregatedCommitter(sinkConfig); + XaFacade xaFacade = mock(XaFacade.class); + XaGroupOps xaGroupOps = mock(XaGroupOps.class); + when(xaFacade.isOpen()).thenReturn(true); + when(xaGroupOps.commit(anyList(), eq(false), eq(3))) + .thenAnswer( + invocation -> { + XidInfo current = invocation.<List<XidInfo>>getArgument(0).get(0); + GroupXaOperationResult<XidInfo> result = new GroupXaOperationResult<>(); + if (current.getAttempts() < 2) { + result.getForRetry().add(current.withAttemptsIncremented()); + } + return result; + }); + setPrivateField(committer, "xaFacade", xaFacade); + setPrivateField(committer, "xaGroupOps", xaGroupOps); + JdbcAggregatedCommitInfo commitInfo = + new JdbcAggregatedCommitInfo( + Collections.singletonList(new XidInfo(mock(Xid.class), 0))); + + committer.commit(Collections.singletonList(commitInfo)); + + verify(xaGroupOps, times(3)).commit(anyList(), eq(false), eq(3)); + } + + /** + * Verifies that the committer fails fast if an alternative XaGroupOps implementation never + * drains or escalates the retry list. + */ + @Test + void testCommitFailsWhenRetryRoundsNeverDrain() throws Exception { + JdbcSinkAggregatedCommitter committer = createCommitter(); + XaFacade xaFacade = mock(XaFacade.class); + XaGroupOps xaGroupOps = mock(XaGroupOps.class); + Xid xid = mock(Xid.class); + when(xaFacade.isOpen()).thenReturn(true); + when(xaGroupOps.commit(anyList(), eq(false), eq(3))) + .thenAnswer( + invocation -> { + GroupXaOperationResult<XidInfo> result = new GroupXaOperationResult<>(); + result.getForRetry().addAll(invocation.getArgument(0)); + return result; + }); + setPrivateField(committer, "xaFacade", xaFacade); + setPrivateField(committer, "xaGroupOps", xaGroupOps); + JdbcAggregatedCommitInfo commitInfo = + new JdbcAggregatedCommitInfo(Collections.singletonList(new XidInfo(xid, 0))); + + JdbcConnectorException exception = + Assertions.assertThrows( + JdbcConnectorException.class, + () -> committer.commit(Collections.singletonList(commitInfo))); + + Assertions.assertTrue(exception.getMessage().contains("did not terminate")); + verify(xaGroupOps, times(3)).commit(anyList(), eq(false), eq(3)); + } + + /** + * Creates a committer with the same explicit retry limit used by every recovery test case. + * + * @return an uninitialized committer whose collaborators can be injected by the test + */ + private JdbcSinkAggregatedCommitter createCommitter() { + JdbcConnectionConfig connectionConfig = + JdbcConnectionConfig.builder().maxCommitAttempts(3).build(); + JdbcSinkConfig sinkConfig = + JdbcSinkConfig.builder().jdbcConnectionConfig(connectionConfig).build(); + return new JdbcSinkAggregatedCommitter(sinkConfig); + } + + /** + * Creates a driver-neutral XID value whose canonical components can be compared during + * recovery. + */ + private Xid createXid(int formatId, byte[] globalTransactionId, byte[] branchQualifier) { + Xid xid = mock(Xid.class); + when(xid.getFormatId()).thenReturn(formatId); + when(xid.getGlobalTransactionId()).thenReturn(globalTransactionId); + when(xid.getBranchQualifier()).thenReturn(branchQualifier); + return xid; + } + + /** + * Injects a test dependency through the existing private field without changing production + * constructors. + */ + private void setPrivateField(Object target, String fieldName, Object value) throws Exception { + Field field = target.getClass().getDeclaredField(fieldName); + field.setAccessible(true); + field.set(target, value); + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/xa/XaGroupOpsImplIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/xa/XaGroupOpsImplIT.java index 28e34b656c..d833e024d3 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/xa/XaGroupOpsImplIT.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/xa/XaGroupOpsImplIT.java @@ -22,15 +22,20 @@ import org.apache.seatunnel.api.common.JobContext; import org.apache.seatunnel.api.sink.DefaultSinkWriterContext; import org.apache.seatunnel.api.sink.SinkWriter; import org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcConnectionConfig; +import org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcSinkConfig; +import org.apache.seatunnel.connectors.seatunnel.jdbc.exception.JdbcConnectorException; import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.connection.DataSourceUtils; +import org.apache.seatunnel.connectors.seatunnel.jdbc.sink.JdbcSinkAggregatedCommitter; +import org.apache.seatunnel.connectors.seatunnel.jdbc.state.JdbcAggregatedCommitInfo; +import org.apache.seatunnel.connectors.seatunnel.jdbc.state.XidInfo; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import org.testcontainers.containers.MySQLContainer; import org.testcontainers.containers.output.Slf4jLogConsumer; +import org.testcontainers.images.PullPolicy; import org.testcontainers.lifecycle.Startables; import org.testcontainers.utility.DockerImageName; import org.testcontainers.utility.DockerLoggerFactory; @@ -42,13 +47,13 @@ import javax.transaction.xa.XAException; import javax.transaction.xa.XAResource; import javax.transaction.xa.Xid; +import java.lang.reflect.Field; +import java.util.Collections; import java.util.stream.Stream; import static javax.transaction.xa.XAResource.TMSTARTRSCAN; @Slf4j -@Disabled( - "Temporary fast fix, reason: JdbcDatabaseContainer: ClassNotFoundException: com.mysql.jdbc.Driver") class XaGroupOpsImplIT { private static final String MYSQL_DOCKER_IMAGE = "mysql:8.0.43"; @@ -66,6 +71,14 @@ class XaGroupOpsImplIT { mc = new MySQLContainer<>(DockerImageName.parse(MYSQL_DOCKER_IMAGE)) .withUsername("root") + // AbstractJdbcIT#tearDown removes the database image from the Docker + // daemon after every JDBC IT (JdbcMysqlIT uses this same image and runs + // in the same surefire JVM right before this class). Testcontainers + // caches "image is present" per JVM, so the default pull policy skips + // the pull and container creation fails with + // "No such image: mysql:8.0.43". Always pull so the image is re-fetched + // whenever a previous test removed it. + .withImagePullPolicy(PullPolicy.alwaysPull()) .withLogConsumer( new Slf4jLogConsumer( DockerLoggerFactory.getLogger(MYSQL_DOCKER_IMAGE))); @@ -114,6 +127,33 @@ class XaGroupOpsImplIT { Assertions.assertTrue(checkPreparedXid(xid2)); } + /** + * Verifies that a real MySQL XA commit failure bubbles up through the aggregated committer + * instead of being silently swallowed when the resource manager becomes unavailable. + */ + @Test + void testCommitFailurePropagatesThroughAggregatedCommitter() throws Exception { + JobContext jobContext = new JobContext(); + SinkWriter.Context writerContext = new DefaultSinkWriterContext(1, 1); + Xid preparedXid = + xidGenerator.generateXid(jobContext, writerContext, System.currentTimeMillis()); + + xaFacade.start(preparedXid); + xaFacade.endAndPrepare(preparedXid); + Assertions.assertTrue(checkPreparedXid(preparedXid)); + + JdbcSinkAggregatedCommitter committer = createCommitterWithCurrentXaFacade(); + + mc.stop(); + + JdbcAggregatedCommitInfo commitInfo = + new JdbcAggregatedCommitInfo( + Collections.singletonList(new XidInfo(preparedXid, 0))); + Assertions.assertThrows( + JdbcConnectorException.class, + () -> committer.commit(Collections.singletonList(commitInfo))); + } + private boolean checkPreparedXid(Xid xidCrr) throws XAException { Xid[] recover = xaResource.recover(TMSTARTRSCAN); for (Xid value : recover) { @@ -129,8 +169,35 @@ class XaGroupOpsImplIT { return false; } + /** + * Reuses the prepared, already-open XA facade so the test can verify failure propagation on the + * exact commit path that handles checkpoint-owned prepared transactions. + */ + private JdbcSinkAggregatedCommitter createCommitterWithCurrentXaFacade() throws Exception { + JdbcSinkConfig sinkConfig = + JdbcSinkConfig.builder().jdbcConnectionConfig(jdbcConnectionConfig).build(); + JdbcSinkAggregatedCommitter committer = new JdbcSinkAggregatedCommitter(sinkConfig); + setPrivateField(committer, "xaFacade", xaFacade); + setPrivateField(committer, "xaGroupOps", new XaGroupOpsImpl(xaFacade)); + return committer; + } + + /** Injects an integration-test collaborator without widening the production constructor API. */ + private void setPrivateField(Object target, String fieldName, Object value) throws Exception { + Field field = target.getClass().getDeclaredField(fieldName); + field.setAccessible(true); + field.set(target, value); + } + @AfterEach - public void closePostgreSqlContainer() { + public void closeResources() { + if (xaFacade != null && xaFacade.isOpen()) { + try { + xaFacade.close(); + } catch (Exception e) { + log.warn("Unable to close XA facade after test", e); + } + } if (mc != null) { mc.stop(); }
