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-11885-29df45855c65beaaffbf7fad2db67954231f12d6 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 67941ddefacebbcadab6ba72f7e3877d10cbaee9 Author: Li Zhi <[email protected]> AuthorDate: Wed Oct 14 11:03:03 2026 +0000 [Fix][Connector-CDC] Resolve stop.mode=latest stop offset at binlog phase start (#11885) Co-authored-by: AtomCode (deepseek-v4-flash) <[email protected]> --- docs/en/connectors/source/MySQL-CDC.md | 22 ++- docs/zh/connectors/source/MySQL-CDC.md | 21 ++- .../enumerator/IncrementalSplitAssigner.java | 92 +++++++++++- .../enumerator/state/IncrementalPhaseState.java | 15 +- .../source/enumerator/HybridSplitAssignerTest.java | 15 +- .../enumerator/IncrementalSplitAssignerTest.java | 83 +++++++++++ .../state/IncrementalPhaseStateTest.java | 91 ++++++++++++ .../cdc/mysql/MysqlCDCStopModeSpecificIT.java | 162 +++++++++++++++++++++ .../test/resources/mysqlcdc_stop_mode_latest.conf | 56 +++++++ .../mysqlcdc_stop_mode_latest_specific.conf | 51 +++++++ .../mysqlcdc_stop_mode_latest_timestamp.conf | 50 +++++++ 11 files changed, 650 insertions(+), 8 deletions(-) diff --git a/docs/en/connectors/source/MySQL-CDC.md b/docs/en/connectors/source/MySQL-CDC.md index 8c73533b18..ee7734c318 100644 --- a/docs/en/connectors/source/MySQL-CDC.md +++ b/docs/en/connectors/source/MySQL-CDC.md @@ -200,7 +200,7 @@ When an initial consistent snapshot is made for large databases, your establishe | startup.specific-offset.skip-events | Long | No | 0 | Number of binlog events to skip after the configured specific startup offset. This option can only be used when `startup.mode` is `specific`. [...] | startup.specific-offset.skip-rows | Long | No | 0 | Number of rows to skip after the configured specific startup offset. This option can only be used when `startup.mode` is `specific`. [...] | startup.timestamp | Long | No | - | Start from the specified timestamp, in milliseconds since Unix epoch. **Note, This option is required when the `startup.mode` option uses `timestamp`.** [...] -| stop.mode | Enum | No | NEVER | Optional stop mode for MySQL CDC consumer, valid enumerations are `never`, `latest` or `specific`. <br/> `never`: Real-time job don't stop the source.<br/> `latest`: Stop from the latest offset.<br/> `specific`: Stop from user-supplied specific offset. [...] +| stop.mode | Enum | No | NEVER | Optional stop mode for MySQL CDC consumer, valid enumerations are `never`, `latest` or `specific`. <br/> `never`: Real-time job don't stop the source.<br/> `latest`: Stop from the latest offset. When combined with a snapshot-taking startup mode (`initial`/`earliest`), the stop offset is resolved once the snapshot phase completes, so changes written to the source while the snapshot is running are still captured [...] | stop.specific-offset.file | String | No | - | Stop from the specified binlog file name. **Note, This option is required when the `stop.mode` option used `specific`.** [...] | stop.specific-offset.pos | Long | No | - | Stop from the specified binlog file position. **Note, This option is required when the `stop.mode` option used `specific`.** [...] | snapshot.split.size | Integer | No | 8096 | The split size (number of rows) of table snapshot, captured tables are split into multiple splits when read the snapshot of table. [...] @@ -541,6 +541,26 @@ startup offset (or startup timestamp) and the configured stop offset, then termi > **Note**: bounded-read termination is currently supported on the **Zeta** > engine only. > Flink and Spark engines do not support bounded incremental-split termination > yet. +`stop.mode = "latest"` also makes the job a bounded read: the stop offset is the binlog +position at the moment the snapshot phase completes (for `startup.mode = "initial"`/`"earliest"`) +or at job start (for the other startup modes). In particular, when combined with a +snapshot-taking startup mode, changes written to the source **while the snapshot is running +are not dropped** — they are captured by the binlog phase before the job terminates. + +```hocon +source { + MySQL-CDC { + server-id = 5654 + username = "st_user_source" + password = "mysqlpw" + table-names = ["mysql_cdc.mysql_cdc_e2e_source_table"] + url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc" + startup.mode = "initial" + stop.mode = "latest" + } +} +``` + ```hocon source { MySQL-CDC { diff --git a/docs/zh/connectors/source/MySQL-CDC.md b/docs/zh/connectors/source/MySQL-CDC.md index 8a7b26c905..1aee4c39b5 100644 --- a/docs/zh/connectors/source/MySQL-CDC.md +++ b/docs/zh/connectors/source/MySQL-CDC.md @@ -199,7 +199,7 @@ show variables where variable_name in ('log_bin', 'binlog_format', 'binlog_row_i | startup.specific-offset.skip-events | Long | 否 | 0 | 配置 specific 启动偏移量后需要跳过的 binlog event 数量. 该选项只能在 `startup.mode` 为 `specific` 时使用. | | startup.specific-offset.skip-rows | Long | 否 | 0 | 配置 specific 启动偏移量后需要跳过的行数. 该选项只能在 `startup.mode` 为 `specific` 时使用. | | startup.timestamp | Long | 否 | - | 从指定时间戳启动，单位为 Unix 纪元以来的毫秒数。**注意，当 `startup.mode` 为 `timestamp` 时，此选项必填。** | -| stop.mode | Enum | 否 | NEVER | MySQL CDC 消费者的可选停止模式, 有效枚举值为 `never`, `latest` 和 `specific`. <br/> `never`: 实时任务一直运行不停止.<br/> `latest`: 从最新的偏移量处停止.<br/> `specific`: 从用户提供的特定偏移量处停止. | +| stop.mode | Enum | 否 | NEVER | MySQL CDC 消费者的可选停止模式, 有效枚举值为 `never`, `latest` 和 `specific`. <br/> `never`: 实时任务一直运行不停止.<br/> `latest`: 从最新的偏移量处停止. 与需要执行快照的启动模式（`initial`/`earliest`）组合使用时, 停止偏移量在快照阶段完成后解析, 因此快照运行期间写入源表的变更仍会被 binlog 阶段捕获, 不会丢失.<br/> `specific`: 从用户提供的特定偏移量处停止. | | stop.specific-offset.file | String | 否 | - | 从指定的binlog日志文件名停止. **注意, 当使用 `stop.mode` 选项为 `specific` 时，此选项为必填项.** | | stop.specific-offset.pos | Long | 否 | - | 从指定的binlog日志文件位置停止. **注意, 当使用 `stop.mode` 选项为 `specific` 时，此选项为必填项.** | | snapshot.split.size | Integer | 否 | 8096 | 表快照的分割大小（行数）,读取表的快照时,被捕获的表会被分割成多个分割块. | @@ -533,6 +533,25 @@ source { > **注意**：有界读取的终止行为目前仅在 **Zeta** 引擎上支持。 > Flink 和 Spark 引擎暂不支持有界增量分片的终止。 +`stop.mode = "latest"` 同样会使作业变为有界读取：停止偏移量为快照阶段完成时的 binlog 位置 +（当 `startup.mode` 为 `initial`/`earliest` 时），或作业启动时的 binlog 位置（其他启动模式）。 +特别是与需要执行快照的启动模式组合使用时，**快照运行期间写入源表的变更不会丢失**—— +它们会在作业终止前被 binlog 阶段捕获。 + +```hocon +source { + MySQL-CDC { + server-id = 5654 + username = "st_user_source" + password = "mysqlpw" + table-names = ["mysql_cdc.mysql_cdc_e2e_source_table"] + url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc" + startup.mode = "initial" + stop.mode = "latest" + } +} +``` + ```hocon source { MySQL-CDC { diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/IncrementalSplitAssigner.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/IncrementalSplitAssigner.java index a6188e9183..aa994efd4a 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/IncrementalSplitAssigner.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/IncrementalSplitAssigner.java @@ -20,9 +20,11 @@ package org.apache.seatunnel.connectors.cdc.base.source.enumerator; import org.apache.seatunnel.shade.com.google.common.annotations.VisibleForTesting; import org.apache.seatunnel.api.table.catalog.CatalogTable; +import org.apache.seatunnel.common.utils.SeaTunnelException; import org.apache.seatunnel.connectors.cdc.base.config.SourceConfig; import org.apache.seatunnel.connectors.cdc.base.config.StartupConfig; import org.apache.seatunnel.connectors.cdc.base.option.StartupMode; +import org.apache.seatunnel.connectors.cdc.base.option.StopMode; import org.apache.seatunnel.connectors.cdc.base.source.enumerator.state.IncrementalPhaseState; import org.apache.seatunnel.connectors.cdc.base.source.event.SnapshotSplitWatermark; import org.apache.seatunnel.connectors.cdc.base.source.offset.Offset; @@ -85,6 +87,12 @@ public class IncrementalSplitAssigner<C extends SourceConfig> implements SplitAs */ private Offset startupOffset; + /** + * The stop offset resolved once when the snapshot phase completes ({@code stop.mode = latest}), + * reused after checkpoint restore so that a restart does not re-resolve (and drift) it. + */ + private Offset resolvedStopOffset; + private final boolean restoredFromCheckpoint; public IncrementalSplitAssigner( @@ -113,6 +121,7 @@ public class IncrementalSplitAssigner<C extends SourceConfig> implements SplitAs this.offsetFactory = offsetFactory; this.restoredFromCheckpoint = true; this.startupOffset = checkpointState == null ? null : checkpointState.getStartupOffset(); + this.resolvedStopOffset = checkpointState == null ? null : checkpointState.getStopOffset(); } @Override @@ -190,6 +199,21 @@ public class IncrementalSplitAssigner<C extends SourceConfig> implements SplitAs if (this.startupOffset == null) { this.startupOffset = startupOffset; } + // Mirror the startupOffset restoration for the resolved latest + // stop offset: a checkpoint written before the stopOffset field + // existed has none, so on restore the re-created incremental + // split carries the previously resolved stop offset. Reuse it + // here instead of re-resolving (and drifting) at split creation. + // If several splits with already-diverged stop offsets are handed + // back in the same call, the first one processed wins (they can + // only diverge in the same narrow legacy-checkpoint upgrade case). + if (resolvedStopOffset == null + && context.getSourceConfig() != null + && context.getSourceConfig().getStopConfig().getStopMode() + == StopMode.LATEST + && incrementalSplit.getStopOffset() != null) { + resolvedStopOffset = incrementalSplit.getStopOffset(); + } checkpointTables = incrementalSplit.getCheckpointTables(); historyTableChanges = incrementalSplit.getHistoryTableChanges(); }); @@ -200,7 +224,7 @@ public class IncrementalSplitAssigner<C extends SourceConfig> implements SplitAs @Override public IncrementalPhaseState snapshotState(long checkpointId) { - return new IncrementalPhaseState(startupOffset); + return new IncrementalPhaseState(startupOffset, resolvedStopOffset); } @Override @@ -296,11 +320,32 @@ public class IncrementalSplitAssigner<C extends SourceConfig> implements SplitAs startupOffset = sourceConfig.getStartupConfig().getStartupOffset(offsetFactory); } Offset incrementalSplitStartOffset = minOffset != null ? minOffset : startupOffset; + // stop.mode=latest: resolve the stop offset exactly once here, at the first + // incremental split creation, and reuse the same value for every subsequent split + // and for the checkpoint. This is the single authoritative resolution point (the + // split handed to the reader and the checkpointed value can never diverge, so a + // restore cannot silently move the stop boundary). Other stop modes keep the + // split's configured offset from StopConfig. + if (resolvedStopOffset == null + && sourceConfig.getStopConfig().getStopMode() == StopMode.LATEST) { + // The latest() resolution opens a fresh JDBC connection to query the current + // binlog position; a transient failure at this exact snapshot-to-incremental + // transition point must not fail the whole job, so retry with a short backoff + // (mirrors the connection-factory retry pattern). + resolvedStopOffset = resolveLatestStopOffsetWithRetry(sourceConfig); + LOG.info( + "stop.mode=latest: resolved stop offset {} at incremental split creation", + resolvedStopOffset); + } + Offset incrementalSplitStopOffset = + resolvedStopOffset != null + ? resolvedStopOffset + : sourceConfig.getStopConfig().getStopOffset(offsetFactory); return new IncrementalSplit( String.format(INCREMENTAL_SPLIT_ID, index), capturedTables, incrementalSplitStartOffset, - sourceConfig.getStopConfig().getStopOffset(offsetFactory), + incrementalSplitStopOffset, completedSnapshotSplitInfos, checkpointTables, historyTableChanges); @@ -321,11 +366,50 @@ public class IncrementalSplitAssigner<C extends SourceConfig> implements SplitAs context.getSplitCompletedOffsets().remove(assignedSplit.splitId()); } } - return context.getAssignedSnapshotSplit().isEmpty() - && context.getSplitCompletedOffsets().isEmpty(); + boolean completed = + context.getAssignedSnapshotSplit().isEmpty() + && context.getSplitCompletedOffsets().isEmpty(); + return completed; } public boolean waitingForAssignedSplits() { return !(splitAssigned && noMoreSplits()); } + + /** + * Resolves the latest stop offset with retry, so a transient failure of the underlying JDBC + * query (e.g. a momentarily unreachable database at the snapshot-to-incremental transition) + * does not fail the whole job. Mirrors the retry pattern used by the CDC connection factory. + * + * <p>Threading note: this runs on the job's own split-enumerator coordinator thread (invoked + * via {@code IncrementalSourceEnumerator.assignSplits()} → {@code getNext()}), which is per-job + * and not shared across concurrently running sources; the bounded backoff (at most 300ms + + * 600ms) blocks only this job's enumerator, matching the connection-factory retry. + */ + private Offset resolveLatestStopOffsetWithRetry(C sourceConfig) { + final int maxRetries = 3; + for (int attempt = 1; ; attempt++) { + try { + return sourceConfig.getStopConfig().getStopOffset(offsetFactory); + } catch (RuntimeException e) { + if (attempt < maxRetries) { + try { + Thread.sleep(300L * attempt); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + throw new SeaTunnelException( + "Interrupted while retrying latest stop offset resolution", ie); + } + LOG.warn("Resolving latest stop offset failed, retry times {}", attempt, e); + } else { + LOG.error("Resolving latest stop offset failed after {} attempts", attempt, e); + throw new SeaTunnelException( + "Failed to resolve the latest stop offset after " + + attempt + + " attempts", + e); + } + } + } + } } diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/state/IncrementalPhaseState.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/state/IncrementalPhaseState.java index 80d6ced1fc..d0b78ecb89 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/state/IncrementalPhaseState.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/state/IncrementalPhaseState.java @@ -30,11 +30,24 @@ public class IncrementalPhaseState implements PendingSplitsState { private final Offset startupOffset; + /** + * The stop offset resolved once at the enumerator when the snapshot phase completes ({@code + * stop.mode = latest}). Stored in the checkpoint so that a restart reuses the same value + * instead of re-resolving (and drifting) it. {@code null} for other stop modes and for + * checkpoints written before this field existed. + */ + private final Offset stopOffset; + public IncrementalPhaseState() { - this(null); + this(null, null); } public IncrementalPhaseState(Offset startupOffset) { + this(startupOffset, null); + } + + public IncrementalPhaseState(Offset startupOffset, Offset stopOffset) { this.startupOffset = startupOffset; + this.stopOffset = stopOffset; } } diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/HybridSplitAssignerTest.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/HybridSplitAssignerTest.java index 2c931eb9e4..ebbbfc09fb 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/HybridSplitAssignerTest.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/HybridSplitAssignerTest.java @@ -17,6 +17,9 @@ package org.apache.seatunnel.connectors.cdc.base.source.enumerator; +import org.apache.seatunnel.connectors.cdc.base.config.SourceConfig; +import org.apache.seatunnel.connectors.cdc.base.config.StopConfig; +import org.apache.seatunnel.connectors.cdc.base.option.StopMode; import org.apache.seatunnel.connectors.cdc.base.source.enumerator.state.HybridPendingSplitsState; import org.apache.seatunnel.connectors.cdc.base.source.enumerator.state.SnapshotPhaseState; import org.apache.seatunnel.connectors.cdc.base.source.event.SnapshotSplitWatermark; @@ -34,6 +37,9 @@ import java.util.Map; import java.util.stream.Collectors; import java.util.stream.Stream; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + public class HybridSplitAssignerTest { @Test public void testCompletedSnapshotPhase() { @@ -51,9 +57,16 @@ public class HybridSplitAssignerTest { false); HybridPendingSplitsState checkpointState = new HybridPendingSplitsState(snapshotPhaseState, null); + // This test only exercises split-tracking bookkeeping, so a minimal mocked + // SourceConfig (non-LATEST stop mode) keeps the Context well-formed. The + // stop.mode=latest resolution lives in IncrementalSplitAssigner.createIncrementalSplit + // and is not exercised here. + SourceConfig sourceConfig = mock(SourceConfig.class); + when(sourceConfig.getStopConfig()) + .thenReturn(new StopConfig(StopMode.NEVER, null, null, null)); SplitAssigner.Context context = new SplitAssigner.Context<>( - null, + sourceConfig, Collections.emptySet(), checkpointState.getSnapshotPhaseState().getAssignedSplits(), checkpointState.getSnapshotPhaseState().getSplitCompletedOffsets()); diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/IncrementalSplitAssignerTest.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/IncrementalSplitAssignerTest.java index c96248e498..c02c79bc1d 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/IncrementalSplitAssignerTest.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/IncrementalSplitAssignerTest.java @@ -42,6 +42,8 @@ import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; 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; @@ -152,6 +154,87 @@ class IncrementalSplitAssignerTest { assertSame(startupOffset, reassignedSplit.get().getStartupOffset()); } + @Test + void shouldResolveLatestStopOffsetOnceAtSplitCreationAndReuseAfterRestore() { + SourceConfig sourceConfig = mock(SourceConfig.class); + OffsetFactory offsetFactory = mock(OffsetFactory.class); + Offset committedOffset = mock(Offset.class); + Offset resolvedOffset = mock(Offset.class); + when(sourceConfig.getStartupConfig()) + .thenReturn(new StartupConfig(StartupMode.COMMITTED_OFFSET, null, null, null)); + when(sourceConfig.getStopConfig()) + .thenReturn(new StopConfig(StopMode.LATEST, null, null, null)); + when(offsetFactory.committedOffset()).thenReturn(committedOffset); + when(offsetFactory.latest()).thenReturn(resolvedOffset); + + SplitAssigner.Context<SourceConfig> context = createContext(sourceConfig); + IncrementalSplitAssigner<SourceConfig> assigner = + new IncrementalSplitAssigner<>(context, 1, offsetFactory); + + // The single authoritative resolution happens exactly once, when the first + // incremental split is created; the same value is used for the split handed to + // the reader and stored in the checkpoint, so they can never diverge. + assertSame(resolvedOffset, assigner.getNext().get().asIncrementalSplit().getStopOffset()); + verify(offsetFactory, times(1)).latest(); + + // Snapshot phase completes: no second resolution, the resolved value is reused. + assertTrue(assigner.completedSnapshotPhase(Collections.emptyList())); + verify(offsetFactory, times(1)).latest(); + + // The checkpoint stores the resolved stop offset... + IncrementalPhaseState checkpoint = assigner.snapshotState(1L); + + // ...and a restored assigner reuses it instead of re-resolving (no drift, and no + // boundary change on restore: the checkpointed value is the same one the running + // reader was already stopping at). + IncrementalSplitAssigner<SourceConfig> restoredAssigner = + new IncrementalSplitAssigner<>( + createContext(sourceConfig), 1, offsetFactory, checkpoint); + assertSame( + resolvedOffset, + restoredAssigner.getNext().get().asIncrementalSplit().getStopOffset()); + verify(offsetFactory, times(1)).latest(); + } + + @Test + void shouldRestoreResolvedStopOffsetFromInFlightSplitOnLegacyCheckpointRestore() { + SourceConfig sourceConfig = mock(SourceConfig.class); + OffsetFactory offsetFactory = mock(OffsetFactory.class); + Offset committedOffset = mock(Offset.class); + Offset resolvedOffset = mock(Offset.class); + when(sourceConfig.getStartupConfig()) + .thenReturn(new StartupConfig(StartupMode.COMMITTED_OFFSET, null, null, null)); + when(sourceConfig.getStopConfig()) + .thenReturn(new StopConfig(StopMode.LATEST, null, null, null)); + when(offsetFactory.committedOffset()).thenReturn(committedOffset); + + // A legacy checkpoint (written before the stopOffset field existed) restores the + // assigner with resolvedStopOffset == null. + IncrementalPhaseState legacyCheckpoint = new IncrementalPhaseState(committedOffset); + SplitAssigner.Context<SourceConfig> context = createContext(sourceConfig); + IncrementalSplitAssigner<SourceConfig> restoredAssigner = + new IncrementalSplitAssigner<>(context, 1, offsetFactory, legacyCheckpoint); + + // An already-assigned incremental split is handed back to the enumerator via + // addSplits() (e.g. a reader task failure/retry within the same job run); it still + // carries the previously resolved stop offset. + IncrementalSplit inFlightSplit = + new IncrementalSplit( + "incremental-1", + Collections.singletonList(TableId.parse("database.schema.table")), + committedOffset, + resolvedOffset, + Collections.emptyList()); + restoredAssigner.addSplits(Collections.singletonList(inFlightSplit)); + + // The assigner adopts the in-flight split's stop offset as its resolvedStopOffset, + // so the next split created reuses it instead of re-resolving latest() (and drifting + // the stop boundary after a restore). + IncrementalSplit nextSplit = restoredAssigner.getNext().get().asIncrementalSplit(); + assertSame(resolvedOffset, nextSplit.getStopOffset()); + verify(offsetFactory, never()).latest(); + } + private SplitAssigner.Context<SourceConfig> createContext(SourceConfig sourceConfig) { return createContext( sourceConfig, Collections.singleton(TableId.parse("database.schema.table"))); diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/state/IncrementalPhaseStateTest.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/state/IncrementalPhaseStateTest.java new file mode 100644 index 0000000000..d66850b01f --- /dev/null +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/state/IncrementalPhaseStateTest.java @@ -0,0 +1,91 @@ +/* + * 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.cdc.base.source.enumerator.state; + +import org.apache.seatunnel.connectors.cdc.base.source.offset.Offset; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.util.HashMap; +import java.util.Map; + +class IncrementalPhaseStateTest { + + @Test + void shouldPreserveStopOffsetThroughJavaSerializationRoundTrip() throws Exception { + Offset startupOffset = new TestOffset("startup", 1L); + Offset stopOffset = new TestOffset("stop", 2L); + IncrementalPhaseState state = new IncrementalPhaseState(startupOffset, stopOffset); + + IncrementalPhaseState restored = roundTrip(state); + + // A real checkpoint round-trip (plain Java serialization, the mechanism used for + // enumerator state) must keep the resolved stop offset: otherwise a restore would + // fall back to re-resolving latest() and drift the stop boundary. + Assertions.assertEquals(startupOffset, restored.getStartupOffset()); + Assertions.assertEquals(stopOffset, restored.getStopOffset()); + } + + @Test + void shouldDefaultStopOffsetToNullOnLegacySerializedState() throws Exception { + // Simulates a checkpoint written before the stopOffset field existed (the single-arg + // constructor is what old versions persisted): with the unchanged serialVersionUID, + // Java field-evolution defaults the missing field to null, which falls through to the + // pre-existing StopConfig resolution path. + Offset startupOffset = new TestOffset("startup", 1L); + IncrementalPhaseState legacy = new IncrementalPhaseState(startupOffset); + + IncrementalPhaseState restored = roundTrip(legacy); + + Assertions.assertEquals(startupOffset, restored.getStartupOffset()); + Assertions.assertNull(restored.getStopOffset()); + } + + private static IncrementalPhaseState roundTrip(IncrementalPhaseState state) throws Exception { + ByteArrayOutputStream buffer = new ByteArrayOutputStream(); + try (ObjectOutputStream output = new ObjectOutputStream(buffer)) { + output.writeObject(state); + } + try (ObjectInputStream input = + new ObjectInputStream(new ByteArrayInputStream(buffer.toByteArray()))) { + return (IncrementalPhaseState) input.readObject(); + } + } + + /** A minimal serializable {@link Offset} implementation for round-trip tests. */ + private static class TestOffset extends Offset { + private static final long serialVersionUID = 1L; + + TestOffset(String file, long position) { + Map<String, String> values = new HashMap<>(); + values.put("file", file); + values.put("pos", String.valueOf(position)); + this.offset = values; + } + + @Override + public int compareTo(Offset that) { + return this.offset.get("pos").compareTo(that.getOffset().get("pos")); + } + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCStopModeSpecificIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCStopModeSpecificIT.java index 4877f46d99..d464a45029 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCStopModeSpecificIT.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCStopModeSpecificIT.java @@ -309,6 +309,151 @@ public class MysqlCDCStopModeSpecificIT extends TestSuiteBase implements TestRes }); } + @TestTemplate + public void testMysqlCdcInitialStartupWithLatestStop(TestContainer container) throws Exception { + runLatestStopStartupMode(container, "initial"); + } + + @TestTemplate + public void testMysqlCdcEarliestStartupWithLatestStop(TestContainer container) + throws Exception { + runLatestStopStartupMode(container, "earliest"); + } + + @TestTemplate + public void testMysqlCdcLatestStartupWithLatestStop(TestContainer container) throws Exception { + runLatestStopStartupMode(container, "latest"); + } + + @TestTemplate + public void testMysqlCdcSpecificStartupWithLatestStop(TestContainer container) + throws Exception { + runLatestStopStartupMode(container, "specific"); + } + + @TestTemplate + public void testMysqlCdcTimestampStartupWithLatestStop(TestContainer container) + throws Exception { + runLatestStopStartupMode(container, "timestamp"); + } + + private void runLatestStopStartupMode(TestContainer container, String startupMode) + throws Exception { + clearTable(MYSQL_DATABASE, SOURCE_TABLE); + clearTable(MYSQL_DATABASE, SINK_TABLE); + + // Bulk-insert rows before starting the job. + // - initial: 2000 rows so the snapshot phase spans many splits deterministically + // (snapshot.split.size=20 -> 100 splits, parallelism=1, consumed in ascending-key + // order): the readiness gate below (first row visible in the sink) then fires while + // ~99 splits are still unread, so the UPDATE is guaranteed to land inside the + // snapshot window. A smaller table could be snapshotted entirely within one polling + // interval and complete before the gate is observed (the flaky race SEZ9 flagged). + // - earliest: keep 200 rows — there is no snapshot phase (binlog replay from the + // earliest position), and a larger bulk would only slow the replay. + // - latest/specific/timestamp: 200 rows, no snapshot window involved. + int bulkRowCount = "initial".equals(startupMode) ? 2000 : 200; + StringBuilder bulkInsert = + new StringBuilder( + String.format( + "INSERT INTO %s.%s (id, f_varchar) VALUES ", + MYSQL_DATABASE, SOURCE_TABLE)); + for (int i = 1; i <= bulkRowCount; i++) { + if (i > 1) { + bulkInsert.append(", "); + } + bulkInsert.append("(").append(i).append(", 'bulk')"); + } + executeSql(bulkInsert.toString()); + + List<String> variables = new ArrayList<>(); + variables.add("startup_mode=" + startupMode); + String jobConfigFile = "/mysqlcdc_stop_mode_latest.conf"; + if ("specific".equals(startupMode)) { + BinlogOffset startOffset = getCurrentBinlogOffset(); + jobConfigFile = "/mysqlcdc_stop_mode_latest_specific.conf"; + variables.add("specific_offset_file=" + startOffset.getFilename()); + variables.add("specific_offset_pos=" + startOffset.getPosition()); + } + if ("timestamp".equals(startupMode)) { + jobConfigFile = "/mysqlcdc_stop_mode_latest_timestamp.conf"; + variables.add("timestamp=" + (getCurrentBinlogTimestamp() - 1000L)); + } + final String configFile = jobConfigFile; + + String jobId = String.valueOf(JobIdGenerator.newJobId()); + CompletableFuture<Container.ExecResult> jobFuture = + CompletableFuture.supplyAsync( + () -> { + try { + return container.executeJob( + configFile, jobId, variables.toArray(new String[0])); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + + await().atMost(60, TimeUnit.SECONDS) + .untilAsserted( + () -> Assertions.assertEquals("RUNNING", container.getJobStatus(jobId))); + + if ("initial".equals(startupMode)) { + // Structural readiness gate for snapshot-taking startups: wait until the first + // bulk row (id=1) has reached the sink. With 2000 rows across snapshot.split.size=20 + // (100 splits, parallelism=1, ascending-key order), this signal fires while ~99 + // splits are still unread — the UPDATE below is therefore guaranteed to land + // inside the snapshot window and be picked up by the binlog phase. + await().atMost(60, TimeUnit.SECONDS) + .untilAsserted( + () -> + Assertions.assertEquals( + "bulk", + queryVarcharById(1), + "snapshot phase must have started reading")); + // Defensive check for the residual race SEZ9 flagged: if the LAST-chunk row is + // already visible, the whole snapshot completed before the gate fired and the + // UPDATE below would land after the stop offset. Fail loudly instead of letting + // the post-UPDATE assertion mislead as data loss. (Row-lock stalls are not usable + // here: the snapshot reader issues a plain MVCC SELECT, so an open FOR UPDATE + // transaction on a trailing row does not block it.) + Assertions.assertNull( + queryVarcharById(2000), + "snapshot finished too early: last-chunk row already in sink; " + + "increase the initial bulk row count"); + } + + // Issue a change while the job is running. + // - initial: the sink-readiness gate above guarantees the snapshot phase is still in + // progress, so the update must be picked up by the binlog phase; with the stale + // split-creation stop offset the binlog phase would terminate immediately past it + // and this row would be silently dropped. + // - earliest: there is no snapshot phase (binlog replay starts from the earliest + // position, completedSnapshotSplitInfos is empty), so whether the update is + // captured depends on the enumerator's stop-offset resolution timing versus the + // update — not a deterministic contract, therefore not asserted. + // - latest/specific/timestamp: no snapshot window, the update lands after the stop + // offset is resolved and is not read — which is expected. + executeSql( + String.format( + "UPDATE %s.%s SET f_varchar = 'latest-stop' WHERE id = 21", + MYSQL_DATABASE, SOURCE_TABLE)); + + if ("initial".equals(startupMode)) { + await().atMost(60, TimeUnit.SECONDS) + .untilAsserted( + () -> + Assertions.assertEquals( + "latest-stop", + queryVarcharById(21), + "post-snapshot update must be synced")); + } + + await().atMost(120, TimeUnit.SECONDS) + .untilAsserted( + () -> Assertions.assertEquals("FINISHED", container.getJobStatus(jobId))); + Assertions.assertEquals(0, jobFuture.get(30, TimeUnit.SECONDS).getExitCode()); + } + private long getCurrentBinlogTimestamp() { BinlogOffset binlogOffset = getCurrentBinlogOffset(); @@ -411,6 +556,23 @@ public class MysqlCDCStopModeSpecificIT extends TestSuiteBase implements TestRes } } + private String queryVarcharById(int id) { + try (Connection connection = getJdbcConnection(); + Statement statement = connection.createStatement(); + ResultSet resultSet = + statement.executeQuery( + String.format( + "select f_varchar from %s.%s where id = %d", + MYSQL_DATABASE, SINK_TABLE, id))) { + if (resultSet.next()) { + return resultSet.getString(1); + } + return null; + } catch (SQLException e) { + throw new RuntimeException(e); + } + } + private void clearTable(String database, String tableName) { try (Connection connection = getJdbcConnection()) { connection diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_stop_mode_latest.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_stop_mode_latest.conf new file mode 100644 index 0000000000..e3fe04e49d --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_stop_mode_latest.conf @@ -0,0 +1,56 @@ +# +# 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. +# + +env { + parallelism = 1 + job.mode = "STREAMING" + checkpoint.interval = 5000 +} + +source { + MySQL-CDC { + plugin_output = "customers_mysql_cdc" + server-id = 5654 + username = "st_user_source" + password = "mysqlpw" + table-names = ["mysql_cdc.mysql_cdc_e2e_source_table"] + url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc" + startup.mode = ${startup_mode} + stop.mode = "latest" + # For snapshot-taking startups (initial), keep the snapshot phase multi-split so the + # sink-readiness signal in runLatestStopStartupMode (first bulk row visible) can only + # fire while the snapshot is still reading the remaining splits — guaranteeing the + # post-readiness UPDATE lands inside the snapshot window. With the default 8096-row + # split size the 200-row table is read in a single split and the snapshot can + # complete before the signal is observed. + snapshot.split.size = 20 + } +} + +sink { + jdbc { + plugin_input = "customers_mysql_cdc" + url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc" + driver = "com.mysql.cj.jdbc.Driver" + user = "st_user_sink" + password = "mysqlpw" + generate_sink_sql = true + database = mysql_cdc + table = mysql_cdc_e2e_sink_table + primary_keys = ["id"] + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_stop_mode_latest_specific.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_stop_mode_latest_specific.conf new file mode 100644 index 0000000000..e7de2cfb5c --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_stop_mode_latest_specific.conf @@ -0,0 +1,51 @@ +# +# 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. +# + +env { + parallelism = 1 + job.mode = "STREAMING" + checkpoint.interval = 5000 +} + +source { + MySQL-CDC { + plugin_output = "customers_mysql_cdc" + server-id = 5654 + username = "st_user_source" + password = "mysqlpw" + table-names = ["mysql_cdc.mysql_cdc_e2e_source_table"] + url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc" + startup.mode = "specific" + startup.specific-offset.file = ${specific_offset_file} + startup.specific-offset.pos = ${specific_offset_pos} + stop.mode = "latest" + } +} + +sink { + jdbc { + plugin_input = "customers_mysql_cdc" + url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc" + driver = "com.mysql.cj.jdbc.Driver" + user = "st_user_sink" + password = "mysqlpw" + generate_sink_sql = true + database = mysql_cdc + table = mysql_cdc_e2e_sink_table + primary_keys = ["id"] + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_stop_mode_latest_timestamp.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_stop_mode_latest_timestamp.conf new file mode 100644 index 0000000000..50108a3196 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_stop_mode_latest_timestamp.conf @@ -0,0 +1,50 @@ +# +# 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. +# + +env { + parallelism = 1 + job.mode = "STREAMING" + checkpoint.interval = 5000 +} + +source { + MySQL-CDC { + plugin_output = "customers_mysql_cdc" + server-id = 5655 + username = "st_user_source" + password = "mysqlpw" + table-names = ["mysql_cdc.mysql_cdc_e2e_source_table"] + url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc" + startup.mode = "timestamp" + startup.timestamp = ${timestamp} + stop.mode = "latest" + } +} + +sink { + jdbc { + plugin_input = "customers_mysql_cdc" + url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc" + driver = "com.mysql.cj.jdbc.Driver" + user = "st_user_sink" + password = "mysqlpw" + generate_sink_sql = true + database = mysql_cdc + table = mysql_cdc_e2e_sink_table + primary_keys = ["id"] + } +}
