This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-11677-787380c35a2affb8a582f6abee0846a1e85a02ac in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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 {
