This is an automated email from the ASF dual-hosted git repository.
leonardBang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new d9766a35b [FLINK-40216][cdc-base] Preserve assignment order of splits
across restore to keep meta groups consistent (#4489)
d9766a35b is described below
commit d9766a35bc286ffb0238008e827a7d43c5debc41
Author: κΉλν <[email protected]>
AuthorDate: Fri Oct 23 10:56:19 2026 +0900
[FLINK-40216][cdc-base] Preserve assignment order of splits across restore
to keep meta groups consistent (#4489)
This closes #4489.
---
.../base/source/assigner/HybridSplitAssigner.java | 6 +-
.../source/assigner/SnapshotSplitAssigner.java | 29 ++--
.../state/PendingSplitsStateSerializer.java | 17 +-
.../base/source/meta/split/StreamSplit.java | 39 ++++-
.../source/reader/IncrementalSourceReader.java | 47 ++----
.../source/assigner/MetaGroupOrderingTest.java | 182 +++++++++++++++++++++
.../state/PendingSplitsStateSerializerTest.java | 51 ++++++
.../state/SnapshotPendingSplitsStateTest.java | 114 +++++++++++++
.../base/source/meta/split/StreamSplitTest.java | 62 +++++++
.../source/reader/IncrementalSourceReaderTest.java | 60 +++++++
10 files changed, 543 insertions(+), 64 deletions(-)
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/HybridSplitAssigner.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/HybridSplitAssigner.java
index 046933e16..6c8202491 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/HybridSplitAssigner.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/HybridSplitAssigner.java
@@ -38,12 +38,10 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
-import java.util.stream.Collectors;
import static
org.apache.flink.cdc.connectors.base.source.assigner.AssignerStatus.isInitialAssigningFinished;
import static
org.apache.flink.cdc.connectors.base.source.assigner.AssignerStatus.isNewlyAddedAssigningFinished;
@@ -303,9 +301,7 @@ public class HybridSplitAssigner<C extends SourceConfig>
implements SplitAssigne
"Cannot create a stream split after the snapshot split
metadata was released.");
}
final List<SchemalessSnapshotSplit> assignedSnapshotSplit =
- snapshotSplitAssigner.getAssignedSplits().values().stream()
- .sorted(Comparator.comparing(SourceSplitBase::splitId))
- .collect(Collectors.toList());
+ new
ArrayList<>(snapshotSplitAssigner.getAssignedSplits().values());
Map<String, Offset> splitFinishedOffsets =
snapshotSplitAssigner.getSplitFinishedOffsets();
final List<FinishedSnapshotSplitInfo> finishedSnapshotSplitInfos = new
ArrayList<>();
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/SnapshotSplitAssigner.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/SnapshotSplitAssigner.java
index 633f08957..5eda2be6c 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/SnapshotSplitAssigner.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/SnapshotSplitAssigner.java
@@ -29,6 +29,7 @@ import
org.apache.flink.cdc.connectors.base.source.meta.split.SchemalessSnapshot
import org.apache.flink.cdc.connectors.base.source.meta.split.SnapshotSplit;
import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitBase;
import
org.apache.flink.cdc.connectors.base.source.metrics.SourceEnumeratorMetrics;
+import
org.apache.flink.cdc.connectors.base.source.reader.IncrementalSourceReader;
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.Preconditions;
@@ -72,7 +73,21 @@ public class SnapshotSplitAssigner<C extends SourceConfig>
implements SplitAssig
private final List<TableId> alreadyProcessedTables;
private final List<SchemalessSnapshotSplit> remainingSplits;
- private final Map<String, SchemalessSnapshotSplit> assignedSplits;
+
+ /**
+ * The splits that have been assigned to a reader. Once a split is
finished, it remains in this
+ * map. An entry added to {@link #splitFinishedOffsets} indicates that the
split has been
+ * finished. If reading the split fails, it is removed from this map.
+ *
+ * <p>{@link IncrementalSourceReader} relies on the order of elements
within the map:
+ *
+ * <ol>
+ * <li>It must correspond to the order of assignment of the splits to
readers.
+ * <li>The order must be retained across job restarts.
+ * </ol>
+ */
+ private final LinkedHashMap<String, SchemalessSnapshotSplit>
assignedSplits;
+
private final Map<TableId, TableChanges.TableChange> tableSchemas;
private final Map<String, Offset> splitFinishedOffsets;
@@ -171,17 +186,7 @@ public class SnapshotSplitAssigner<C extends SourceConfig>
implements SplitAssig
this.currentParallelism = currentParallelism;
this.alreadyProcessedTables = alreadyProcessedTables;
this.remainingSplits = remainingSplits;
- // When job restore from savepoint, sort the existing tables and newly
added tables
- // to let enumerator only send newly added tables' StreamSplitMetaEvent
- this.assignedSplits =
- assignedSplits.entrySet().stream()
- .sorted(Map.Entry.comparingByKey())
- .collect(
- Collectors.toMap(
- Map.Entry::getKey,
- Map.Entry::getValue,
- (o, o2) -> o,
- LinkedHashMap::new));
+ this.assignedSplits = new LinkedHashMap<>(assignedSplits);
this.tableSchemas = tableSchemas;
this.splitFinishedOffsets = splitFinishedOffsets;
this.assignerStatus = assignerStatus;
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/state/PendingSplitsStateSerializer.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/state/PendingSplitsStateSerializer.java
index d07c2320a..e1430eff4 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/state/PendingSplitsStateSerializer.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/state/PendingSplitsStateSerializer.java
@@ -35,6 +35,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -236,12 +237,12 @@ public class PendingSplitsStateSerializer implements
SimpleVersionedSerializer<P
int splitVersion, DataInputDeserializer in) throws IOException {
List<TableId> alreadyProcessedTables = readTableIds(2, in);
List<SnapshotSplit> remainingSplits = readSnapshotSplits(splitVersion,
in);
- Map<String, SnapshotSplit> assignedSnapshotSplits =
+ LinkedHashMap<String, SnapshotSplit> assignedSnapshotSplits =
readAssignedSnapshotSplits(splitVersion, in);
final List<SchemalessSnapshotSplit> remainingSchemalessSplits = new
ArrayList<>();
- final Map<String, SchemalessSnapshotSplit>
assignedSchemalessSnapshotSplits =
- new HashMap<>();
+ final LinkedHashMap<String, SchemalessSnapshotSplit>
assignedSchemalessSnapshotSplits =
+ new LinkedHashMap<>();
final Map<TableId, TableChanges.TableChange> tableSchemas = new
HashMap<>();
remainingSplits.forEach(
split -> {
@@ -290,7 +291,7 @@ public class PendingSplitsStateSerializer implements
SimpleVersionedSerializer<P
int version, int splitVersion, DataInputDeserializer in) throws
IOException {
List<TableId> alreadyProcessedTables = readTableIds(version, in);
List<SnapshotSplit> remainingSplits = readSnapshotSplits(splitVersion,
in);
- Map<String, SnapshotSplit> assignedSnapshotSplits =
+ LinkedHashMap<String, SnapshotSplit> assignedSnapshotSplits =
readAssignedSnapshotSplits(splitVersion, in);
Map<String, Offset> finishedOffsets =
readFinishedOffsets(splitVersion, in);
AssignerStatus assignerStatus;
@@ -307,8 +308,8 @@ public class PendingSplitsStateSerializer implements
SimpleVersionedSerializer<P
List<TableId> remainingTableIds = readTableIds(version, in);
boolean isTableIdCaseSensitive = in.readBoolean();
final List<SchemalessSnapshotSplit> remainingSchemalessSplits = new
ArrayList<>();
- final Map<String, SchemalessSnapshotSplit>
assignedSchemalessSnapshotSplits =
- new HashMap<>();
+ final LinkedHashMap<String, SchemalessSnapshotSplit>
assignedSchemalessSnapshotSplits =
+ new LinkedHashMap<>();
final Map<TableId, TableChanges.TableChange> tableSchemas = new
HashMap<>();
remainingSplits.forEach(
split -> {
@@ -444,9 +445,9 @@ public class PendingSplitsStateSerializer implements
SimpleVersionedSerializer<P
}
}
- private Map<String, SnapshotSplit> readAssignedSnapshotSplits(
+ private LinkedHashMap<String, SnapshotSplit> readAssignedSnapshotSplits(
int splitVersion, DataInputDeserializer in) throws IOException {
- Map<String, SnapshotSplit> assignedSplits = new HashMap<>();
+ LinkedHashMap<String, SnapshotSplit> assignedSplits = new
LinkedHashMap<>();
final int size = in.readInt();
for (int i = 0; i < size; i++) {
String splitId = in.readUTF();
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/meta/split/StreamSplit.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/meta/split/StreamSplit.java
index cdeb247f3..99b45c2ea 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/meta/split/StreamSplit.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/meta/split/StreamSplit.java
@@ -28,6 +28,7 @@ import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -42,7 +43,10 @@ public class StreamSplit extends SourceSplitBase {
private final Offset startingOffset;
private final Offset endingOffset;
+
+ /** Split IDs of all elements must be unique. */
private final List<FinishedSnapshotSplitInfo> finishedSnapshotSplitInfos;
+
private final Map<TableId, TableChange> tableSchemas;
private final int totalFinishedSplitSize;
@@ -66,6 +70,9 @@ public class StreamSplit extends SourceSplitBase {
boolean isSuspended,
boolean isSnapshotCompleted) {
super(splitId);
+
+ ensureNoDuplicates(finishedSnapshotSplitInfos);
+
this.startingOffset = startingOffset;
this.endingOffset = endingOffset;
this.finishedSnapshotSplitInfos = finishedSnapshotSplitInfos;
@@ -82,14 +89,30 @@ public class StreamSplit extends SourceSplitBase {
List<FinishedSnapshotSplitInfo> finishedSnapshotSplitInfos,
Map<TableId, TableChange> tableSchemas,
int totalFinishedSplitSize) {
- super(splitId);
- this.startingOffset = startingOffset;
- this.endingOffset = endingOffset;
- this.finishedSnapshotSplitInfos = finishedSnapshotSplitInfos;
- this.tableSchemas = tableSchemas;
- this.totalFinishedSplitSize = totalFinishedSplitSize;
- this.isSuspended = false;
- this.isSnapshotCompleted = false;
+ this(
+ splitId,
+ startingOffset,
+ endingOffset,
+ finishedSnapshotSplitInfos,
+ tableSchemas,
+ totalFinishedSplitSize,
+ false,
+ false);
+ }
+
+ private static void ensureNoDuplicates(
+ List<FinishedSnapshotSplitInfo> finishedSnapshotSplitInfos) {
+ Set<String> seenSplitIds = new HashSet<>();
+ for (FinishedSnapshotSplitInfo splitInfo : finishedSnapshotSplitInfos)
{
+ if (seenSplitIds.contains(splitInfo.getSplitId())) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Found duplicate split ID %s in finished
snapshot split infos",
+ splitInfo.getSplitId()));
+ }
+
+ seenSplitIds.add(splitInfo.getSplitId());
+ }
}
public Offset getStartingOffset() {
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceReader.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceReader.java
index b629d1fd6..dd68c8663 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceReader.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceReader.java
@@ -60,10 +60,8 @@ import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import java.util.function.Supplier;
import java.util.stream.Collectors;
@@ -384,28 +382,6 @@ public class IncrementalSourceReader<T, C extends
SourceConfig>
}
}
- private Set<String> getExistedSplitsOfLastGroup(
- List<FinishedSnapshotSplitInfo> finishedSnapshotSplits, int
metaGroupSize) {
- int splitsNumOfLastGroup =
- finishedSnapshotSplits.size() %
sourceConfig.getSplitMetaGroupSize();
- if (splitsNumOfLastGroup != 0) {
- int lastGroupStart =
- ((int) (finishedSnapshotSplits.size() /
sourceConfig.getSplitMetaGroupSize()))
- * metaGroupSize;
- // Keep same order with HybridSplitAssigner.createStreamSplit() to
avoid
- // 'invalid request meta group id' error
- List<String> sortedFinishedSnapshotSplits =
- finishedSnapshotSplits.stream()
- .map(FinishedSnapshotSplitInfo::getSplitId)
- .sorted()
- .collect(Collectors.toList());
- return new HashSet<>(
- sortedFinishedSnapshotSplits.subList(
- lastGroupStart, lastGroupStart +
splitsNumOfLastGroup));
- }
- return new HashSet<>();
- }
-
@Override
public void handleSourceEvents(SourceEvent sourceEvent) {
if (sourceEvent instanceof FinishedSnapshotSplitsAckEvent) {
@@ -462,15 +438,24 @@ public class IncrementalSourceReader<T, C extends
SourceConfig>
streamSplit = toNormalStreamSplit(streamSplit,
receivedTotalFinishedSplitSize);
uncompletedStreamSplits.put(streamSplit.splitId(),
streamSplit);
} else if (receivedMetaGroupId == expectedMetaGroupId) {
- Set<String> existedSplitsOfLastGroup =
- getExistedSplitsOfLastGroup(
- streamSplit.getFinishedSnapshotSplitInfos(),
- sourceConfig.getSplitMetaGroupSize());
-
+ int expectedNumberOfAlreadyRetrievedElements =
+ streamSplit.getFinishedSnapshotSplitInfos().size()
+ % sourceConfig.getSplitMetaGroupSize();
+ List<byte[]> metaGroup = metadataEvent.getMetaGroup();
+ if (expectedNumberOfAlreadyRetrievedElements > 0) {
+ LOG.info(
+ "Source reader {} is discarding the first {} out
of {} elements of meta group {}.",
+ subtaskId,
+ expectedNumberOfAlreadyRetrievedElements,
+ metaGroup.size(),
+ receivedMetaGroupId);
+ metaGroup =
+ metaGroup.subList(
+ expectedNumberOfAlreadyRetrievedElements,
metaGroup.size());
+ }
List<FinishedSnapshotSplitInfo> newAddedMetadataGroup =
- metadataEvent.getMetaGroup().stream()
+ metaGroup.stream()
.map(sourceSplitSerializer::deserialize)
- .filter(r ->
!existedSplitsOfLastGroup.contains(r.getSplitId()))
.collect(Collectors.toList());
uncompletedStreamSplits.put(
streamSplit.splitId(),
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/assigner/MetaGroupOrderingTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/assigner/MetaGroupOrderingTest.java
new file mode 100644
index 000000000..93ad35bc5
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/assigner/MetaGroupOrderingTest.java
@@ -0,0 +1,182 @@
+/*
+ * 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.flink.cdc.connectors.base.source.assigner;
+
+import org.apache.flink.cdc.connectors.base.config.SourceConfig;
+import org.apache.flink.cdc.connectors.base.dialect.DataSourceDialect;
+import
org.apache.flink.cdc.connectors.base.source.assigner.state.ChunkSplitterState;
+import
org.apache.flink.cdc.connectors.base.source.assigner.state.SnapshotPendingSplitsState;
+import org.apache.flink.cdc.connectors.base.source.meta.offset.Offset;
+import org.apache.flink.cdc.connectors.base.source.meta.offset.OffsetFactory;
+import
org.apache.flink.cdc.connectors.base.source.meta.split.FinishedSnapshotSplitInfo;
+import
org.apache.flink.cdc.connectors.base.source.meta.split.SchemalessSnapshotSplit;
+import
org.apache.flink.cdc.connectors.base.source.reader.IncrementalSourceReader;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.apache.flink.shaded.guava31.com.google.common.collect.Lists;
+
+import io.debezium.relational.TableId;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+
+/**
+ * Tests that the meta groups of finished snapshot splits stay consistent
across a checkpoint
+ * restore (FLINK-40216).
+ *
+ * <p>The meta groups served by {@code
IncrementalSourceEnumerator#sendStreamMetaRequestEvent} are
+ * partitioned from {@link SnapshotSplitAssigner#getFinishedSplitInfos()}, so
the assignment order
+ * of the splits must be retained across job restarts; otherwise a reader that
synchronized only
+ * part of the meta groups before a restart receives duplicated split infos
while never receiving
+ * others.
+ */
+class MetaGroupOrderingTest {
+
+ private static final TableId TABLE_ID = new TableId("db", null, "t");
+ private static final RowType SPLIT_KEY_TYPE =
+ new RowType(Collections.singletonList(new RowType.RowField("id",
new BigIntType())));
+
+ @Test
+ void testFinishedSplitInfosPreserveAssignmentOrderAfterRestore() {
+ List<String> assignmentOrder = new ArrayList<>();
+ SnapshotSplitAssigner<SourceConfig> restoredAssigner =
restoreAssigner(assignmentOrder, 12);
+
+ List<String> restoredOrder =
+ restoredAssigner.getFinishedSplitInfos().stream()
+ .map(FinishedSnapshotSplitInfo::getSplitId)
+ .collect(Collectors.toList());
+
+ assertThat(restoredOrder).isEqualTo(assignmentOrder);
+ }
+
+ @Test
+ void testReaderResumingAfterRestoreReceivesEachSplitExactlyOnce() {
+ int metaGroupSize = 2;
+ List<String> assignmentOrder = new ArrayList<>();
+ SnapshotSplitAssigner<SourceConfig> restoredAssigner =
restoreAssigner(assignmentOrder, 12);
+
+ // Before the restart the enumerator serves groups partitioned from
the assignment order;
+ // the reader synchronized groups 0..2 into its stream split state:
+ // [t:0, t:1], [t:2, t:3], [t:4, t:5]
+ List<List<String>> groupsBeforeRestart =
Lists.partition(assignmentOrder, metaGroupSize);
+ List<String> readerState = new ArrayList<>();
+ for (int groupId = 0; groupId < 3; groupId++) {
+ readerState.addAll(groupsBeforeRestart.get(groupId));
+ }
+
+ // The job restarts: the enumerator rebuilds its meta groups from the
restored assigner
+ List<List<FinishedSnapshotSplitInfo>> groupsAfterRestart =
+ Lists.partition(restoredAssigner.getFinishedSplitInfos(),
metaGroupSize);
+
+ // The reader resumes requesting group ids based on how many infos it
already holds
+ int nextGroupId =
+ IncrementalSourceReader.getNextMetaGroupId(readerState.size(),
metaGroupSize);
+ while (readerState.size() < assignmentOrder.size()
+ && nextGroupId < groupsAfterRestart.size()) {
+ for (FinishedSnapshotSplitInfo info :
groupsAfterRestart.get(nextGroupId)) {
+ readerState.add(info.getSplitId());
+ }
+ nextGroupId =
+
IncrementalSourceReader.getNextMetaGroupId(readerState.size(), metaGroupSize);
+ }
+
+ assertThat(readerState).containsExactlyElementsOf(assignmentOrder);
+ }
+
+ private static SnapshotSplitAssigner<SourceConfig> restoreAssigner(
+ List<String> assignmentOrder, int splitCount) {
+ LinkedHashMap<String, SchemalessSnapshotSplit> assignedSplits = new
LinkedHashMap<>();
+ Map<String, Offset> splitFinishedOffsets = new HashMap<>();
+ // 10+ splits so that the lexicographic order of split ids (t:0, t:1,
t:10, t:11, t:2,
+ // ...) differs from the assignment order (t:0, t:1, t:2, ..., t:11)
+ for (int i = 0; i < splitCount; i++) {
+ String splitId = TABLE_ID + ":" + i;
+ assignmentOrder.add(splitId);
+ assignedSplits.put(
+ splitId,
+ new SchemalessSnapshotSplit(
+ TABLE_ID,
+ splitId,
+ SPLIT_KEY_TYPE,
+ new Object[] {i},
+ new Object[] {i + 1},
+ null));
+ splitFinishedOffsets.put(splitId, null);
+ }
+
+ return new SnapshotSplitAssigner<>(
+ mock(SourceConfig.class),
+ 4,
+ new SnapshotPendingSplitsState(
+ Collections.singletonList(TABLE_ID),
+ Collections.emptyList(),
+ assignedSplits,
+ Collections.emptyMap(),
+ splitFinishedOffsets,
+ AssignerStatus.INITIAL_ASSIGNING_FINISHED,
+ Collections.emptyList(),
+ false,
+ true,
+ Collections.emptyMap(),
+ ChunkSplitterState.NO_SPLITTING_TABLE_STATE),
+ mock(DataSourceDialect.class),
+ new MockOffsetFactory());
+ }
+
+ private static class MockOffsetFactory extends OffsetFactory {
+ @Override
+ public Offset newOffset(Map<String, String> offset) {
+ return null;
+ }
+
+ @Override
+ public Offset newOffset(String filename, Long position) {
+ return null;
+ }
+
+ @Override
+ public Offset newOffset(Long position) {
+ return null;
+ }
+
+ @Override
+ public Offset createTimestampOffset(long timestampMillis) {
+ return null;
+ }
+
+ @Override
+ public Offset createInitialOffset() {
+ return null;
+ }
+
+ @Override
+ public Offset createNoStoppingOffset() {
+ return null;
+ }
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/assigner/state/PendingSplitsStateSerializerTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/assigner/state/PendingSplitsStateSerializerTest.java
index c7bdb617d..2d5cf33e1 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/assigner/state/PendingSplitsStateSerializerTest.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/assigner/state/PendingSplitsStateSerializerTest.java
@@ -20,6 +20,7 @@ package
org.apache.flink.cdc.connectors.base.source.assigner.state;
import org.apache.flink.cdc.connectors.base.source.assigner.AssignerStatus;
import org.apache.flink.cdc.connectors.base.source.meta.offset.Offset;
import org.apache.flink.cdc.connectors.base.source.meta.offset.OffsetFactory;
+import
org.apache.flink.cdc.connectors.base.source.meta.split.SchemalessSnapshotSplit;
import org.apache.flink.cdc.connectors.base.source.meta.split.SnapshotSplit;
import
org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitSerializer;
import org.apache.flink.cdc.connectors.base.source.meta.split.StreamSplit;
@@ -38,6 +39,8 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
@@ -176,6 +179,54 @@ class PendingSplitsStateSerializerTest {
snapshotMetaReleased);
}
+ @Test
+ void testAssignedSplitsKeepAssignmentOrderAfterSerde() throws Exception {
+ TableId tableId = constructTableId();
+ RowType splitKeyType =
+ new RowType(
+ Collections.singletonList(new RowType.RowField("id",
new BigIntType())));
+ // 10+ splits so that the lexicographic order of split ids differs
from the assignment
+ // order, which determines the meta group partitioning of finished
snapshot splits
+ List<String> assignmentOrder = new ArrayList<>();
+ LinkedHashMap<String, SchemalessSnapshotSplit> assignedSplits = new
LinkedHashMap<>();
+ for (int i = 0; i < 12; i++) {
+ String splitId = tableId + ":" + i;
+ assignmentOrder.add(splitId);
+ assignedSplits.put(
+ splitId,
+ new SchemalessSnapshotSplit(
+ tableId,
+ splitId,
+ splitKeyType,
+ new Object[] {i},
+ new Object[] {i + 1},
+ null));
+ }
+ PendingSplitsStateSerializer serializer =
+ new
PendingSplitsStateSerializer(constructSourceSplitSerializer());
+ SnapshotPendingSplitsState state =
+ new SnapshotPendingSplitsState(
+ Collections.singletonList(tableId),
+ Collections.emptyList(),
+ assignedSplits,
+ constructTableSchema(),
+ Collections.emptyMap(),
+ AssignerStatus.INITIAL_ASSIGNING_FINISHED,
+ Collections.emptyList(),
+ false,
+ true,
+ Collections.emptyMap(),
+ ChunkSplitterState.NO_SPLITTING_TABLE_STATE);
+
+ SnapshotPendingSplitsState restoredState =
+ (SnapshotPendingSplitsState)
+ serializer.deserialize(
+ serializer.getVersion(),
serializer.serialize(state));
+
+ assertThat(new ArrayList<>(restoredState.getAssignedSplits().keySet()))
+ .isEqualTo(assignmentOrder);
+ }
+
private SourceSplitSerializer constructSourceSplitSerializer() {
return new SourceSplitSerializer() {
@Override
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/assigner/state/SnapshotPendingSplitsStateTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/assigner/state/SnapshotPendingSplitsStateTest.java
new file mode 100644
index 000000000..40ee65480
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/assigner/state/SnapshotPendingSplitsStateTest.java
@@ -0,0 +1,114 @@
+/*
+ * 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.flink.cdc.connectors.base.source.assigner.state;
+
+import org.apache.flink.cdc.connectors.base.source.assigner.AssignerStatus;
+import
org.apache.flink.cdc.connectors.base.source.meta.split.SchemalessSnapshotSplit;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.RowType;
+
+import io.debezium.relational.TableId;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+/** Tests for {@link SnapshotPendingSplitsState}. */
+class SnapshotPendingSplitsStateTest {
+
+ /**
+ * Connectors compiled against earlier versions of flink-cdc-base link
against these exact
+ * method descriptors, so the parameter and return types must stay {@link
Map}.
+ */
+ @Test
+ void testPublicSignaturesKeepMapType() throws Exception {
+
assertThat(SnapshotPendingSplitsState.class.getMethod("getAssignedSplits").getReturnType())
+ .isEqualTo(Map.class);
+ assertThatCode(
+ () ->
+
SnapshotPendingSplitsState.class.getConstructor(
+ List.class,
+ List.class,
+ Map.class,
+ Map.class,
+ Map.class,
+ AssignerStatus.class,
+ List.class,
+ boolean.class,
+ boolean.class,
+ Map.class,
+ ChunkSplitterState.class))
+ .doesNotThrowAnyException();
+ assertThatCode(
+ () ->
+
SnapshotPendingSplitsState.class.getConstructor(
+ List.class,
+ List.class,
+ Map.class,
+ Map.class,
+ Map.class,
+ AssignerStatus.class,
+ List.class,
+ boolean.class,
+ boolean.class,
+ Map.class,
+ ChunkSplitterState.class,
+ boolean.class))
+ .doesNotThrowAnyException();
+ }
+
+ @Test
+ void testAssignedSplitsKeepIterationOrderOfGivenMap() {
+ TableId tableId = TableId.parse("catalog.schema.table");
+ RowType splitKeyType =
+ new RowType(
+ Collections.singletonList(new RowType.RowField("id",
new BigIntType())));
+ // not in lexicographic order of split ids
+ List<String> assignmentOrder =
+ Arrays.asList(tableId + ":2", tableId + ":10", tableId + ":1");
+ Map<String, SchemalessSnapshotSplit> assignedSplits = new
LinkedHashMap<>();
+ for (String splitId : assignmentOrder) {
+ assignedSplits.put(
+ splitId,
+ new SchemalessSnapshotSplit(tableId, splitId,
splitKeyType, null, null, null));
+ }
+
+ SnapshotPendingSplitsState state =
+ new SnapshotPendingSplitsState(
+ Collections.singletonList(tableId),
+ Collections.emptyList(),
+ assignedSplits,
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ AssignerStatus.INITIAL_ASSIGNING_FINISHED,
+ Collections.emptyList(),
+ false,
+ true,
+ Collections.emptyMap(),
+ ChunkSplitterState.NO_SPLITTING_TABLE_STATE);
+
+ assertThat(new
ArrayList<>(state.getAssignedSplits().keySet())).isEqualTo(assignmentOrder);
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/meta/split/StreamSplitTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/meta/split/StreamSplitTest.java
new file mode 100644
index 000000000..116a6b90c
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/meta/split/StreamSplitTest.java
@@ -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.
+ */
+
+package org.apache.flink.cdc.connectors.base.source.meta.split;
+
+import org.apache.flink.cdc.connectors.base.source.meta.offset.OffsetFactory;
+
+import io.debezium.relational.TableId;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+
+/** Tests for {@link StreamSplit}. */
+class StreamSplitTest {
+
+ @Test
+ void testDuplicateFinishedSnapshotSplitInfosAreRejected() {
+ FinishedSnapshotSplitInfo info =
+ new FinishedSnapshotSplitInfo(
+ new TableId("catalog", "schema", "table"),
+ "split",
+ null,
+ null,
+ null,
+ mock(OffsetFactory.class));
+ List<FinishedSnapshotSplitInfo> infos = new ArrayList<>();
+ infos.add(info);
+ infos.add(info);
+
+ assertThatThrownBy(
+ () ->
+ new StreamSplit(
+ "stream-split",
+ null,
+ null,
+ infos,
+ Collections.emptyMap(),
+ 0,
+ false,
+ false))
+ .isExactlyInstanceOf(IllegalArgumentException.class);
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceReaderTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceReaderTest.java
index c9ea456fe..fbd4871ba 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceReaderTest.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceReaderTest.java
@@ -175,6 +175,62 @@ class IncrementalSourceReaderTest {
reader.close();
}
+ @Test
+ void testReaderDiscardsAlreadyReceivedPrefixOfPartiallyReceivedMetaGroup()
throws Exception {
+ // A reader restored in the middle of meta group synchronization holds
a partial last
+ // group. With a meta group size of 10, a reader holding 13 split
infos requests group 1
+ // (infos 10..19) and must discard the 3 infos it already holds. Split
ids are chosen so
+ // that their lexicographic order (":0, :1, :10, :11, :12, :2, ...")
differs from the
+ // assignment order.
+ TestingReaderContext readerContext = new TestingReaderContext();
+ TestingIncrementalSourceReader reader = createReader(readerContext);
+ reader.start();
+
+ List<FinishedSnapshotSplitInfo> receivedInfos = new ArrayList<>();
+ for (int i = 0; i < 13; i++) {
+ receivedInfos.add(finishedSplitInfo(i));
+ }
+ StreamSplit partialSplit =
+ new StreamSplit(
+ STREAM_SPLIT_ID,
+ new TestOffset(100L),
+ null,
+ receivedInfos,
+ new HashMap<>(),
+ 21);
+ reader.addSplits(Collections.singletonList(partialSplit));
+
+ List<byte[]> group1 = new ArrayList<>();
+ for (int i = 10; i < 20; i++) {
+ group1.add(finishedSplitInfo(i).serialize());
+ }
+ reader.handleSourceEvents(
+ new StreamSplitMetaEvent(partialSplit.splitId(), 1, group1,
21, 1));
+
+ // appendFinishedSplitInfos does not keep the order of the infos, so
only check that every
+ // split info is held exactly once
+ List<String> expectedSplitIds = new ArrayList<>();
+ for (int i = 0; i < 20; i++) {
+ expectedSplitIds.add(TABLE_ID + ":" + i);
+ }
+ assertThat(reader.getUncompletedStreamSplit(partialSplit.splitId()))
+ .extracting(StreamSplit::getFinishedSnapshotSplitInfos)
+ .asList()
+ .extracting(info -> ((FinishedSnapshotSplitInfo)
info).getSplitId())
+ .containsExactlyInAnyOrderElementsOf(expectedSplitIds);
+ reader.close();
+ }
+
+ private static FinishedSnapshotSplitInfo finishedSplitInfo(int chunkId) {
+ return new FinishedSnapshotSplitInfo(
+ TABLE_ID,
+ TABLE_ID + ":" + chunkId,
+ new Object[] {chunkId},
+ new Object[] {chunkId + 1},
+ new TestOffset(50L),
+ TEST_OFFSET_FACTORY);
+ }
+
private static TestingIncrementalSourceReader
createReader(TestingReaderContext readerContext) {
return new TestingIncrementalSourceReader(
new TrackingRecordEmitter(new
SourceReaderMetrics(readerContext.metricGroup())),
@@ -211,6 +267,10 @@ class IncrementalSourceReaderTest {
private SourceSplitState initialize(SourceSplitBase split) {
return initializedState(split);
}
+
+ private StreamSplit getUncompletedStreamSplit(String splitId) {
+ return uncompletedStreamSplits.get(splitId);
+ }
}
private static class TrackingRecordEmitter extends
IncrementalSourceRecordEmitter<Object> {
