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-12360-4648c9956ec88dec560a4af27cbd610568f47a60 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit eb76f81c823af25c9f4d75ba41b986bc8a0dab57 Author: Goutam Adwant <[email protected]> AuthorDate: Thu Sep 24 10:44:54 2026 +0000 [Test][CDC] Cover checkpoint offsets around fetcher failure (#12360) Signed-off-by: Goutam Adwant <[email protected]> --- .../source/reader/IncrementalSourceReaderTest.java | 415 +++++++++++++++++++++ 1 file changed, 415 insertions(+) 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 39f2dd8d45..6d29ddc13f 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 @@ -18,6 +18,8 @@ package org.apache.seatunnel.connectors.cdc.base.source.reader; import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.serialization.DefaultSerializer; +import org.apache.seatunnel.api.source.Collector; import org.apache.seatunnel.api.source.SourceReader; import org.apache.seatunnel.api.table.catalog.CatalogTable; import org.apache.seatunnel.api.table.catalog.TableIdentifier; @@ -30,14 +32,22 @@ import org.apache.seatunnel.api.table.type.SeaTunnelRowType; 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.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.SourceRecords; import org.apache.seatunnel.connectors.cdc.base.source.split.SourceSplitBase; import org.apache.seatunnel.connectors.cdc.base.source.split.state.SourceSplitStateBase; import org.apache.seatunnel.connectors.cdc.debezium.DebeziumDeserializationSchema; import org.apache.seatunnel.connectors.seatunnel.common.source.reader.RecordEmitter; +import org.apache.seatunnel.connectors.seatunnel.common.source.reader.RecordsBySplits; import org.apache.seatunnel.connectors.seatunnel.common.source.reader.RecordsWithSplitIds; import org.apache.seatunnel.connectors.seatunnel.common.source.reader.SourceReaderOptions; +import org.apache.seatunnel.connectors.seatunnel.common.source.reader.splitreader.SplitsChange; + +import org.apache.kafka.connect.data.Schema; +import org.apache.kafka.connect.data.SchemaBuilder; +import org.apache.kafka.connect.data.Struct; +import org.apache.kafka.connect.source.SourceRecord; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -46,6 +56,11 @@ import org.mockito.Mockito; import io.debezium.relational.TableId; +import java.io.IOException; +import java.lang.management.ManagementFactory; +import java.lang.management.ThreadInfo; +import java.lang.management.ThreadMXBean; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; @@ -54,9 +69,20 @@ import java.util.List; import java.util.Map; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.LockSupport; import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; class IncrementalSourceReaderTest { @@ -65,6 +91,395 @@ class IncrementalSourceReaderTest { private static final TableId REMOVED_TABLE = new TableId("alpha_online", null, "account_interests"); + @Test + void testFetcherFailureBeforeHandoffRestoresUnemittedBatch() throws Exception { + byte[] checkpoint; + try (CheckpointHarness harness = new CheckpointHarness(false)) { + harness.startAt(0); + harness.awaitPublishedBatch(); + harness.failFetcher(); + + // A successful snapshot after fetcher failure is safe if it still precedes A/B. + checkpoint = harness.snapshot(); + assertEquals(0, checkpointPosition(checkpoint)); + assertTrue(harness.emitted.isEmpty()); + assertFetcherFailure(harness); + } + assertRestoredRecords(checkpoint, Arrays.asList(1, 2)); + } + + @Test + void testFetcherFailureWithPartiallyEmittedFetchRestoresRemainingRecords() throws Exception { + byte[] checkpoint; + try (CheckpointHarness harness = new CheckpointHarness(true)) { + harness.startAt(0); + harness.awaitPublishedBatch(); + harness.reader.pollNext(harness.output); + assertEquals(Collections.singletonList(1), harness.emitted); + harness.failFetcher(); + + checkpoint = harness.snapshot(); + assertEquals(1, checkpointPosition(checkpoint)); + + // currentFetch can still be drained; its records have not become checkpointed early. + harness.reader.pollNext(harness.output); + assertEquals(Arrays.asList(1, 2), harness.emitted); + assertFetcherFailure(harness); + } + assertRestoredRecords(checkpoint, Collections.singletonList(2)); + } + + @Test + void testFetcherFailureDuringSnapshotRestoresOnlyEmittedPosition() throws Exception { + byte[] checkpoint; + try (CheckpointHarness harness = new CheckpointHarness(true)) { + harness.startAt(0); + harness.awaitPublishedBatch(); + harness.reader.pollNext(harness.output); + // Metadata is captured after SourceReaderBase has copied the live split states. + Mockito.when(harness.schema.getProducedType()) + .thenAnswer( + invocation -> { + harness.failFetcher(); + return Collections.emptyList(); + }); + + checkpoint = harness.snapshot(); + assertEquals(1, checkpointPosition(checkpoint)); + assertEquals(Collections.singletonList(1), harness.emitted); + } + assertRestoredRecords(checkpoint, Collections.singletonList(2)); + } + + @Test + void testHealthySnapshotDoesNotCheckpointQueuedRecords() throws Exception { + byte[] queuedCheckpoint; + byte[] emittedCheckpoint; + try (CheckpointHarness harness = new CheckpointHarness(false)) { + harness.startAt(0); + harness.awaitPublishedBatch(); + queuedCheckpoint = harness.snapshot(); + assertEquals(0, checkpointPosition(queuedCheckpoint)); + harness.reader.pollNext(harness.output); + assertEquals(Arrays.asList(1, 2), harness.emitted); + emittedCheckpoint = harness.snapshot(); + assertEquals(2, checkpointPosition(emittedCheckpoint)); + assertEquals(0, checkpointPosition(queuedCheckpoint)); + } + assertRestoredRecords(queuedCheckpoint, Arrays.asList(1, 2)); + assertRestoredRecords(emittedCheckpoint, Collections.emptyList()); + } + + @Test + void testCheckpointLockExcludesSnapshotDuringBatchHandoff() throws Exception { + ExecutorService executor = Executors.newFixedThreadPool(2); + CountDownLatch handoffStarted = new CountDownLatch(1); + CountDownLatch finishHandoff = new CountDownLatch(1); + CountDownLatch snapshotAttempted = new CountDownLatch(1); + AtomicReference<Thread> pollingThread = new AtomicReference<>(); + AtomicReference<Thread> snapshotThread = new AtomicReference<>(); + byte[] checkpoint; + try (CheckpointHarness harness = new CheckpointHarness(false)) { + harness.startAt(0); + harness.awaitPublishedBatch(); + Collector<Integer> blockingOutput = + new Collector<Integer>() { + @Override + public void collect(Integer record) { + if (record == 1) { + handoffStarted.countDown(); + await(finishHandoff); + } + harness.output.collect(record); + } + + @Override + public Object getCheckpointLock() { + return harness.output.getCheckpointLock(); + } + }; + Future<?> poll = + executor.submit( + () -> { + pollingThread.set(Thread.currentThread()); + harness.reader.pollNext(blockingOutput); + return null; + }); + await(handoffStarted); + harness.failFetcher(); + Future<byte[]> snapshot = + executor.submit( + () -> { + snapshotThread.set(Thread.currentThread()); + snapshotAttempted.countDown(); + return harness.snapshot(); + }); + await(snapshotAttempted); + ThreadMXBean threads = ManagementFactory.getThreadMXBean(); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + while (true) { + ThreadInfo info = threads.getThreadInfo(snapshotThread.get().getId()); + if (info != null + && info.getThreadState() == Thread.State.BLOCKED + && info.getLockOwnerId() == pollingThread.get().getId() + && info.getLockInfo().getIdentityHashCode() + == System.identityHashCode(harness.output.getCheckpointLock())) { + break; + } + assertTrue( + System.nanoTime() < deadline, + "Snapshot did not block on the collector checkpoint lock: " + info); + assertFalse(Thread.currentThread().isInterrupted()); + LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1)); + } + finishHandoff.countDown(); + poll.get(10, TimeUnit.SECONDS); + checkpoint = snapshot.get(10, TimeUnit.SECONDS); + assertEquals(Arrays.asList(1, 2), harness.emitted); + assertEquals(2, checkpointPosition(checkpoint)); + } finally { + finishHandoff.countDown(); + executor.shutdownNow(); + assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + } + assertRestoredRecords(checkpoint, Collections.emptyList()); + } + + private static void assertFetcherFailure(CheckpointHarness harness) { + RuntimeException failure = + assertThrows(RuntimeException.class, () -> harness.reader.pollNext(harness.output)); + assertEquals("One or more fetchers have encountered exception", failure.getMessage()); + Throwable cause = failure; + while (cause.getCause() != null) { + cause = cause.getCause(); + } + assertSame(harness.failure, cause); + } + + private static int checkpointPosition(byte[] checkpoint) throws IOException { + SourceSplitBase split = new DefaultSerializer<SourceSplitBase>().deserialize(checkpoint); + return Integer.parseInt( + split.asIncrementalSplit().getStartupOffset().getOffset().get("pos")); + } + + private static void assertRestoredRecords(byte[] checkpoint, List<Integer> expected) + throws Exception { + try (CheckpointHarness restored = new CheckpointHarness(false)) { + SourceSplitBase split = + new DefaultSerializer<SourceSplitBase>().deserialize(checkpoint); + restored.reader.addSplits(Collections.singletonList(split)); + restored.awaitPublishedBatch(); + assertEquals(checkpointPosition(checkpoint), checkpointPosition(restored.snapshot())); + restored.reader.pollNext(restored.output); + assertEquals(expected, restored.emitted); + assertEquals(2, checkpointPosition(restored.snapshot())); + } + } + + private static void await(CountDownLatch latch) { + try { + assertTrue( + latch.await(10, TimeUnit.SECONDS), "Timed out waiting for reader/fetcher gate"); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError("Interrupted waiting for reader/fetcher gate", e); + } + } + + private static final class CheckpointHarness implements AutoCloseable { + private static final String SPLIT_ID = "incremental-split-0"; + private final CountDownLatch publishedBatch = new CountDownLatch(1); + private final CountDownLatch releaseFetcher = new CountDownLatch(1); + private final CountDownLatch fetcherClosed = new CountDownLatch(1); + private final AtomicInteger candidatePosition = new AtomicInteger(); + private final IOException failure = new IOException("injected fatal fetch failure"); + private volatile boolean failing; + private final List<Integer> emitted = new ArrayList<>(); + private final Collector<Integer> output = + new Collector<Integer>() { + @Override + public void collect(Integer record) { + assertTrue(Thread.holdsLock(getCheckpointLock())); + emitted.add(record); + } + + @Override + public Object getCheckpointLock() { + return this; + } + }; + private final DebeziumDeserializationSchema<Integer> schema; + private final IncrementalSourceReader<Integer, SourceConfig> reader; + + @SuppressWarnings("unchecked") + private CheckpointHarness(boolean separateRecordGroups) throws Exception { + SourceReader.Context context = + Mockito.mock(SourceReader.Context.class, Mockito.RETURNS_DEEP_STUBS); + schema = Mockito.mock(DebeziumDeserializationSchema.class); + Mockito.doAnswer( + invocation -> { + SourceRecord record = invocation.getArgument(0); + Collector<Integer> collector = invocation.getArgument(1); + collector.collect( + ((Number) record.sourceOffset().get("pos")).intValue()); + return null; + }) + .when(schema) + .deserialize(Mockito.any(), Mockito.any()); + OffsetFactory offsets = Mockito.mock(OffsetFactory.class); + Mockito.when(offsets.specific(Mockito.anyMap())) + .thenAnswer( + invocation -> { + Map<String, String> offset = invocation.getArgument(0); + return new TestOffset(Integer.parseInt(offset.get("pos"))); + }); + IncrementalSourceSplitReader<SourceConfig> splitReader = + Mockito.mock(IncrementalSourceSplitReader.class); + Mockito.doAnswer( + invocation -> { + SplitsChange<SourceSplitBase> change = invocation.getArgument(0); + candidatePosition.set( + Integer.parseInt( + change.splits() + .iterator() + .next() + .asIncrementalSplit() + .getStartupOffset() + .getOffset() + .get("pos"))); + return null; + }) + .when(splitReader) + .handleSplitsChanges(Mockito.any()); + Mockito.when(splitReader.fetch()) + .thenAnswer( + invocation -> { + List<SourceRecords> groups = new ArrayList<>(); + List<SourceRecord> records = new ArrayList<>(); + for (int position = candidatePosition.get() + 1; + position <= 2; + position++) { + records.add(dataRecord(position)); + if (separateRecordGroups) { + groups.add(new SourceRecords(new ArrayList<>(records))); + records.clear(); + } + } + if (!separateRecordGroups) { + groups.add(new SourceRecords(records)); + } + candidatePosition.set(2); + return new RecordsBySplits<>( + Collections.singletonMap(SPLIT_ID, groups), + Collections.emptySet()); + }) + .thenAnswer( + invocation -> { + // FetchTask has enqueued the previous result before requesting the + // next batch. + publishedBatch.countDown(); + await(releaseFetcher); + if (failing) { + throw failure; + } + return new RecordsBySplits<>( + Collections.emptyMap(), Collections.emptySet()); + }); + Mockito.doAnswer( + invocation -> { + releaseFetcher.countDown(); + return null; + }) + .when(splitReader) + .wakeUp(); + Mockito.doAnswer( + invocation -> { + // SplitFetcher records its fatal error before closing the split + // reader. + fetcherClosed.countDown(); + return null; + }) + .when(splitReader) + .close(); + reader = + new IncrementalSourceReader<>( + Mockito.mock(DataSourceDialect.class), + new ArrayBlockingQueue<>(2), + () -> splitReader, + new IncrementalSourceRecordEmitter<>(schema, offsets, context), + new SourceReaderOptions(ReadonlyConfig.fromMap(Collections.emptyMap())), + context, + Mockito.mock(SourceConfig.class), + schema); + } + + private void startAt(int position) { + reader.addSplits( + Collections.singletonList( + new IncrementalSplit( + SPLIT_ID, + Collections.singletonList(KEPT_TABLE), + new TestOffset(position), + null, + Collections.emptyList()))); + } + + private void awaitPublishedBatch() { + await(publishedBatch); + assertEquals(2, candidatePosition.get()); + } + + private void failFetcher() { + failing = true; + releaseFetcher.countDown(); + await(fetcherClosed); + } + + private byte[] snapshot() throws IOException { + // Engine-style checkpoint exclusion; Flink also serializes poll/snapshot in its + // mailbox. + synchronized (output.getCheckpointLock()) { + List<SourceSplitBase> splits = reader.snapshotState(1L); + assertEquals(1, splits.size()); + return new DefaultSerializer<SourceSplitBase>().serialize(splits.get(0)); + } + } + + @Override + public void close() { + reader.close(); + await(fetcherClosed); + } + } + + private static SourceRecord dataRecord(int position) { + Schema valueSchema = + SchemaBuilder.struct() + .name("test.Envelope") + .field("op", Schema.STRING_SCHEMA) + .build(); + return new SourceRecord( + Collections.singletonMap("server", "test"), + Collections.singletonMap("pos", position), + "test", + null, + null, + valueSchema, + new Struct(valueSchema).put("op", "c")); + } + + private static final class TestOffset extends Offset { + private TestOffset(int position) { + offset = Collections.singletonMap("pos", Integer.toString(position)); + } + + @Override + public int compareTo(Offset other) { + return Long.compare( + longOffsetValue(offset, "pos"), longOffsetValue(other.getOffset(), "pos")); + } + } + @Test void testAddSplitsEnqueuesPrunedRestoredIncrementalSplit() { SourceConfig sourceConfig = Mockito.mock(SourceConfig.class);
