This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 c355b2f719 [Fix][Connector-CDC] Reassign restored splits to waiting
readers after checkpoint recovery (#11677)
c355b2f719 is described below
commit c355b2f719c35870c5545899a637ae2ba040036a
Author: Li Zhi <[email protected]>
AuthorDate: Sat Oct 17 07:23:22 2026 +0000
[Fix][Connector-CDC] Reassign restored splits to waiting readers after
checkpoint recovery (#11677)
---
.../api/source/SourceSplitEnumerator.java | 11 +-
.../enumerator/IncrementalSourceEnumerator.java | 27 +++-
.../IncrementalSourceEnumeratorTest.java | 162 +++++++++++++++++++++
.../cdc/mysql/AbstractMysqlCDCITBase.java | 151 +++++++++++++++++++
...sqlcdc_to_mysql_with_sink_failure_recovery.conf | 65 +++++++++
.../server/task/SourceSplitEnumeratorTask.java | 7 +
6 files changed, 418 insertions(+), 5 deletions(-)
diff --git
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/source/SourceSplitEnumerator.java
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/source/SourceSplitEnumerator.java
index d1488d00db..7cf5e16cbe 100644
---
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/source/SourceSplitEnumerator.java
+++
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/source/SourceSplitEnumerator.java
@@ -39,9 +39,9 @@ public interface SourceSplitEnumerator<SplitT extends
SourceSplit, StateT>
void open();
/**
- * Executes engine setup steps in a fixed, non‑concurrent sequence.
+ * Executes engine setup steps.
*
- * <p>Before the first {@link #run()} invocation, methods are called in
this order:
+ * <p>Methods are typically called in this order before the first {@link
#run()} invocation:
*
* <ol>
* <li>{@link #open()}
@@ -49,8 +49,11 @@ public interface SourceSplitEnumerator<SplitT extends
SourceSplit, StateT>
* <li>{@link #registerReader(int)}
* </ol>
*
- * <p>{@implNote The engine guarantees this invocation order and ensures
there are no
- * concurrency issues between these calls.}
+ * <p>The engine only guarantees that {@link #run()} is invoked after
{@link #open()} and after
+ * every reader has been registered. It does not guarantee the order
listed above: restored
+ * splits may be returned via {@link #addSplitsBack(List, int)} at any
point of the source
+ * lifecycle, for example after a checkpoint restore, before or after
{@link #run()}. How such
+ * late returned splits are dispatched to readers is left to the
implementation.
*/
void run() throws Exception;
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/IncrementalSourceEnumerator.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/IncrementalSourceEnumerator.java
index 87ca7dc507..48ab91a096 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/IncrementalSourceEnumerator.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/IncrementalSourceEnumerator.java
@@ -38,6 +38,13 @@ import java.util.stream.Collectors;
/**
* Incremental source enumerator that enumerates receive the split request and
assign the split to
* source readers.
+ *
+ * <p>This enumerator tolerates out-of-order or late {@link
#addSplitsBack(List, int)} calls, e.g.
+ * checkpoint-restored splits returned after a reader already requested a
split: returned splits are
+ * always queued in the {@link SplitAssigner}, and they are dispatched to
waiting readers either on
+ * the first {@link #run()} assignment pass (when the enumerator was not
running yet) or on the next
+ * assignment cycle triggered by {@link #addSplitsBack(List, int)} (when the
enumerator is already
+ * running).
*/
public class IncrementalSourceEnumerator
implements SourceSplitEnumerator<SourceSplitBase, PendingSplitsState> {
@@ -67,6 +74,13 @@ public class IncrementalSourceEnumerator
@Override
public synchronized void run() throws Exception {
this.running = true;
+ // Dispatch the splits that addSplitsBack() returned before run()
executed: the engine
+ // hands the checkpoint-restored splits of a restarting reader to the
enumerator and only
+ // invokes run() once every reader has registered, so in a restart the
restored splits are
+ // normally queued in the assigner while running is still false and
addSplitsBack() skipped
+ // its assignment pass. They are not lost: any split request that
arrived in the meantime
+ // is already queued in readersAwaitingSplit by handleSplitRequest(),
so this first
+ // assignment pass hands out the returned splits to those waiting
readers.
assignSplits();
}
@@ -84,9 +98,20 @@ public class IncrementalSourceEnumerator
}
@Override
- public void addSplitsBack(List<SourceSplitBase> splits, int subtaskId) {
+ public synchronized void addSplitsBack(List<SourceSplitBase> splits, int
subtaskId) {
LOG.debug("Incremental Source Enumerator adds splits back: {}",
splits);
splitAssigner.addSplits(splits);
+ // running == false: the engine delivers the checkpoint-restored
splits of a restarting
+ // reader before it invokes run() (run() waits for every reader to
register), so the
+ // returned splits are still queued in the assigner and get dispatched
by run()'s first
+ // assignment pass - skipping the assignment here does not lose them.
+ // running == true: a restored split may arrive after the reader
already sent its split
+ // request and is parked in readersAwaitingSplit; re-run the
assignment loop so the
+ // waiting reader receives the restored split immediately instead of
stalling until
+ // another trigger.
+ if (running) {
+ assignSplits();
+ }
}
@Override
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/IncrementalSourceEnumeratorTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/IncrementalSourceEnumeratorTest.java
new file mode 100644
index 0000000000..86c5e7269a
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/enumerator/IncrementalSourceEnumeratorTest.java
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.cdc.base.source.enumerator;
+
+import org.apache.seatunnel.api.common.metrics.MetricsContext;
+import org.apache.seatunnel.api.event.EventListener;
+import org.apache.seatunnel.api.source.SourceEvent;
+import org.apache.seatunnel.api.source.SourceSplitEnumerator;
+import
org.apache.seatunnel.connectors.cdc.base.source.enumerator.state.PendingSplitsState;
+import
org.apache.seatunnel.connectors.cdc.base.source.event.SnapshotSplitWatermark;
+import org.apache.seatunnel.connectors.cdc.base.source.split.SnapshotSplit;
+import org.apache.seatunnel.connectors.cdc.base.source.split.SourceSplitBase;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+
+public class IncrementalSourceEnumeratorTest {
+
+ @Test
+ public void shouldAssignRestoredSplitsToWaitingReader() throws Exception {
+ TestingEnumeratorContext context = new TestingEnumeratorContext();
+ TestingSplitAssigner splitAssigner = new TestingSplitAssigner();
+ SourceSplitBase restoredSplit = new SnapshotSplit("restored", null,
null, null, null);
+
+ IncrementalSourceEnumerator enumerator =
+ new IncrementalSourceEnumerator(context, splitAssigner);
+ enumerator.open();
+ enumerator.run();
+ enumerator.handleSplitRequest(0);
+ enumerator.addSplitsBack(Collections.singletonList(restoredSplit), 0);
+
+ Assertions.assertEquals(
+ Collections.singletonList(restoredSplit),
splitAssigner.restoredSplits);
+ Assertions.assertEquals(Collections.singletonList(restoredSplit),
context.assignedSplits);
+ }
+
+ @Test
+ public void shouldAssignSplitsAddedBackBeforeRunExactlyOnce() throws
Exception {
+ TestingEnumeratorContext context = new TestingEnumeratorContext();
+ TestingSplitAssigner splitAssigner = new TestingSplitAssigner();
+ SourceSplitBase restoredSplit = new SnapshotSplit("restored", null,
null, null, null);
+
+ IncrementalSourceEnumerator enumerator =
+ new IncrementalSourceEnumerator(context, splitAssigner);
+ enumerator.open();
+ // Mirror the engine's restore ordering: the reader's split request
and the restored
+ // splits are delivered before the first run() invocation. Nothing is
dispatched yet
+ // because the enumerator is not running.
+ enumerator.handleSplitRequest(0);
+ enumerator.addSplitsBack(Collections.singletonList(restoredSplit), 0);
+ Assertions.assertEquals(Collections.emptyList(),
context.allAssignedSplits);
+
+ enumerator.run();
+
+ Assertions.assertEquals(
+ Collections.singletonList(restoredSplit),
splitAssigner.restoredSplits);
+ Assertions.assertEquals(Collections.singletonList(restoredSplit),
context.assignedSplits);
+ Assertions.assertEquals(
+ Collections.singletonList(restoredSplit),
context.allAssignedSplits);
+ }
+
+ private static final class TestingSplitAssigner implements SplitAssigner {
+ private List<SourceSplitBase> addedSplits = Collections.emptyList();
+ private List<SourceSplitBase> restoredSplits = Collections.emptyList();
+
+ @Override
+ public void open() {}
+
+ @Override
+ public Optional<SourceSplitBase> getNext() {
+ if (addedSplits.isEmpty()) {
+ return Optional.empty();
+ }
+ SourceSplitBase split = addedSplits.get(0);
+ addedSplits = Collections.emptyList();
+ return Optional.of(split);
+ }
+
+ @Override
+ public boolean waitingForCompletedSplits() {
+ return true;
+ }
+
+ @Override
+ public void onCompletedSplits(
+ List<SnapshotSplitWatermark> completedSnapshotSplitWatermarks)
{}
+
+ @Override
+ public void addSplits(Collection<SourceSplitBase> splits) {
+ restoredSplits = Collections.unmodifiableList(new
ArrayList<>(splits));
+ addedSplits = restoredSplits;
+ }
+
+ @Override
+ public PendingSplitsState snapshotState(long checkpointId) {
+ return null;
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) {}
+ }
+
+ private static final class TestingEnumeratorContext
+ implements SourceSplitEnumerator.Context<SourceSplitBase> {
+ private List<SourceSplitBase> assignedSplits = Collections.emptyList();
+ private final List<SourceSplitBase> allAssignedSplits = new
ArrayList<>();
+
+ @Override
+ public int currentParallelism() {
+ return 1;
+ }
+
+ @Override
+ public Set<Integer> registeredReaders() {
+ return Collections.singleton(0);
+ }
+
+ @Override
+ public void assignSplit(int subtaskId, List<SourceSplitBase> splits) {
+ assignedSplits = splits;
+ allAssignedSplits.addAll(splits);
+ }
+
+ @Override
+ public void signalNoMoreSplits(int subtask) {}
+
+ @Override
+ public void sendEventToSourceReader(int subtaskId, SourceEvent event)
{}
+
+ @Override
+ public MetricsContext getMetricsContext() {
+ return null;
+ }
+
+ @Override
+ public EventListener getEventListener() {
+ return null;
+ }
+ }
+}
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 55fa1d9662..00c2c066ac 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
@@ -179,6 +179,157 @@ public abstract class AbstractMysqlCDCITBase extends
TestSuiteBase implements Te
});
}
+ @TestTemplate
+ @DisabledOnContainer(
+ value = {},
+ type = {EngineType.SPARK, EngineType.FLINK},
+ disabledReason =
+ "This case verifies Zeta automatic checkpoint recovery
after a sink failure.")
+ public void testMysqlCdcContinuesAfterSinkFailure(TestContainer container)
throws Exception {
+ // Regression guard for restored-split reassignment after a sink
failure. With
+ // parallelism = 1 (see the conf), this clean full-pipeline restart
drives the
+ // pre-existing handleSplitRequest() dispatch path: the restored split
lands in the
+ // assigner while the enumerator is not yet running, and the reader
picks it up via
+ // its requestSplit() -> handleSplitRequest() call. The new running ==
true branch in
+ // IncrementalSourceEnumerator.addSplitsBack() is covered
deterministically by the unit
+ // test
IncrementalSourceEnumeratorTest#shouldAssignRestoredSplitsToWaitingReader; this
+ // e2e proves the end-to-end recovery path through the real engine.
+ clearTable(MYSQL_DATABASE, SOURCE_TABLE_1);
+ clearTable(MYSQL_DATABASE, SINK_TABLE);
+ Long jobId = JobIdGenerator.newJobId();
+ upsertDeleteSourceTable(MYSQL_DATABASE, SOURCE_TABLE_1);
+ CompletableFuture<Container.ExecResult> executeJobFuture =
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return container.executeJob(
+
"/mysqlcdc_to_mysql_with_sink_failure_recovery.conf",
+ String.valueOf(jobId));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ await().atMost(60, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+ "RUNNING",
container.getJobStatus(String.valueOf(jobId))));
+
+ await().atMost(60, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertIterableEquals(
+
query(getSourceQuerySQL(MYSQL_DATABASE, SOURCE_TABLE_1)),
+ query(getSinkQuerySQL(MYSQL_DATABASE,
SINK_TABLE))));
+
+ // The test config checkpoints every five seconds. Wait until the
snapshot state is
+ // durably checkpointed before failing the incremental CDC stream.
+ await().atMost(2, TimeUnit.MINUTES)
+ .untilAsserted(
+ () ->
+ Assertions.assertTrue(
+
container.getCompletedCheckpointCount(String.valueOf(jobId))
+ > 0));
+ String triggerName = "fail_cdc_sink_once";
+ executeSql("DROP TRIGGER IF EXISTS " + MYSQL_DATABASE + "." +
triggerName);
+ executeSql(
+ "CREATE TRIGGER "
+ + MYSQL_DATABASE
+ + "."
+ + triggerName
+ + " BEFORE INSERT ON "
+ + MYSQL_DATABASE
+ + "."
+ + SINK_TABLE
+ + " FOR EACH ROW SIGNAL SQLSTATE '45000' SET
MESSAGE_TEXT = 'injected sink failure'");
+ try {
+ int logOffset = container.getServerLogs().length();
+ executeSql(
+ "UPDATE "
+ + MYSQL_DATABASE
+ + "."
+ + SOURCE_TABLE_1
+ + " SET f_bigint = 20000 WHERE id = 5");
+ // The injected trigger makes the JDBC sink's insert fail, so the
engine detects
+ // the sink failure and enters the restart cycle
(job.retry.interval.seconds = 15).
+ // Wait for a positive signal in the engine log that the injected
failure actually
+ // fired instead of a fixed sleep, so the test cannot silently
pass without
+ // exercising the recovery path. The job status cannot be used
here because on dev
+ // it stays RUNNING throughout the restart window. Once the
failure is observed the
+ // finally block removes the trigger, so the post-restart replay
of the
+ // f_bigint=20000 binlog event normally does not hit the trigger a
second time. If
+ // the restart still wins that race, the replay fails once more
and consumes one of
+ // the extra retries granted by job.retry.times in the conf; the
trigger is gone by
+ // the next cycle, so the job recovers anyway. The two-minute
budget matches the
+ // other waits in this test.
+ await().atMost(2, TimeUnit.MINUTES)
+ .untilAsserted(
+ () -> {
+ String serverLogs = container.getServerLogs();
+ // The engine log can be truncated or rotated
while the job
+ // restarts; clamp the offset so a shorter log
is treated as
+ // "no match yet" instead of throwing
+ // StringIndexOutOfBoundsException and
aborting the wait.
+ Assertions.assertTrue(
+ serverLogs
+ .substring(Math.min(logOffset,
serverLogs.length()))
+ .contains("injected sink
failure"),
+ "injected sink failure was not
observed in the "
+ + "engine logs after the
f_bigint=20000 update");
+ });
+ } finally {
+ executeSql("DROP TRIGGER IF EXISTS " + MYSQL_DATABASE + "." +
triggerName);
+ }
+
+ // The job restarts from the last checkpoint and replays the binlog
events written
+ // after it (including the f_bigint=20000 update). The trigger must be
gone before
+ // that replay reaches the sink, otherwise the restored job fails
again. Verify the
+ // trigger was actually dropped, then confirm the job is still
RUNNING; this status
+ // check only confirms that the job is alive during recovery. The
source-to-sink
+ // equality assertion below is the actual signal that recovery has
converged.
+ await().atMost(30, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+ 0L,
+ ((Number)
+ query(
+
"SELECT COUNT(*) FROM information_schema.triggers WHERE trigger_schema = '"
+
+ MYSQL_DATABASE
+
+ "' AND trigger_name = '"
+
+ triggerName
+
+ "'")
+ .get(0)
+ .get(0))
+ .longValue()));
+ await().atMost(2, TimeUnit.MINUTES)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+ "RUNNING",
container.getJobStatus(String.valueOf(jobId))));
+
+ executeSql(
+ "UPDATE "
+ + MYSQL_DATABASE
+ + "."
+ + SOURCE_TABLE_1
+ + " SET f_bigint = 30000 WHERE id = 5");
+ await().atMost(2, TimeUnit.MINUTES)
+ .untilAsserted(
+ () ->
+ Assertions.assertIterableEquals(
+
query(getSourceQuerySQL(MYSQL_DATABASE, SOURCE_TABLE_1)),
+ query(getSinkQuerySQL(MYSQL_DATABASE,
SINK_TABLE))));
+ Assertions.assertEquals("RUNNING",
container.getJobStatus(String.valueOf(jobId)));
+ Assertions.assertEquals(0,
container.cancelJob(String.valueOf(jobId)).getExitCode());
+ await().atMost(60, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+ "CANCELED",
container.getJobStatus(String.valueOf(jobId))));
+ Assertions.assertEquals(0, executeJobFuture.get(60,
TimeUnit.SECONDS).getExitCode());
+ }
+
@TestTemplate
@DisabledOnContainer(
value = {},
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_sink_failure_recovery.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_sink_failure_recovery.conf
new file mode 100644
index 0000000000..626f7f52f4
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_sink_failure_recovery.conf
@@ -0,0 +1,65 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 5000
+ # The test drops the injected-failure trigger as soon as the failure shows
up in the engine
+ # log. If a restart replays the failing binlog event before the trigger is
gone, that replay
+ # fails once more, so keep the default retry budget instead of a single
retry to give the
+ # job room to recover on the next restart cycle.
+ job.retry.times = 3
+ job.retry.interval.seconds = 15
+}
+
+source {
+ MySQL-CDC {
+ plugin_output = "customers_mysql_cdc"
+ server-id = 5652
+ username = "st_user_source"
+ password = "mysqlpw"
+ table-names = ["mysql_cdc.mysql_cdc_e2e_source_table"]
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc"
+ }
+}
+
+transform {
+ sql {
+ plugin_input = "customers_mysql_cdc"
+ query = """ select id, f_binary, f_blob, f_long_varbinary, f_longblob,
f_tinyblob, f_varbinary, f_smallint, f_smallint_unsigned, f_mediumint,
+ f_mediumint_unsigned, f_int, f_int_unsigned, f_integer,
f_integer_unsigned, f_bigint, f_bigint_unsigned, f_numeric, f_decimal,
+ f_float, f_double, f_double_precision, f_longtext,
f_mediumtext, f_text, f_tinytext, f_varchar, f_date, f_datetime, f_timestamp,
+ f_bit1, f_bit64, f_char, f_enum, f_mediumblob, f_long_varchar,
f_real, f_time, f_tinyint, f_tinyint_unsigned, f_json, f_year
+ from dual """
+ plugin_output = "trans_mysql_cdc"
+ }
+}
+
+sink {
+ jdbc {
+ plugin_input = "trans_mysql_cdc"
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/mysql_cdc"
+ driver = "com.mysql.cj.jdbc.Driver"
+ user = "st_user_sink"
+ password = "mysqlpw"
+ generate_sink_sql = true
+ database = mysql_cdc
+ table = mysql_cdc_e2e_sink_table
+ primary_keys = ["id"]
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SourceSplitEnumeratorTask.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SourceSplitEnumeratorTask.java
index 18e1e8217d..02340b48ec 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SourceSplitEnumeratorTask.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SourceSplitEnumeratorTask.java
@@ -238,6 +238,13 @@ public class SourceSplitEnumeratorTask<SplitT extends
SourceSplit> extends Coord
}
int taskSize = taskMemberMapping.size();
if (maxReaderSize == taskSize) {
+ // Cross-class invariant: a reader's restoreState() synchronously
delivers its
+ // checkpoint-restored splits to the enumerator
(RestoredSplitOperation) before it
+ // registers, and run() only fires after every reader has
registered below. So in a
+ // clean full-pipeline restart the restored splits normally reach
the assigner while
+ // the enumerator is not running yet. That ordering is not
guaranteed (multi-reader
+ // interleaving, retry-induced delay), so
IncrementalSourceEnumerator.addSplitsBack()
+ // defensively re-runs assignment when the enumerator is already
running.
readerRegisterComplete = true;
log.debug(String.format("reader register complete, current task
size %d", taskSize));
} else {
