This is an automated email from the ASF dual-hosted git repository.
ruanhang1993 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 2ae8f874f [FLINK-40741][cdc-connector][mysql] Fix newly added table
recovery after region failover (#4557)
2ae8f874f is described below
commit 2ae8f874f2ccb57b6ee882c0b75655f1f795b479
Author: Hang Ruan <[email protected]>
AuthorDate: Wed Sep 23 11:51:05 2026 +0800
[FLINK-40741][cdc-connector][mysql] Fix newly added table recovery after
region failover (#4557)
---
.../source/enumerator/MySqlSourceEnumerator.java | 25 +++-
.../mysql/source/reader/MySqlSourceReader.java | 6 +
.../enumerator/MySqlSourceEnumeratorTest.java | 136 +++++++++++++++++++++
.../mysql/source/reader/MySqlSourceReaderTest.java | 28 +++++
4 files changed, 191 insertions(+), 4 deletions(-)
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/enumerator/MySqlSourceEnumerator.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/enumerator/MySqlSourceEnumerator.java
index ff0279e4a..8696877f8 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/enumerator/MySqlSourceEnumerator.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/enumerator/MySqlSourceEnumerator.java
@@ -60,6 +60,7 @@ import java.util.Optional;
import java.util.TreeSet;
import java.util.stream.Collectors;
+import static
org.apache.flink.cdc.connectors.mysql.source.assigners.AssignerStatus.isNewlyAddedAssigningFinished;
import static
org.apache.flink.cdc.connectors.mysql.source.assigners.AssignerStatus.isNewlyAddedAssigningSnapshotFinished;
/**
@@ -146,9 +147,17 @@ public class MySqlSourceEnumerator implements
SplitEnumerator<MySqlSplit, Pendin
@Override
public void addReader(int subtaskId) {
- // send BinlogSplitUpdateRequestEvent to source reader after newly
added table
- // snapshot splits finished.
- if
(isNewlyAddedAssigningSnapshotFinished(splitAssigner.getAssignerStatus())) {
+ // Reconcile the binlog split after newly added table snapshot splits
finished. The
+ // NEWLY_ADDED_ASSIGNING_FINISHED branch is needed for region
failover: the enumerator
+ // remains alive with the latest finished split metadata, while the
restarted reader may
+ // restore an older binlog split from the last completed checkpoint.
+ if
(isNewlyAddedAssigningSnapshotFinished(splitAssigner.getAssignerStatus())
+ ||
isNewlyAddedAssigningFinished(splitAssigner.getAssignerStatus())) {
+ LOG.info(
+ "The enumerator sends BinlogSplitUpdateRequestEvent to
subtask {} when adding "
+ + "reader under assigner status {}.",
+ subtaskId,
+ splitAssigner.getAssignerStatus());
context.sendEventToSourceReader(subtaskId, new
BinlogSplitUpdateRequestEvent());
}
}
@@ -181,7 +190,15 @@ public class MySqlSourceEnumerator implements
SplitEnumerator<MySqlSplit, Pendin
LOG.info(
"The enumerator receives event that the binlog split has
been updated from subtask {}. ",
subtaskId);
- splitAssigner.onBinlogSplitUpdated();
+ if
(isNewlyAddedAssigningFinished(splitAssigner.getAssignerStatus())) {
+ LOG.info(
+ "Ignore the duplicated binlog split update
acknowledgement from subtask {} "
+ + "under assigner status {}.",
+ subtaskId,
+ splitAssigner.getAssignerStatus());
+ } else {
+ splitAssigner.onBinlogSplitUpdated();
+ }
} else if (sourceEvent instanceof
LatestFinishedSplitsNumberRequestEvent) {
LOG.info(
"The enumerator receives request from subtask {} for the
latest finished splits number after added newly tables. ",
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlSourceReader.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlSourceReader.java
index 4d6f4a862..48207aace 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlSourceReader.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlSourceReader.java
@@ -287,6 +287,12 @@ public class MySqlSourceReader<T>
// the binlog split is suspended
if (binlogSplit.isSuspended()) {
suspendedBinlogSplit = binlogSplit;
+ // A suspended split can be restored directly from a
checkpoint after region
+ // failover. In that case there is no running binlog
reader to finish and send
+ // this request from onSplitFinished(), so request the
latest finished split
+ // count here to resume the update protocol.
+ context.sendSourceEventToCoordinator(
+ new LatestFinishedSplitsNumberRequestEvent());
} else if (!binlogSplit.isCompletedSplit()) {
uncompletedBinlogSplits.put(binlogSplit.splitId(),
binlogSplit);
requestBinlogSplitMetaIfNeeded(binlogSplit);
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/enumerator/MySqlSourceEnumeratorTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/enumerator/MySqlSourceEnumeratorTest.java
index 7fb220664..232497896 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/enumerator/MySqlSourceEnumeratorTest.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/enumerator/MySqlSourceEnumeratorTest.java
@@ -18,16 +18,31 @@
package org.apache.flink.cdc.connectors.mysql.source.enumerator;
import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.SourceEvent;
+import org.apache.flink.cdc.connectors.mysql.source.assigners.AssignerStatus;
import
org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlBinlogSplitAssigner;
+import
org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlSplitAssigner;
+import
org.apache.flink.cdc.connectors.mysql.source.assigners.state.PendingSplitsState;
import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfig;
import
org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfigFactory;
import
org.apache.flink.cdc.connectors.mysql.source.events.BinlogSplitMetaAssembledEvent;
+import
org.apache.flink.cdc.connectors.mysql.source.events.BinlogSplitUpdateAckEvent;
+import
org.apache.flink.cdc.connectors.mysql.source.events.BinlogSplitUpdateRequestEvent;
+import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffset;
+import
org.apache.flink.cdc.connectors.mysql.source.split.FinishedSnapshotSplitInfo;
+import org.apache.flink.cdc.connectors.mysql.source.split.MySqlSplit;
+import
org.apache.flink.cdc.connectors.mysql.source.utils.MockMySqlSplitEnumeratorEnumeratorContext;
import org.apache.flink.cdc.connectors.mysql.table.StartupOptions;
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.Test;
import java.time.ZoneId;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
import static
org.apache.flink.cdc.connectors.mysql.testutils.MetricsUtils.getMySqlSplitEnumeratorContext;
@@ -63,4 +78,125 @@ class MySqlSourceEnumeratorTest {
0, new
BinlogSplitMetaAssembledEvent(0L)))
.doesNotThrowAnyException();
}
+
+ @Test
+ void testAddReaderRequestsBinlogSplitUpdateAfterRegionFailover() throws
Exception {
+ MockMySqlSplitEnumeratorEnumeratorContext context =
+ new MockMySqlSplitEnumeratorEnumeratorContext(1);
+ TestingSplitAssigner splitAssigner =
+ new
TestingSplitAssigner(AssignerStatus.NEWLY_ADDED_ASSIGNING_FINISHED);
+ MySqlSourceEnumerator enumerator =
+ new MySqlSourceEnumerator(
+ context, null, splitAssigner,
Boundedness.CONTINUOUS_UNBOUNDED);
+
+ enumerator.addReader(0);
+
+ List<SourceEvent> sentEvents = context.getSentSourceEvent().get(0);
+ Assertions.assertThat(sentEvents)
+ .singleElement()
+ .isInstanceOf(BinlogSplitUpdateRequestEvent.class);
+ }
+
+ @Test
+ void testBinlogSplitUpdateAckIsIdempotentAfterRegionFailover() {
+ MockMySqlSplitEnumeratorEnumeratorContext context =
+ new MockMySqlSplitEnumeratorEnumeratorContext(1);
+ TestingSplitAssigner splitAssigner =
+ new
TestingSplitAssigner(AssignerStatus.NEWLY_ADDED_ASSIGNING_FINISHED);
+ MySqlSourceEnumerator enumerator =
+ new MySqlSourceEnumerator(
+ context, null, splitAssigner,
Boundedness.CONTINUOUS_UNBOUNDED);
+
+ enumerator.handleSourceEvent(0, new BinlogSplitUpdateAckEvent());
+
+
Assertions.assertThat(splitAssigner.getBinlogSplitUpdatedCount()).isZero();
+ Assertions.assertThat(splitAssigner.getAssignerStatus())
+ .isEqualTo(AssignerStatus.NEWLY_ADDED_ASSIGNING_FINISHED);
+ }
+
+ @Test
+ void testBinlogSplitUpdateAckCompletesNormalUpdate() {
+ MockMySqlSplitEnumeratorEnumeratorContext context =
+ new MockMySqlSplitEnumeratorEnumeratorContext(1);
+ TestingSplitAssigner splitAssigner =
+ new
TestingSplitAssigner(AssignerStatus.NEWLY_ADDED_ASSIGNING_SNAPSHOT_FINISHED);
+ MySqlSourceEnumerator enumerator =
+ new MySqlSourceEnumerator(
+ context, null, splitAssigner,
Boundedness.CONTINUOUS_UNBOUNDED);
+
+ enumerator.handleSourceEvent(0, new BinlogSplitUpdateAckEvent());
+
+
Assertions.assertThat(splitAssigner.getBinlogSplitUpdatedCount()).isOne();
+ Assertions.assertThat(splitAssigner.getAssignerStatus())
+ .isEqualTo(AssignerStatus.NEWLY_ADDED_ASSIGNING_FINISHED);
+ }
+
+ private static class TestingSplitAssigner implements MySqlSplitAssigner {
+ private AssignerStatus assignerStatus;
+ private int binlogSplitUpdatedCount;
+
+ private TestingSplitAssigner(AssignerStatus assignerStatus) {
+ this.assignerStatus = assignerStatus;
+ }
+
+ @Override
+ public void open() {}
+
+ @Override
+ public Optional<MySqlSplit> getNext() {
+ return Optional.empty();
+ }
+
+ @Override
+ public boolean waitingForFinishedSplits() {
+ return false;
+ }
+
+ @Override
+ public List<FinishedSnapshotSplitInfo> getFinishedSplitInfos() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public void onFinishedSplits(Map<String, BinlogOffset>
splitFinishedOffsets) {}
+
+ @Override
+ public void addSplits(Collection<MySqlSplit> splits) {}
+
+ @Override
+ public PendingSplitsState snapshotState(long checkpointId) {
+ return null;
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) {}
+
+ @Override
+ public AssignerStatus getAssignerStatus() {
+ return assignerStatus;
+ }
+
+ @Override
+ public void startAssignNewlyAddedTables() {
+ assignerStatus = assignerStatus.startAssignNewlyTables();
+ }
+
+ @Override
+ public boolean noMoreSplits() {
+ return true;
+ }
+
+ @Override
+ public void onBinlogSplitUpdated() {
+ binlogSplitUpdatedCount++;
+ assignerStatus = assignerStatus.onBinlogSplitUpdated();
+ }
+
+ @Override
+ public void close() {}
+
+ private int getBinlogSplitUpdatedCount() {
+ return binlogSplitUpdatedCount;
+ }
+ }
}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlSourceReaderTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlSourceReaderTest.java
index 137df37c5..93d8a42f0 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlSourceReaderTest.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlSourceReaderTest.java
@@ -31,6 +31,7 @@ import
org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfigFact
import
org.apache.flink.cdc.connectors.mysql.source.events.BinlogSplitMetaAssembledEvent;
import
org.apache.flink.cdc.connectors.mysql.source.events.BinlogSplitMetaEvent;
import
org.apache.flink.cdc.connectors.mysql.source.events.FinishedSnapshotSplitsAckEvent;
+import
org.apache.flink.cdc.connectors.mysql.source.events.LatestFinishedSplitsNumberRequestEvent;
import
org.apache.flink.cdc.connectors.mysql.source.metrics.MySqlSourceReaderMetrics;
import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffset;
import
org.apache.flink.cdc.connectors.mysql.source.split.FinishedSnapshotSplitInfo;
@@ -122,6 +123,33 @@ class MySqlSourceReaderTest extends MySqlSourceTestBase {
inventoryDatabase.dropDatabase();
}
+ @Test
+ void testRestoredSuspendedBinlogSplitRequestsLatestFinishedSplitNumber()
throws Exception {
+ MySqlSourceConfig sourceConfig = getConfig(new String[] {"customers"});
+ TestingReaderContext readerContext = new TestingReaderContext();
+ MySqlSourceReader<SourceRecord> reader =
+ createReader(sourceConfig, readerContext, 0,
SnapshotPhaseHooks.empty());
+ MySqlBinlogSplit binlogSplit =
+ new MySqlBinlogSplit(
+ MySqlBinlogSplitAssigner.BINLOG_SPLIT_ID,
+ BinlogOffset.ofBinlogFilePosition("mysql-bin.000001",
4L),
+ BinlogOffset.ofNonStopping(),
+ Collections.emptyList(),
+ Collections.emptyMap(),
+ 0);
+
+ reader.addSplits(
+
Collections.singletonList(MySqlBinlogSplit.toSuspendedBinlogSplit(binlogSplit)));
+
+ Assertions.assertThat(readerContext.getSentEvents())
+ .anySatisfy(
+ event ->
+ Assertions.assertThat(event)
+ .isInstanceOf(
+
LatestFinishedSplitsNumberRequestEvent.class));
+ reader.close();
+ }
+
@Test
void testRemoveTableUsingStateFromSnapshotPhase() throws Exception {
customerDatabase.createAndInitialize();
