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-12528-283bf863cd32f1603149d03d8eee8fa48c6df65d in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 283bf863cd32f1603149d03d8eee8fa48c6df65d Author: Jast <[email protected]> AuthorDate: Thu Oct 22 15:46:16 2026 +0000 [Fix][CDC] Wait for enumerator ack before clearing finished snapshot splits (#12163) Co-authored-by: zhangshenghang <[email protected]> Co-authored-by: jast <[email protected]> Co-authored-by: Claude Code <[email protected]> Co-authored-by: multica-agent <[email protected]> --- .../source/reader/IncrementalSourceReader.java | 29 ++++++- .../source/reader/IncrementalSourceReaderTest.java | 97 ++++++++++++++++++++++ .../serializable/TaskDataSerializerHook.java | 5 ++ .../engine/server/task/SourceSeaTunnelTask.java | 5 ++ .../context/SeaTunnelSplitEnumeratorContext.java | 24 +++++- .../server/task/flow/SourceFlowLifeCycle.java | 14 ++++ .../source/SourceEnumeratorEventOperation.java | 71 ++++++++++++++++ 7 files changed, 240 insertions(+), 5 deletions(-) diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/reader/IncrementalSourceReader.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/reader/IncrementalSourceReader.java index 98b61593f6..4e577ee7cd 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/reader/IncrementalSourceReader.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/reader/IncrementalSourceReader.java @@ -18,6 +18,7 @@ package org.apache.seatunnel.connectors.cdc.base.source.reader; import org.apache.seatunnel.api.source.Collector; +import org.apache.seatunnel.api.source.SourceEvent; import org.apache.seatunnel.api.source.SourceReader; import org.apache.seatunnel.api.table.catalog.CatalogTable; import org.apache.seatunnel.api.table.catalog.CatalogTableUtil; @@ -28,6 +29,7 @@ import org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceConfig; import org.apache.seatunnel.connectors.cdc.base.config.SourceConfig; import org.apache.seatunnel.connectors.cdc.base.dialect.DataSourceDialect; import org.apache.seatunnel.connectors.cdc.base.source.event.CompletedSnapshotPhaseEvent; +import org.apache.seatunnel.connectors.cdc.base.source.event.CompletedSnapshotSplitsAckEvent; import org.apache.seatunnel.connectors.cdc.base.source.event.CompletedSnapshotSplitsReportEvent; import org.apache.seatunnel.connectors.cdc.base.source.event.SnapshotSplitWatermark; import org.apache.seatunnel.connectors.cdc.base.source.offset.Offset; @@ -51,10 +53,10 @@ import lombok.extern.slf4j.Slf4j; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; -import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Supplier; import java.util.stream.Collectors; @@ -101,7 +103,9 @@ public class IncrementalSourceReader<T, C extends SourceConfig> context); this.dataSourceDialect = dataSourceDialect; this.sourceConfig = sourceConfig; - this.finishedUnackedSplits = new HashMap<>(); + // Ack events are handled on the engine's event-delivery thread while splits finish on the + // reader thread, so the map must tolerate concurrent access. + this.finishedUnackedSplits = new ConcurrentHashMap<>(); this.subtaskId = context.getIndexOfSubtask(); this.debeziumDeserializationSchema = debeziumDeserializationSchema; } @@ -224,8 +228,10 @@ public class IncrementalSourceReader<T, C extends SourceConfig> new CompletedSnapshotSplitsReportEvent(); reportEvent.setCompletedSnapshotSplitWatermarks(completedSnapshotSplitWatermarks); context.sendSourceEventToEnumerator(reportEvent); - // TODO need enumerator return ack - finishedUnackedSplits.clear(); + // The splits stay in finishedUnackedSplits until the enumerator returns a + // CompletedSnapshotSplitsAckEvent via handleSourceEvent. Clearing them eagerly + // here would let a snapshot phase advance without the enumerator ever knowing + // which splits the reader finished. log.debug( "The subtask {} reports offsets of finished snapshot splits {}.", subtaskId, @@ -233,6 +239,21 @@ public class IncrementalSourceReader<T, C extends SourceConfig> } } + @Override + public void handleSourceEvent(SourceEvent sourceEvent) { + if (sourceEvent instanceof CompletedSnapshotSplitsAckEvent) { + CompletedSnapshotSplitsAckEvent ackEvent = + (CompletedSnapshotSplitsAckEvent) sourceEvent; + ackEvent.getCompletedSplits().forEach(finishedUnackedSplits::remove); + log.debug( + "The subtask {} receives acknowledgements for finished snapshot splits {}.", + subtaskId, + ackEvent.getCompletedSplits()); + } else { + super.handleSourceEvent(sourceEvent); + } + } + @Override protected SourceSplitStateBase initializedState(SourceSplitBase split) { if (split.isSnapshotSplit()) { diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/reader/IncrementalSourceReaderTest.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/reader/IncrementalSourceReaderTest.java index c7832751cb..18bebd9213 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/reader/IncrementalSourceReaderTest.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/source/reader/IncrementalSourceReaderTest.java @@ -32,9 +32,13 @@ import org.apache.seatunnel.api.table.type.SeaTunnelRowType; import org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceConfig; import org.apache.seatunnel.connectors.cdc.base.config.SourceConfig; import org.apache.seatunnel.connectors.cdc.base.dialect.DataSourceDialect; +import org.apache.seatunnel.connectors.cdc.base.source.event.CompletedSnapshotSplitsAckEvent; +import org.apache.seatunnel.connectors.cdc.base.source.event.CompletedSnapshotSplitsReportEvent; +import org.apache.seatunnel.connectors.cdc.base.source.event.SnapshotSplitWatermark; import org.apache.seatunnel.connectors.cdc.base.source.offset.Offset; import org.apache.seatunnel.connectors.cdc.base.source.offset.OffsetFactory; import org.apache.seatunnel.connectors.cdc.base.source.split.IncrementalSplit; +import org.apache.seatunnel.connectors.cdc.base.source.split.SnapshotSplit; import org.apache.seatunnel.connectors.cdc.base.source.split.SourceRecords; import org.apache.seatunnel.connectors.cdc.base.source.split.SourceSplitBase; import org.apache.seatunnel.connectors.cdc.base.source.split.state.SourceSplitStateBase; @@ -78,6 +82,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.LockSupport; +import java.util.stream.Collectors; import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -578,6 +583,76 @@ class IncrementalSourceReaderTest { } } + @Test + void testAckRemovesFinishedSplitsFromUnackedState() { + SourceConfig sourceConfig = Mockito.mock(SourceConfig.class); + DataSourceDialect<SourceConfig> dialect = + Mockito.mock(DataSourceDialect.class, Mockito.CALLS_REAL_METHODS); + SourceReader.Context context = Mockito.mock(SourceReader.Context.class); + IncrementalSourceReader<Object, SourceConfig> reader = + createReader(dialect, sourceConfig, context); + + try { + reader.addSplits(Collections.singletonList(finishedSnapshotSplit("snapshot-split-0"))); + + Assertions.assertEquals( + Collections.singletonList("snapshot-split-0"), + splitIds(reader.snapshotState(1L))); + + reader.handleSourceEvent( + new CompletedSnapshotSplitsAckEvent( + Collections.singletonList("snapshot-split-0"))); + + Assertions.assertTrue(reader.snapshotState(2L).isEmpty()); + } finally { + reader.close(); + } + } + + @Test + void testFinishedSplitsAreResentUntilAcknowledged() { + SourceConfig sourceConfig = Mockito.mock(SourceConfig.class); + DataSourceDialect<SourceConfig> dialect = + Mockito.mock(DataSourceDialect.class, Mockito.CALLS_REAL_METHODS); + SourceReader.Context context = Mockito.mock(SourceReader.Context.class); + List<CompletedSnapshotSplitsReportEvent> reports = new ArrayList<>(); + Mockito.doAnswer( + invocation -> { + reports.add(invocation.getArgument(0)); + return null; + }) + .when(context) + .sendSourceEventToEnumerator(Mockito.any()); + IncrementalSourceReader<Object, SourceConfig> reader = + createReader(dialect, sourceConfig, context); + + try { + reader.addSplits( + Arrays.asList( + finishedSnapshotSplit("snapshot-split-0"), + finishedSnapshotSplit("snapshot-split-1"))); + reader.addSplits(Collections.singletonList(finishedSnapshotSplit("snapshot-split-2"))); + + // no ack received yet, so the latest report still carries the full unacked set + Assertions.assertEquals( + Arrays.asList("snapshot-split-0", "snapshot-split-1", "snapshot-split-2"), + reportSplitIds(reports.get(reports.size() - 1))); + + reader.handleSourceEvent( + new CompletedSnapshotSplitsAckEvent( + Arrays.asList( + "snapshot-split-0", "snapshot-split-1", "snapshot-split-2"))); + reader.addSplits(Collections.singletonList(finishedSnapshotSplit("snapshot-split-3"))); + + // previously acked splits are pruned from the report + Assertions.assertEquals( + Collections.singletonList("snapshot-split-3"), + reportSplitIds(reports.get(reports.size() - 1))); + } finally { + reader.close(); + } + } + private static IncrementalSourceReader<Object, SourceConfig> createReader( DataSourceDialect<SourceConfig> dialect, SourceConfig sourceConfig, @@ -640,6 +715,28 @@ class IncrementalSourceReaderTest { historyTableChanges); } + private static SnapshotSplit finishedSnapshotSplit(String splitId) { + return new SnapshotSplit( + splitId, + KEPT_TABLE, + null, + new Object[] {1}, + new Object[] {2}, + Mockito.mock(Offset.class), + Mockito.mock(Offset.class)); + } + + private static List<String> splitIds(List<SourceSplitBase> splits) { + return splits.stream().map(SourceSplitBase::splitId).collect(Collectors.toList()); + } + + private static List<String> reportSplitIds(CompletedSnapshotSplitsReportEvent event) { + return event.getCompletedSnapshotSplitWatermarks().stream() + .map(SnapshotSplitWatermark::getSplitId) + .sorted() + .collect(Collectors.toList()); + } + private static CatalogTable catalogTable(TableId tableId) { TablePath tablePath = TablePath.of(tableId.catalog(), tableId.table()); return CatalogTable.of( diff --git a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/serializable/TaskDataSerializerHook.java b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/serializable/TaskDataSerializerHook.java index da2e185108..228d215f70 100644 --- a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/serializable/TaskDataSerializerHook.java +++ b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/serializable/TaskDataSerializerHook.java @@ -42,6 +42,7 @@ import org.apache.seatunnel.engine.server.task.operation.source.CloseIdleReaderO import org.apache.seatunnel.engine.server.task.operation.source.LastCheckpointNotifyOperation; import org.apache.seatunnel.engine.server.task.operation.source.RequestSplitOperation; import org.apache.seatunnel.engine.server.task.operation.source.RestoredSplitOperation; +import org.apache.seatunnel.engine.server.task.operation.source.SourceEnumeratorEventOperation; import org.apache.seatunnel.engine.server.task.operation.source.SourceNoMoreElementOperation; import org.apache.seatunnel.engine.server.task.operation.source.SourceReaderEventOperation; import org.apache.seatunnel.engine.server.task.operation.source.SourceRegisterOperation; @@ -110,6 +111,8 @@ public class TaskDataSerializerHook implements DataSerializerHook { public static final int REPORT_METRICS_OPERATION = 28; + public static final int SOURCE_ENUMERATOR_EVENT_TYPE = 29; + public static final int FACTORY_ID = FactoryIdHelper.getFactoryId( SeaTunnelFactoryIdConstant.SEATUNNEL_TASK_DATA_SERIALIZER_FACTORY, @@ -170,6 +173,8 @@ public class TaskDataSerializerHook implements DataSerializerHook { return new CleanTaskGroupContextOperation(); case SOURCE_READER_EVENT_OPERATOR: return new SourceReaderEventOperation(); + case SOURCE_ENUMERATOR_EVENT_TYPE: + return new SourceEnumeratorEventOperation(); case CHECK_TASKGROUP_IS_EXECUTING: return new CheckTaskGroupIsExecutingOperation(); case GET_METRICS_OPERATION: diff --git a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SourceSeaTunnelTask.java b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SourceSeaTunnelTask.java index 1c78baf963..8ba6088eec 100644 --- a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SourceSeaTunnelTask.java +++ b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SourceSeaTunnelTask.java @@ -20,6 +20,7 @@ package org.apache.seatunnel.engine.server.task; import org.apache.seatunnel.api.common.metrics.MetricsContext; import org.apache.seatunnel.api.configuration.ReadonlyConfig; import org.apache.seatunnel.api.serialization.Serializer; +import org.apache.seatunnel.api.source.SourceEvent; import org.apache.seatunnel.api.source.SourceSplit; import org.apache.seatunnel.api.table.catalog.CatalogTable; import org.apache.seatunnel.api.table.catalog.CatalogTableUtil; @@ -162,6 +163,10 @@ public class SourceSeaTunnelTask<T, SplitT extends SourceSplit> extends SeaTunne ((SourceFlowLifeCycle<T, SplitT>) startFlowLifeCycle).receivedSplits(splits); } + public void receivedSourceEvent(SourceEvent sourceEvent) { + ((SourceFlowLifeCycle<T, SplitT>) startFlowLifeCycle).receivedSourceEvent(sourceEvent); + } + @Override public void triggerBarrier(Barrier barrier) throws Exception { SourceFlowLifeCycle<T, SplitT> sourceFlow = diff --git a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/context/SeaTunnelSplitEnumeratorContext.java b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/context/SeaTunnelSplitEnumeratorContext.java index b4e3de60fb..03c2442115 100644 --- a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/context/SeaTunnelSplitEnumeratorContext.java +++ b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/context/SeaTunnelSplitEnumeratorContext.java @@ -22,9 +22,12 @@ import org.apache.seatunnel.api.event.EventListener; import org.apache.seatunnel.api.source.SourceEvent; import org.apache.seatunnel.api.source.SourceSplit; import org.apache.seatunnel.api.source.SourceSplitEnumerator; +import org.apache.seatunnel.engine.server.execution.TaskLocation; import org.apache.seatunnel.engine.server.task.SourceSplitEnumeratorTask; import org.apache.seatunnel.engine.server.task.operation.source.AssignSplitOperation; +import org.apache.seatunnel.engine.server.task.operation.source.SourceEnumeratorEventOperation; +import com.hazelcast.cluster.Address; import lombok.extern.slf4j.Slf4j; import java.util.Collections; @@ -102,7 +105,26 @@ public class SeaTunnelSplitEnumeratorContext<SplitT extends SourceSplit> } @Override - public void sendEventToSourceReader(int subtaskId, SourceEvent event) {} + public void sendEventToSourceReader(int subtaskId, SourceEvent event) { + TaskLocation targetLocation = task.getTaskMemberLocationByIndex(subtaskId); + Address targetAddress = task.getTaskMemberAddressByIndex(subtaskId); + if (targetLocation == null || targetAddress == null) { + log.warn( + "No registered reader for subtask {}, skip sending source event {}", + subtaskId, + event); + return; + } + // Fire-and-forget: this can be invoked on a Hazelcast operation thread while dispatching a + // reader event (e.g. a split report), so blocking on the send future may deadlock the + // operation system. A lost event is safe to drop: sources that rely on acknowledgements + // re-send their reports. + task.getExecutionContext() + .sendToMember( + new SourceEnumeratorEventOperation( + targetLocation, task.getTaskLocation(), event), + targetAddress); + } @Override public MetricsContext getMetricsContext() { diff --git a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/flow/SourceFlowLifeCycle.java b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/flow/SourceFlowLifeCycle.java index 54c1e9af68..40a2cea117 100644 --- a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/flow/SourceFlowLifeCycle.java +++ b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/flow/SourceFlowLifeCycle.java @@ -461,6 +461,20 @@ public class SourceFlowLifeCycle<T, SplitT extends SourceSplit> extends ActionFl } } + /** + * Delivers a source event sent by the remote split enumerator to this reader. + * + * <p>This runs on a Hazelcast operation thread, not the reader's task thread, so the + * implementation must be safe against concurrent access with the {@code pollNext} loop. The + * event is dispatched to {@link SourceReader#handleSourceEvent(SourceEvent)}, which connectors + * such as the CDC sources use to receive enumerator acknowledgements. + * + * @param sourceEvent the event sent by the enumerator + */ + public void receivedSourceEvent(SourceEvent sourceEvent) { + reader.handleSourceEvent(sourceEvent); + } + /** * Injects a checkpoint barrier into the record stream. * diff --git a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/operation/source/SourceEnumeratorEventOperation.java b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/operation/source/SourceEnumeratorEventOperation.java new file mode 100644 index 0000000000..206fe62cd3 --- /dev/null +++ b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/operation/source/SourceEnumeratorEventOperation.java @@ -0,0 +1,71 @@ +/* + * 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.engine.server.task.operation.source; + +import org.apache.seatunnel.api.source.SourceEvent; +import org.apache.seatunnel.common.utils.RetryUtils; +import org.apache.seatunnel.common.utils.SerializationUtils; +import org.apache.seatunnel.engine.common.Constant; +import org.apache.seatunnel.engine.server.SeaTunnelServer; +import org.apache.seatunnel.engine.server.exception.TaskGroupContextNotFoundException; +import org.apache.seatunnel.engine.server.execution.TaskLocation; +import org.apache.seatunnel.engine.server.serializable.TaskDataSerializerHook; +import org.apache.seatunnel.engine.server.task.SourceSeaTunnelTask; + +/** + * For {@link org.apache.seatunnel.api.source.SourceSplitEnumerator} send event to the {@link + * org.apache.seatunnel.api.source.SourceReader} + */ +public class SourceEnumeratorEventOperation extends SourceEventOperation { + public SourceEnumeratorEventOperation() {} + + public SourceEnumeratorEventOperation( + TaskLocation targetTaskLocation, TaskLocation currentTaskLocation, SourceEvent event) { + super(targetTaskLocation, currentTaskLocation, event); + } + + @Override + public int getClassId() { + return TaskDataSerializerHook.SOURCE_ENUMERATOR_EVENT_TYPE; + } + + @Override + public void runInternal() throws Exception { + SeaTunnelServer server = getService(); + RetryUtils.retryWithException( + () -> { + SourceSeaTunnelTask<?, ?> task = + server.getTaskExecutionService().getTask(taskLocation); + ClassLoader classLoader = + server.getTaskExecutionService() + .getExecutionContext(taskLocation.getTaskGroupLocation()) + .getClassLoader(task.getTaskID()); + SourceEvent sourceEvent = + SerializationUtils.deserialize(this.sourceEvent, classLoader); + task.receivedSourceEvent(sourceEvent); + return null; + }, + new RetryUtils.RetryMaterial( + Constant.OPERATION_RETRY_TIME, + true, + exception -> + exception instanceof TaskGroupContextNotFoundException + && !server.taskIsEnded(taskLocation.getTaskGroupLocation()), + Constant.OPERATION_RETRY_SLEEP)); + } +}
