This is an automated email from the ASF dual-hosted git repository.
nzw921rx pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 2aa67dbd37 [Fix][CDC] Keep per-table restored offsets when tables are
added across a savepoint (#13006)
2aa67dbd37 is described below
commit 2aa67dbd370d1591255da9a51d8ef70f01373010
Author: Goutam Adwant <[email protected]>
AuthorDate: Tue Oct 27 22:08:26 2026 -0700
[Fix][CDC] Keep per-table restored offsets when tables are added across a
savepoint (#13006)
---
.../enumerator/IncrementalSplitAssigner.java | 37 ++++-
.../external/IncrementalSourceStreamFetcher.java | 36 +++-
.../cdc/base/source/split/IncrementalSplit.java | 52 +++++-
.../source/split/state/IncrementalSplitState.java | 30 +++-
.../enumerator/IncrementalSplitAssignerTest.java | 99 +++++++++++
.../IncrementalSourceStreamFetcherTest.java | 181 +++++++++++++++++++++
.../split/state/IncrementalSplitStateTest.java | 33 ++++
.../cdc/mysql/AbstractMysqlCDCITBase.java | 113 ++++++++++++-
..._mysql_with_exactly_once_restore_add_table.conf | 68 ++++++++
...mysql_with_exactly_once_restore_two_tables.conf | 62 +++++++
10 files changed, 699 insertions(+), 12 deletions(-)
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 aa994efd4a..5755cd2631 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
@@ -72,6 +72,14 @@ public class IncrementalSplitAssigner<C extends
SourceConfig> implements SplitAs
*/
private final Map<TableId, Offset> tableWatermarks = new HashMap<>();
+ /**
+ * Offset each restored table had already been read up to. It is the
startup offset of the split
+ * the table was restored from, or a later offset that split still carried
for the table. Unlike
+ * {@link #tableWatermarks} it does not decide where an incremental split
starts; it only tells
+ * the reader which records of the table were already emitted. Used to
support Exactly-Once.
+ */
+ private final Map<TableId, Offset> tableStartupOffsets = new HashMap<>();
+
private boolean splitAssigned = false;
private final List<IncrementalSplit> remainingSplits = new ArrayList<>();
@@ -195,6 +203,13 @@ public class IncrementalSplitAssigner<C extends
SourceConfig> implements SplitAs
continue;
}
tableWatermarks.put(tableId, startupOffset);
+ tableStartupOffsets.put(
+ tableId,
+ laterOf(
+ startupOffset,
+ incrementalSplit
+
.getTableStartupOffsets()
+ .get(tableId)));
}
if (this.startupOffset == null) {
this.startupOffset = startupOffset;
@@ -320,6 +335,18 @@ public class IncrementalSplitAssigner<C extends
SourceConfig> implements SplitAs
startupOffset =
sourceConfig.getStartupConfig().getStartupOffset(offsetFactory);
}
Offset incrementalSplitStartOffset = minOffset != null ? minOffset :
startupOffset;
+ // Restored tables from different splits can be regrouped into this
split, which starts at
+ // the earliest of their offsets. Keep the offset of every table that
was already read
+ // further, so the reader does not emit those records a second time.
+ Map<TableId, Offset> splitTableStartupOffsets = new HashMap<>();
+ for (TableId tableId : capturedTables) {
+ Offset tableStartupOffset = tableStartupOffsets.get(tableId);
+ if (tableStartupOffset != null
+ && incrementalSplitStartOffset != null
+ &&
tableStartupOffset.isAfter(incrementalSplitStartOffset)) {
+ splitTableStartupOffsets.put(tableId, tableStartupOffset);
+ }
+ }
// 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
@@ -348,7 +375,15 @@ public class IncrementalSplitAssigner<C extends
SourceConfig> implements SplitAs
incrementalSplitStopOffset,
completedSnapshotSplitInfos,
checkpointTables,
- historyTableChanges);
+ historyTableChanges,
+ splitTableStartupOffsets);
+ }
+
+ private static Offset laterOf(Offset offset, Offset other) {
+ if (offset == null) {
+ return other;
+ }
+ return other != null && other.isAfter(offset) ? other : offset;
}
@VisibleForTesting
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/reader/external/IncrementalSourceStreamFetcher.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/reader/external/IncrementalSourceStreamFetcher.java
index 03df95855e..54695e56e5 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/reader/external/IncrementalSourceStreamFetcher.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/reader/external/IncrementalSourceStreamFetcher.java
@@ -85,6 +85,8 @@ public class IncrementalSourceStreamFetcher implements
Fetcher<SourceRecords, So
private Map<TableId, Offset> maxSplitHighWatermarkMap;
// finished spilt info
private Map<TableId, List<CompletedSnapshotSplitInfo>> finishedSplitsInfo;
+ // offsets restored tables had already been read up to
+ private Map<TableId, Offset> tableStartupOffsets;
private static final long READER_CLOSE_TIMEOUT_SECONDS = 30L;
@@ -282,6 +284,10 @@ public class IncrementalSourceStreamFetcher implements
Fetcher<SourceRecords, So
}
// not enter pure binlog mode and need to check whether the
current record meets the
// emitting conditions.
+ Offset tableStartupOffset = tableStartupOffsets.get(tableId);
+ if (tableStartupOffset != null &&
position.isBefore(tableStartupOffset)) {
+ return false;
+ }
if (finishedSplitsInfo.containsKey(tableId)) {
for (CompletedSnapshotSplitInfo splitInfo :
finishedSplitsInfo.get(tableId)) {
if (taskContext.isRecordBetween(
@@ -316,16 +322,11 @@ public class IncrementalSourceStreamFetcher implements
Fetcher<SourceRecords, So
splitStartWatermark = currentIncrementalSplit.getStartupOffset();
Map<TableId, List<CompletedSnapshotSplitInfo>> splitsInfoMap = new
HashMap<>();
Map<TableId, Offset> tableIdBinlogPositionMap = new HashMap<>();
+ Map<TableId, Offset> restoredTableOffsets =
+ currentIncrementalSplit.getTableStartupOffsets();
List<CompletedSnapshotSplitInfo> completedSnapshotSplitInfos =
currentIncrementalSplit.getCompletedSnapshotSplitInfos();
- // latest-offset mode
- if (completedSnapshotSplitInfos.isEmpty()) {
- for (TableId tableId : currentIncrementalSplit.getTableIds()) {
- tableIdBinlogPositionMap.put(tableId,
currentIncrementalSplit.getStartupOffset());
- }
- }
-
// calculate the max high watermark of every table
for (CompletedSnapshotSplitInfo finishedSplitInfo :
completedSnapshotSplitInfos) {
TableId tableId = finishedSplitInfo.getTableId();
@@ -340,8 +341,29 @@ public class IncrementalSourceStreamFetcher implements
Fetcher<SourceRecords, So
tableIdBinlogPositionMap.put(tableId, highWatermark);
}
}
+
+ for (TableId tableId : currentIncrementalSplit.getTableIds()) {
+ Offset restoredOffset = restoredTableOffsets.get(tableId);
+ Offset maxHighWatermark = tableIdBinlogPositionMap.get(tableId);
+ if (maxHighWatermark == null) {
+ // The table has no snapshot split in this split: either the
split starts from a
+ // log offset (latest-offset mode), or the table was already
in the pure
+ // incremental phase before a restore that added other tables.
Emit its changes
+ // from the offset it was already read up to. As in
latest-offset mode, a record
+ // at exactly that offset is emitted: some dialects (e.g.
Oracle SCN) share one
+ // offset between several records, so skipping it could lose
data.
+ Offset tableStartupOffset =
+ restoredOffset != null ? restoredOffset :
splitStartWatermark;
+ if (tableStartupOffset != null) {
+ tableIdBinlogPositionMap.put(tableId, tableStartupOffset);
+ }
+ } else if (restoredOffset != null &&
restoredOffset.isAfter(maxHighWatermark)) {
+ tableIdBinlogPositionMap.put(tableId, restoredOffset);
+ }
+ }
this.finishedSplitsInfo = splitsInfoMap;
this.maxSplitHighWatermarkMap = tableIdBinlogPositionMap;
+ this.tableStartupOffsets = restoredTableOffsets;
this.pureBinlogPhaseTables.clear();
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/split/IncrementalSplit.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/split/IncrementalSplit.java
index 39cb117c68..9ea1af5c11 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/split/IncrementalSplit.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/split/IncrementalSplit.java
@@ -28,6 +28,7 @@ import lombok.ToString;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -63,6 +64,16 @@ public class IncrementalSplit extends SourceSplitBase {
// debezium history table changes
private final Map<TableId, byte[]> historyTableChanges;
+ /**
+ * Offsets that tables restored from a checkpoint had already been read up
to, kept only for
+ * tables whose offset is after {@link #startupOffset}. This happens when
tables restored from
+ * different incremental splits are regrouped into one split, which then
starts at the earliest
+ * of their offsets. Used to support Exactly-Once.
+ *
+ * <p>Null for splits restored from checkpoints written before this field
existed.
+ */
+ private final Map<TableId, Offset> tableStartupOffsets;
+
public IncrementalSplit(
String splitId,
List<TableId> capturedTables,
@@ -101,7 +112,8 @@ public class IncrementalSplit extends SourceSplitBase {
split.getStopOffset(),
split.getCompletedSnapshotSplitInfos(),
tables,
- historyTableChanges);
+ historyTableChanges,
+ split.getTableStartupOffsets());
}
@Deprecated
@@ -119,6 +131,7 @@ public class IncrementalSplit extends SourceSplitBase {
this.completedSnapshotSplitInfos = completedSnapshotSplitInfos;
this.checkpointDataType = checkpointDataType;
this.historyTableChanges = new HashMap<>();
+ this.tableStartupOffsets = new HashMap<>();
}
public IncrementalSplit(
@@ -129,6 +142,26 @@ public class IncrementalSplit extends SourceSplitBase {
List<CompletedSnapshotSplitInfo> completedSnapshotSplitInfos,
List<CatalogTable> checkpointTables,
Map<TableId, byte[]> historyTableChanges) {
+ this(
+ splitId,
+ capturedTables,
+ startupOffset,
+ stopOffset,
+ completedSnapshotSplitInfos,
+ checkpointTables,
+ historyTableChanges,
+ new HashMap<>());
+ }
+
+ public IncrementalSplit(
+ String splitId,
+ List<TableId> capturedTables,
+ Offset startupOffset,
+ Offset stopOffset,
+ List<CompletedSnapshotSplitInfo> completedSnapshotSplitInfos,
+ List<CatalogTable> checkpointTables,
+ Map<TableId, byte[]> historyTableChanges,
+ Map<TableId, Offset> tableStartupOffsets) {
super(splitId);
this.tableIds = capturedTables;
this.startupOffset = startupOffset;
@@ -136,6 +169,16 @@ public class IncrementalSplit extends SourceSplitBase {
this.completedSnapshotSplitInfos = completedSnapshotSplitInfos;
this.checkpointTables = checkpointTables;
this.historyTableChanges = historyTableChanges;
+ this.tableStartupOffsets = tableStartupOffsets;
+ }
+
+ /**
+ * Returns the offsets restored tables had already been read up to, keyed
by table.
+ *
+ * @return the per-table offsets, empty for splits restored from older
checkpoints
+ */
+ public Map<TableId, Offset> getTableStartupOffsets() {
+ return tableStartupOffsets == null ? Collections.emptyMap() :
tableStartupOffsets;
}
/**
@@ -184,6 +227,10 @@ public class IncrementalSplit extends SourceSplitBase {
: historyTableChanges.entrySet().stream()
.filter(entry ->
capturedTableSet.contains(entry.getKey()))
.collect(Collectors.toMap(Map.Entry::getKey,
Map.Entry::getValue));
+ Map<TableId, Offset> filteredTableStartupOffsets =
+ getTableStartupOffsets().entrySet().stream()
+ .filter(entry ->
capturedTableSet.contains(entry.getKey()))
+ .collect(Collectors.toMap(Map.Entry::getKey,
Map.Entry::getValue));
IncrementalSplit prunedSplit =
new IncrementalSplit(
splitId(),
@@ -192,7 +239,8 @@ public class IncrementalSplit extends SourceSplitBase {
stopOffset,
filteredCompletedSnapshotSplitInfos,
filteredCheckpointTables,
- filteredHistoryTableChanges);
+ filteredHistoryTableChanges,
+ filteredTableStartupOffsets);
// Keep compatibility with checkpoints created before table-level
schema history.
prunedSplit.checkpointDataType = checkpointDataType;
return prunedSplit;
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/split/state/IncrementalSplitState.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/split/state/IncrementalSplitState.java
index c04026bf1e..e35d6ca200 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/split/state/IncrementalSplitState.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/split/state/IncrementalSplitState.java
@@ -24,8 +24,11 @@ import io.debezium.relational.TableId;
import lombok.Getter;
import lombok.Setter;
+import java.util.ArrayList;
import java.util.Comparator;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
/** The state of split to describe the change log of table(s). */
@Getter
@@ -43,11 +46,15 @@ public class IncrementalSplitState extends
SourceSplitStateBase {
private Offset maxSnapshotSplitsHighWatermark;
private volatile boolean enterPureIncrementPhase;
+ /** Restored per-table offsets, see {@link
IncrementalSplit#getTableStartupOffsets()}. */
+ private Map<TableId, Offset> tableStartupOffsets;
+
public IncrementalSplitState(IncrementalSplit split) {
super(split);
this.tableIds = split.getTableIds();
this.startupOffset = split.getStartupOffset();
this.stopOffset = split.getStopOffset();
+ this.tableStartupOffsets = split.getTableStartupOffsets();
if (split.getCompletedSnapshotSplitInfos().isEmpty()) {
this.maxSnapshotSplitsHighWatermark = null;
@@ -71,7 +78,28 @@ public class IncrementalSplitState extends
SourceSplitStateBase {
getTableIds(),
getStartupOffset(),
getStopOffset(),
- incrementalSplit.getCompletedSnapshotSplitInfos());
+ incrementalSplit.getCompletedSnapshotSplitInfos(),
+ new ArrayList<>(),
+ new HashMap<>(),
+ pendingTableStartupOffsets());
+ }
+
+ /**
+ * A restored table offset only needs to be checkpointed until the split
has read past it, so
+ * entries at or before the current startup offset are dropped.
+ */
+ private Map<TableId, Offset> pendingTableStartupOffsets() {
+ Map<TableId, Offset> pending = new HashMap<>();
+ if (tableStartupOffsets == null) {
+ return pending;
+ }
+ tableStartupOffsets.forEach(
+ (tableId, offset) -> {
+ if (startupOffset == null ||
offset.isAfter(startupOffset)) {
+ pending.put(tableId, offset);
+ }
+ });
+ return pending;
}
public synchronized boolean markEnterPureIncrementPhaseIfNeed(Offset
currentRecordPosition) {
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 c02c79bc1d..67591fb024 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
@@ -31,8 +31,10 @@ import org.junit.jupiter.api.Test;
import io.debezium.relational.TableId;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.Optional;
import java.util.Set;
@@ -235,6 +237,87 @@ class IncrementalSplitAssignerTest {
verify(offsetFactory, never()).latest();
}
+ @Test
+ void
shouldStartRegroupedSplitAtEarliestRestoredOffsetAndKeepLaterTableOffsets() {
+ SourceConfig sourceConfig = mock(SourceConfig.class);
+ when(sourceConfig.getStartupConfig())
+ .thenReturn(new StartupConfig(StartupMode.INITIAL, null, null,
null));
+ when(sourceConfig.getStopConfig())
+ .thenReturn(new StopConfig(StopMode.NEVER, null, null, null));
+ TableId table1 = TableId.parse("database.schema.table1");
+ TableId table2 = TableId.parse("database.schema.table2");
+ Offset stoppingOffset = mock(Offset.class);
+
+ // incremental.parallelism was 2: each table was read by its own
split, and the split of
+ // table2 had read further than the split of table1 when the
checkpoint was taken.
+ IncrementalSplit restoredSplit1 =
+ new IncrementalSplit(
+ "incremental-split-0",
+ Collections.singletonList(table1),
+ new TestOffset(100),
+ stoppingOffset,
+ Collections.emptyList());
+ IncrementalSplit restoredSplit2 =
+ new IncrementalSplit(
+ "incremental-split-1",
+ Collections.singletonList(table2),
+ new TestOffset(200),
+ stoppingOffset,
+ Collections.emptyList());
+ IncrementalSplitAssigner<SourceConfig> assigner =
+ new IncrementalSplitAssigner<>(
+ createContext(sourceConfig, new
HashSet<>(Arrays.asList(table1, table2))),
+ 1,
+ mock(OffsetFactory.class));
+ assigner.addSplits(Arrays.asList(restoredSplit1, restoredSplit2));
+
+ IncrementalSplit split = assigner.getNext().get().asIncrementalSplit();
+
+ assertEquals(
+ new HashSet<>(Arrays.asList(table1, table2)), new
HashSet<>(split.getTableIds()));
+ assertEquals(new TestOffset(100), split.getStartupOffset());
+ assertEquals(
+ Collections.singletonMap(table2, new TestOffset(200)),
+ split.getTableStartupOffsets());
+ }
+
+ @Test
+ void shouldKeepTableOffsetCarriedByRestoredSplit() {
+ SourceConfig sourceConfig = mock(SourceConfig.class);
+ when(sourceConfig.getStartupConfig())
+ .thenReturn(new StartupConfig(StartupMode.INITIAL, null, null,
null));
+ when(sourceConfig.getStopConfig())
+ .thenReturn(new StopConfig(StopMode.NEVER, null, null, null));
+ TableId table1 = TableId.parse("database.schema.table1");
+ TableId table2 = TableId.parse("database.schema.table2");
+
+ // The split was checkpointed at 150 while it was still replaying up
to table2's
+ // restored offset 200, so it carries that offset for table2.
+ IncrementalSplit restoredSplit =
+ new IncrementalSplit(
+ "incremental-split-0",
+ Arrays.asList(table1, table2),
+ new TestOffset(150),
+ mock(Offset.class),
+ Collections.emptyList(),
+ null,
+ Collections.emptyMap(),
+ Collections.singletonMap(table2, new TestOffset(200)));
+ IncrementalSplitAssigner<SourceConfig> assigner =
+ new IncrementalSplitAssigner<>(
+ createContext(sourceConfig, new
HashSet<>(Arrays.asList(table1, table2))),
+ 1,
+ mock(OffsetFactory.class));
+ assigner.addSplits(Collections.singletonList(restoredSplit));
+
+ IncrementalSplit split = assigner.getNext().get().asIncrementalSplit();
+
+ assertEquals(new TestOffset(150), split.getStartupOffset());
+ assertEquals(
+ Collections.singletonMap(table2, new TestOffset(200)),
+ split.getTableStartupOffsets());
+ }
+
private SplitAssigner.Context<SourceConfig> createContext(SourceConfig
sourceConfig) {
return createContext(
sourceConfig,
Collections.singleton(TableId.parse("database.schema.table")));
@@ -248,4 +331,20 @@ class IncrementalSplitAssignerTest {
new HashMap<>(),
new HashMap<>());
}
+
+ private static class TestOffset extends Offset {
+ private static final long serialVersionUID = 1L;
+
+ private final long position;
+
+ TestOffset(long position) {
+ this.position = position;
+ this.offset = Collections.singletonMap("pos",
String.valueOf(position));
+ }
+
+ @Override
+ public int compareTo(Offset o) {
+ return Long.compare(position, ((TestOffset) o).position);
+ }
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/reader/external/IncrementalSourceStreamFetcherTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/reader/external/IncrementalSourceStreamFetcherTest.java
index ab3f93883e..4cf3478cbf 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/reader/external/IncrementalSourceStreamFetcherTest.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/reader/external/IncrementalSourceStreamFetcherTest.java
@@ -18,7 +18,9 @@
package org.apache.seatunnel.connectors.cdc.base.source.reader.external;
import org.apache.seatunnel.connectors.cdc.base.schema.SchemaChangeResolver;
+import
org.apache.seatunnel.connectors.cdc.base.source.event.SnapshotSplitWatermark;
import org.apache.seatunnel.connectors.cdc.base.source.offset.Offset;
+import
org.apache.seatunnel.connectors.cdc.base.source.split.CompletedSnapshotSplitInfo;
import org.apache.seatunnel.connectors.cdc.base.source.split.IncrementalSplit;
import org.apache.seatunnel.connectors.cdc.base.source.split.SourceRecords;
import org.apache.seatunnel.connectors.cdc.base.source.split.SourceSplitBase;
@@ -598,6 +600,185 @@ public class IncrementalSourceStreamFetcherTest {
Assertions.assertFalse(fetcher.shouldEmit(record));
}
+ @Test
+ public void
testShouldEmitRestoredTableChangesWhileAddedTableIsSnapshotting() {
+ // Restored after a savepoint that captured old_table, with new_table
added to the
+ // configuration: the re-created split only carries snapshot info for
new_table.
+ TableId oldTable = new TableId("testdb", "public", "old_table");
+ TableId newTable = new TableId("testdb", "public", "new_table");
+ IncrementalSplit split =
+ new IncrementalSplit(
+ "incremental-split-0",
+ Arrays.asList(oldTable, newTable),
+ new TestOffset(100),
+ null,
+
Collections.singletonList(completedSnapshotSplitInfo(newTable, 140, 150)));
+
+ IncrementalSourceStreamFetcher fetcher = submitExactlyOnceTask(split);
+ try {
+ Assertions.assertTrue(
+ fetcher.shouldEmit(createDataEventWithSource(oldTable,
120)),
+ "change of the restored table between its checkpoint
offset and the added table's high watermark was dropped");
+
Assertions.assertTrue(fetcher.shouldEmit(createDataEventWithSource(oldTable,
160)));
+
Assertions.assertFalse(fetcher.shouldEmit(createDataEventWithSource(newTable,
120)));
+
Assertions.assertTrue(fetcher.shouldEmit(createDataEventWithSource(newTable,
160)));
+ } finally {
+ fetcher.close();
+ }
+ }
+
+ @Test
+ public void testShouldEmitRegroupedRestoredTablesFromTheirOwnOffsets() {
+ // With incremental.parallelism > 1, table1 and table2 were read by
different splits and
+ // checkpointed at 100 and 200. After the restore they share one split
that starts at 100.
+ TableId table1 = new TableId("testdb", "public", "table1");
+ TableId table2 = new TableId("testdb", "public", "table2");
+ TableId newTable = new TableId("testdb", "public", "new_table");
+ IncrementalSplit split =
+ new IncrementalSplit(
+ "incremental-split-0",
+ Arrays.asList(table1, table2, newTable),
+ new TestOffset(100),
+ null,
+
Collections.singletonList(completedSnapshotSplitInfo(newTable, 140, 150)),
+ null,
+ Collections.emptyMap(),
+ Collections.singletonMap(table2, new TestOffset(200)));
+
+ IncrementalSourceStreamFetcher fetcher =
submitTask(exactlyOnceTaskContext(), split);
+ try {
+
Assertions.assertTrue(fetcher.shouldEmit(createDataEventWithSource(table1,
120)));
+ Assertions.assertFalse(
+ fetcher.shouldEmit(createDataEventWithSource(table2, 150)),
+ "change of table2 already emitted before the checkpoint
was emitted again");
+
Assertions.assertFalse(fetcher.shouldEmit(createDataEventWithSource(table2,
199)));
+
Assertions.assertTrue(fetcher.shouldEmit(createDataEventWithSource(table2,
200)));
+
Assertions.assertTrue(fetcher.shouldEmit(createDataEventWithSource(table2,
250)));
+
Assertions.assertFalse(fetcher.shouldEmit(createDataEventWithSource(newTable,
120)));
+
Assertions.assertTrue(fetcher.shouldEmit(createDataEventWithSource(newTable,
160)));
+ } finally {
+ fetcher.close();
+ }
+ }
+
+ @Test
+ public void
testShouldNotEmitSnapshotRangeChangesBeforeRestoredTableOffset() {
+ // The table was checkpointed at 200 before its split entered the pure
incremental phase,
+ // and is regrouped into a split that starts earlier.
+ TableId table = new TableId("testdb", "public", "table1");
+ IncrementalSplit split =
+ new IncrementalSplit(
+ "incremental-split-0",
+ Collections.singletonList(table),
+ new TestOffset(100),
+ null,
+
Collections.singletonList(completedSnapshotSplitInfo(table, 140, 150)),
+ null,
+ Collections.emptyMap(),
+ Collections.singletonMap(table, new TestOffset(200)));
+ FetchTask.Context taskContext = exactlyOnceTaskContext();
+ when(taskContext.isRecordBetween(any(), any(),
any())).thenReturn(true);
+
+ IncrementalSourceStreamFetcher fetcher = submitTask(taskContext,
split);
+ try {
+
Assertions.assertFalse(fetcher.shouldEmit(createDataEventWithSource(table,
170)));
+
Assertions.assertTrue(fetcher.shouldEmit(createDataEventWithSource(table,
200)));
+ } finally {
+ fetcher.close();
+ }
+ }
+
+ private static IncrementalSourceStreamFetcher
submitExactlyOnceTask(IncrementalSplit split) {
+ return submitTask(exactlyOnceTaskContext(), split);
+ }
+
+ private static FetchTask.Context exactlyOnceTaskContext() {
+ FetchTask.Context taskContext = mock(FetchTask.Context.class);
+ when(taskContext.isExactlyOnce()).thenReturn(true);
+ when(taskContext.isDataChangeRecord(any())).thenReturn(true);
+ when(taskContext.getStreamOffset(any()))
+ .thenAnswer(
+ invocation -> {
+ SourceRecord record = invocation.getArgument(0);
+ return new TestOffset(
+ Long.parseLong((String)
record.sourceOffset().get("pos")));
+ });
+ return taskContext;
+ }
+
+ private static IncrementalSourceStreamFetcher submitTask(
+ FetchTask.Context taskContext, IncrementalSplit split) {
+ FetchTask<SourceSplitBase> fetchTask = mock(FetchTask.class);
+ when(fetchTask.getSplit()).thenReturn(split);
+
+ IncrementalSourceStreamFetcher fetcher =
+ new IncrementalSourceStreamFetcher(taskContext, 0, null);
+ fetcher.submitTask(fetchTask);
+ return fetcher;
+ }
+
+ private static CompletedSnapshotSplitInfo completedSnapshotSplitInfo(
+ TableId tableId, long lowWatermark, long highWatermark) {
+ String splitId = tableId + ":0";
+ return new CompletedSnapshotSplitInfo(
+ splitId,
+ tableId,
+ null,
+ null,
+ null,
+ new SnapshotSplitWatermark(
+ splitId, new TestOffset(lowWatermark), new
TestOffset(highWatermark)));
+ }
+
+ static SourceRecord createDataEventWithSource(TableId tableId, long
position) {
+ Schema sourceSchema =
+ SchemaBuilder.struct()
+ .name("io.debezium.connector.postgresql.Source")
+ .field("db", Schema.STRING_SCHEMA)
+ .field("schema", Schema.STRING_SCHEMA)
+ .field("table", Schema.STRING_SCHEMA)
+ .field(DEBEZIUM_CONNECTOR_KEY, Schema.STRING_SCHEMA)
+ .build();
+ Struct sourceStruct = new Struct(sourceSchema);
+ sourceStruct.put("db", tableId.catalog());
+ sourceStruct.put("schema", tableId.schema());
+ sourceStruct.put("table", tableId.table());
+ sourceStruct.put(DEBEZIUM_CONNECTOR_KEY, "postgresql");
+
+ Schema valueSchema =
+ SchemaBuilder.struct()
+ .field(Envelope.FieldName.SOURCE, sourceSchema)
+ .field(Envelope.FieldName.OPERATION,
Schema.STRING_SCHEMA)
+ .build();
+ Struct value = new Struct(valueSchema);
+ value.put(Envelope.FieldName.SOURCE, sourceStruct);
+ value.put(Envelope.FieldName.OPERATION, "u");
+ return new SourceRecord(
+ Collections.emptyMap(),
+ Collections.singletonMap("pos", String.valueOf(position)),
+ tableId.toString(),
+ null,
+ null,
+ valueSchema,
+ value);
+ }
+
+ static class TestOffset extends Offset {
+ private static final long serialVersionUID = 1L;
+
+ private final long position;
+
+ TestOffset(long position) {
+ this.position = position;
+ this.offset = Collections.singletonMap("pos",
String.valueOf(position));
+ }
+
+ @Override
+ public int compareTo(Offset o) {
+ return Long.compare(position, ((TestOffset) o).position);
+ }
+ }
+
public static class TestConnectorConfig extends CommonConnectorConfig {
protected TestConnectorConfig(
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/split/state/IncrementalSplitStateTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/split/state/IncrementalSplitStateTest.java
index 70fa137b16..b008250a9a 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/split/state/IncrementalSplitStateTest.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/split/state/IncrementalSplitStateTest.java
@@ -32,7 +32,9 @@ import lombok.ToString;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -130,6 +132,37 @@ public class IncrementalSplitStateTest {
Assertions.assertFalse(splitState.autoEnterPureIncrementPhaseIfAllowed());
}
+ @Test
+ public void
testCheckpointKeepsRestoredTableOffsetsUntilTheSplitReadsPastThem() {
+ TableId table1 = new TableId("db", "schema", "table1");
+ TableId table2 = new TableId("db", "schema", "table2");
+ Map<TableId, Offset> tableStartupOffsets = new HashMap<>();
+ tableStartupOffsets.put(table1, new TestOffset(150));
+ tableStartupOffsets.put(table2, new TestOffset(200));
+ IncrementalSplit split =
+ new IncrementalSplit(
+ "test",
+ Arrays.asList(table1, table2),
+ new TestOffset(100),
+ null,
+ Collections.emptyList(),
+ null,
+ Collections.emptyMap(),
+ tableStartupOffsets);
+ IncrementalSplitState splitState = new IncrementalSplitState(split);
+
+ Assertions.assertEquals(
+ tableStartupOffsets,
splitState.toSourceSplit().getTableStartupOffsets());
+
+ splitState.setStartupOffset(new TestOffset(150));
+ Assertions.assertEquals(
+ Collections.singletonMap(table2, new TestOffset(200)),
+ splitState.toSourceSplit().getTableStartupOffsets());
+
+ splitState.setStartupOffset(new TestOffset(200));
+
Assertions.assertTrue(splitState.toSourceSplit().getTableStartupOffsets().isEmpty());
+ }
+
private static IncrementalSplit createIncrementalSplit(
Offset startupOffset, List<CompletedSnapshotSplitInfo>
snapshotSplits) {
return new IncrementalSplit(
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/AbstractMysqlCDCITBase.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/AbstractMysqlCDCITBase.java
index ede5b78ebb..e403266b80 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/AbstractMysqlCDCITBase.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/AbstractMysqlCDCITBase.java
@@ -65,6 +65,7 @@ public abstract class AbstractMysqlCDCITBase extends
TestSuiteBase implements Te
protected static final String MYSQL_USER_PASSWORD = "mysqlpw";
protected static final String MYSQL_DATABASE = "mysql_cdc";
private static final String MYSQL_DATABASE2 = "mysql_cdc2";
+ private static final String APPEND_SINK_DATABASE = "mysql_cdc_append";
private final String QUERY_SQL = "select * from %s.%s";
@@ -1053,6 +1054,111 @@ public abstract class AbstractMysqlCDCITBase extends
TestSuiteBase implements Te
log.info("****************** container logs end ******************");
}
+ @TestTemplate
+ @DisabledOnContainer(
+ value = {},
+ type = {EngineType.SPARK, EngineType.FLINK},
+ disabledReason = "Currently SPARK and FLINK do not support
restore")
+ public void
testExactlyOnceRestoreWithAddedTableKeepsChangesOfExistingTables(
+ TestContainer container) throws IOException, InterruptedException {
+ String[] tables = {SOURCE_TABLE_1, SOURCE_TABLE_2,
SOURCE_TABLE_1_CUSTOM_PRIMARY_KEY};
+ for (String table : tables) {
+ clearTable(MYSQL_DATABASE, table);
+ clearTable(MYSQL_DATABASE2, table);
+ createAppendOnlySinkTable(table);
+ }
+
+ initSourceTable(MYSQL_DATABASE, SOURCE_TABLE_1);
+ initSourceTable(MYSQL_DATABASE, SOURCE_TABLE_2);
+
+ Long jobId = JobIdGenerator.newJobId();
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return container.executeJob(
+
"/mysqlcdc_to_mysql_with_exactly_once_restore_two_tables.conf",
+ String.valueOf(jobId));
+ } catch (Exception e) {
+ log.error("Commit task exception :" + e.getMessage());
+ throw new RuntimeException(e);
+ }
+ });
+ awaitSinksMatchSource(SOURCE_TABLE_1, SOURCE_TABLE_2);
+
+ insertSourceTableRows(MYSQL_DATABASE, SOURCE_TABLE_1, 100, 110);
+ insertSourceTableRows(MYSQL_DATABASE, SOURCE_TABLE_2, 100, 110);
+ awaitSinksMatchSource(SOURCE_TABLE_1, SOURCE_TABLE_2);
+
+ Assertions.assertEquals(0,
container.savepointJob(String.valueOf(jobId)).getExitCode());
+
+ // Changes made while the job is stopped are after the saved offset
and before the
+ // snapshot of the added table.
+ insertSourceTableRows(MYSQL_DATABASE, SOURCE_TABLE_1, 200, 210);
+ insertSourceTableRows(MYSQL_DATABASE, SOURCE_TABLE_2, 200, 210);
+ initSourceTable(MYSQL_DATABASE, SOURCE_TABLE_1_CUSTOM_PRIMARY_KEY);
+
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ container.restoreJob(
+
"/mysqlcdc_to_mysql_with_exactly_once_restore_add_table.conf",
+ String.valueOf(jobId));
+ } catch (Exception e) {
+ log.error("Commit task exception :" + e.getMessage());
+ throw new RuntimeException(e);
+ }
+ return null;
+ });
+ insertSourceTableRows(MYSQL_DATABASE, SOURCE_TABLE_1, 300, 310);
+
+ awaitSinksMatchSource(tables);
+
+ insertSourceTableRows(MYSQL_DATABASE, SOURCE_TABLE_2, 400, 410);
+ insertSourceTableRows(MYSQL_DATABASE,
SOURCE_TABLE_1_CUSTOM_PRIMARY_KEY, 400, 410);
+ awaitSinksMatchSource(tables);
+
+ Assertions.assertEquals(0,
container.cancelJob(String.valueOf(jobId)).getExitCode());
+ }
+
+ /**
+ * Creates a copy of the sink table without a primary key, so that every
emitted row is appended
+ * and a change emitted twice shows up as an extra row.
+ */
+ private void createAppendOnlySinkTable(String tableName) {
+ String appendSinkTable = APPEND_SINK_DATABASE + "." + tableName;
+ executeSql("CREATE DATABASE IF NOT EXISTS " + APPEND_SINK_DATABASE);
+ executeSql("DROP TABLE IF EXISTS " + appendSinkTable);
+ executeSql(
+ "CREATE TABLE " + appendSinkTable + " LIKE " + MYSQL_DATABASE2
+ "." + tableName);
+ executeSql("ALTER TABLE " + appendSinkTable + " MODIFY id INT NOT
NULL, DROP PRIMARY KEY");
+ }
+
+ private void awaitSinksMatchSource(String... tableNames) {
+ await().atMost(300000, TimeUnit.MILLISECONDS)
+ .pollInterval(1000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () -> {
+ for (String tableName : tableNames) {
+ List<List<Object>> sourceRows =
+ query(
+
getSourceQuerySQL(MYSQL_DATABASE, tableName)
+ + " order by id");
+ Assertions.assertIterableEquals(
+ sourceRows,
+ query(
+
getSourceQuerySQL(MYSQL_DATABASE2, tableName)
+ + " order by id"),
+ tableName);
+ Assertions.assertIterableEquals(
+ sourceRows,
+ query(
+
getSourceQuerySQL(APPEND_SINK_DATABASE, tableName)
+ + " order by id"),
+ APPEND_SINK_DATABASE + "." +
tableName);
+ }
+ });
+ }
+
@TestTemplate
@DisabledOnContainer(
value = {},
@@ -1818,7 +1924,12 @@ public abstract class AbstractMysqlCDCITBase extends
TestSuiteBase implements Te
}
private void initSourceTable(String database, String tableName) {
- for (int i = 1; i < 100; i++) {
+ insertSourceTableRows(database, tableName, 1, 100);
+ }
+
+ private void insertSourceTableRows(
+ String database, String tableName, int fromId, int toIdExclusive) {
+ for (int i = fromId; i < toIdExclusive; i++) {
executeSql(
"INSERT INTO "
+ database
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_exactly_once_restore_add_table.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_exactly_once_restore_add_table.conf
new file mode 100644
index 0000000000..379894e505
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_exactly_once_restore_add_table.conf
@@ -0,0 +1,68 @@
+#
+# 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.
+#
+######
+###### Restores the two-table job with a third table added
+######
+
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 5000
+}
+
+source {
+ MySQL-CDC {
+ plugin_output = "customers_mysql_cdc"
+ server-id = 5652-5660
+ username = "st_user_source"
+ password = "mysqlpw"
+ table-names = ["mysql_cdc.mysql_cdc_e2e_source_table",
"mysql_cdc.mysql_cdc_e2e_source_table2",
"mysql_cdc.mysql_cdc_e2e_source_table_1_custom_primary_key"]
+ table-names-config = [
+ {
+ table = "mysql_cdc.mysql_cdc_e2e_source_table_1_custom_primary_key"
+ primaryKeys = ["id"]
+ }
+ ]
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc"
+ exactly_once = true
+ }
+}
+
+sink {
+ jdbc {
+ plugin_input = "customers_mysql_cdc"
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc2"
+ driver = "com.mysql.cj.jdbc.Driver"
+ user = "st_user_sink"
+ password = "mysqlpw"
+ database = "mysql_cdc2"
+ table = "${table_name}"
+ primary_keys = ["${primary_key}"]
+ generate_sink_sql = true
+ }
+
+ jdbc {
+ plugin_input = "customers_mysql_cdc"
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc_append"
+ driver = "com.mysql.cj.jdbc.Driver"
+ user = "st_user_sink"
+ password = "mysqlpw"
+ database = "mysql_cdc_append"
+ table = "${table_name}"
+ generate_sink_sql = true
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_exactly_once_restore_two_tables.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_exactly_once_restore_two_tables.conf
new file mode 100644
index 0000000000..f7aef0ac5c
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_exactly_once_restore_two_tables.conf
@@ -0,0 +1,62 @@
+#
+# 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.
+#
+######
+###### Exactly-once job on two tables, savepointed before a table is added
+######
+
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 5000
+}
+
+source {
+ MySQL-CDC {
+ plugin_output = "customers_mysql_cdc"
+ server-id = 5652-5660
+ username = "st_user_source"
+ password = "mysqlpw"
+ table-names = ["mysql_cdc.mysql_cdc_e2e_source_table",
"mysql_cdc.mysql_cdc_e2e_source_table2"]
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc"
+ exactly_once = true
+ }
+}
+
+sink {
+ jdbc {
+ plugin_input = "customers_mysql_cdc"
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc2"
+ driver = "com.mysql.cj.jdbc.Driver"
+ user = "st_user_sink"
+ password = "mysqlpw"
+ database = "mysql_cdc2"
+ table = "${table_name}"
+ primary_keys = ["${primary_key}"]
+ generate_sink_sql = true
+ }
+
+ jdbc {
+ plugin_input = "customers_mysql_cdc"
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc_append"
+ driver = "com.mysql.cj.jdbc.Driver"
+ user = "st_user_sink"
+ password = "mysqlpw"
+ database = "mysql_cdc_append"
+ table = "${table_name}"
+ generate_sink_sql = true
+ }
+}
