(shardingsphere) branch master updated: Add CDCImporterTest (#37386)

[zh...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

zhangliang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere.git


The following commit(s) were added to refs/heads/master by this push:
     new 9a7ab34a15c Add CDCImporterTest (#37386)
9a7ab34a15c is described below

commit 9a7ab34a15c4df320968f77eadee88d2f5b61fbc
Author: Liang Zhang <zh...@apache.org>
AuthorDate: Mon Dec 15 00:32:10 2025 +0800

    Add CDCImporterTest (#37386)
    
    * Add CDCImporterTest
    
    * Add CDCImporterTest
---
 .../cdc/core/importer/CDCImporterTest.java         | 225 +++++++++++++++++++++
 1 file changed, 225 insertions(+)

diff --git a/kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/core/importer/CDCImporterTest.java b/kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/core/importer/CDCImporterTest.java
new file mode 100644
index 00000000000..48569d7dedb
--- /dev/null
+++ b/kernel/data-pipeline/scenario/cdc/core/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/core/importer/CDCImporterTest.java
@@ -0,0 +1,225 @@
+/*
+ * 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.shardingsphere.data.pipeline.cdc.core.importer;
+
+import org.apache.shardingsphere.data.pipeline.core.channel.PipelineChannel;
+import org.apache.shardingsphere.data.pipeline.core.constant.PipelineSQLOperationType;
+import org.apache.shardingsphere.data.pipeline.core.importer.sink.PipelineSink;
+import org.apache.shardingsphere.data.pipeline.core.ingest.position.type.placeholder.IngestPlaceholderPosition;
+import org.apache.shardingsphere.data.pipeline.core.ingest.record.DataRecord;
+import org.apache.shardingsphere.data.pipeline.core.ingest.record.FinishedRecord;
+import org.apache.shardingsphere.data.pipeline.core.ingest.record.Record;
+import org.apache.shardingsphere.data.pipeline.core.job.progress.listener.PipelineJobProgressListener;
+import org.apache.shardingsphere.data.pipeline.core.job.progress.listener.PipelineJobUpdateProgress;
+import org.apache.shardingsphere.data.pipeline.core.ratelimit.JobRateLimitAlgorithm;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.contains;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyCollection;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class CDCImporterTest {
+    
+    @Test
+    void assertDoWithoutSortingCoversEmptyNonDataAndData() {
+        List<List<Record>> fetchResults = Arrays.asList(Collections.emptyList(), asList(mock(Record.class)), asList(createDataRecord(1L), createFinishedRecord()));
+        PipelineChannel channel = mockNonSortingChannel(fetchResults);
+        PipelineJobProgressListener progressListener = mock(PipelineJobProgressListener.class);
+        AtomicReference<CDCImporter> importerHolder = new AtomicReference<>();
+        PipelineSink sink = mockSinkWithAck(importerHolder, false);
+        CDCImporter importer = new CDCImporter(new LinkedList<>(Collections.singletonList(new CDCChannelProgressPair(channel, progressListener))), 10, 1L, sink, false, null);
+        importerHolder.set(importer);
+        importer.start();
+        importer.stop();
+        assertThat(captureAckRecords(channel, 2).size(), is(2));
+        assertThat(captureProgressCounts(progressListener, 3), contains(0, 0, 1));
+        assertNull(CDCImporterManager.getImporter(importer.getImporterId()));
+    }
+    
+    @Test
+    void assertDoWithoutSortingNonDataFinishedRemovesChannel() {
+        PipelineChannel channel = mockNonSortingChannel(Collections.singletonList(asList(createFinishedRecord())));
+        PipelineJobProgressListener progressListener = mock(PipelineJobProgressListener.class);
+        PipelineSink sink = mockSimpleSink();
+        CDCImporter importer = new CDCImporter(new LinkedList<>(Collections.singletonList(new CDCChannelProgressPair(channel, progressListener))), 1, 1L, sink, false, null);
+        importer.start();
+        importer.stop();
+        assertThat(captureAckRecords(channel, 1).get(0).get(0), instanceOf(FinishedRecord.class));
+        assertThat(captureProgressCounts(progressListener, 2), contains(0, 0));
+    }
+    
+    @Test
+    void assertDoWithoutSortingWithRateLimitAndAck() {
+        PipelineChannel channel = mockNonSortingChannel(Collections.singletonList(asList(createDataRecord(5L))));
+        PipelineJobProgressListener progressListener = mock(PipelineJobProgressListener.class);
+        JobRateLimitAlgorithm rateLimitAlgorithm = mock(JobRateLimitAlgorithm.class);
+        AtomicReference<CDCImporter> importerHolder = new AtomicReference<>();
+        PipelineSink sink = mockSinkWithAck(importerHolder, true);
+        CDCImporter importer = new CDCImporter(new LinkedList<>(Collections.singletonList(new CDCChannelProgressPair(channel, progressListener))), 1, 1L, sink, false, rateLimitAlgorithm);
+        importerHolder.set(importer);
+        importer.start();
+        verify(rateLimitAlgorithm).intercept(PipelineSQLOperationType.INSERT, 1);
+        assertThat(captureProgressCounts(progressListener, 2), contains(0, 1));
+        assertThat(captureAckRecords(channel, 1).size(), is(1));
+    }
+    
+    @Test
+    void assertDoWithSortingHandlesEmptyAndSingleTransaction() {
+        Queue<List<Record>> transactions = new LinkedList<>(Arrays.asList(Collections.emptyList(), asList(createFinishedRecord(), createDataRecord(2L), createFinishedRecord())));
+        PipelineChannel channel = mockSortingChannel(transactions);
+        PipelineJobProgressListener progressListener = mock(PipelineJobProgressListener.class);
+        AtomicReference<CDCImporter> importerHolder = new AtomicReference<>();
+        PipelineSink sink = mockSinkWithAck(importerHolder, false);
+        CDCImporter importer = new CDCImporter(new LinkedList<>(Collections.singletonList(new CDCChannelProgressPair(channel, progressListener))), 1, 1L, sink, true, null);
+        importerHolder.set(importer);
+        importer.start();
+        importer.stop();
+        List<List<Record>> ackedRecords = captureAckRecords(channel, 1);
+        assertThat(ackedRecords.size(), is(1));
+        assertThat(ackedRecords.get(0).get(0), instanceOf(FinishedRecord.class));
+        assertThat(captureProgressCounts(progressListener, 2), contains(0, 1));
+    }
+    
+    @Test
+    void assertDoWithSortingProcessesMultipleCsnBatches() {
+        Queue<List<Record>> channelOneTransactions = new LinkedList<>(Arrays.asList(asList(createFinishedRecord(), createDataRecord(4L)), asList(createDataRecord(4L)), Collections.emptyList()));
+        PipelineChannel channelOne = mockSortingChannel(channelOneTransactions);
+        Queue<List<Record>> channelTwoTransactions = new LinkedList<>(Arrays.asList(asList(createDataRecord(4L), createFinishedRecord()), asList(createFinishedRecord()), Collections.emptyList()));
+        PipelineChannel channelTwo = mockSortingChannel(channelTwoTransactions);
+        Queue<List<Record>> channelThreeTransactions = new LinkedList<>(Arrays.asList(asList(createDataRecord(6L)), asList(createDataRecord(7L))));
+        PipelineChannel channelThree = mockSortingChannel(channelThreeTransactions);
+        Queue<List<Record>> channelFourTransactions = new LinkedList<>(Arrays.asList(asList(createFinishedRecord()), Collections.emptyList()));
+        PipelineChannel channelFour = mockSortingChannel(channelFourTransactions);
+        List<CDCChannelProgressPair> pairs = new LinkedList<>();
+        pairs.add(new CDCChannelProgressPair(channelOne, mock(PipelineJobProgressListener.class)));
+        pairs.add(new CDCChannelProgressPair(channelTwo, mock(PipelineJobProgressListener.class)));
+        pairs.add(new CDCChannelProgressPair(channelThree, mock(PipelineJobProgressListener.class)));
+        pairs.add(new CDCChannelProgressPair(channelFour, mock(PipelineJobProgressListener.class)));
+        JobRateLimitAlgorithm rateLimitAlgorithm = mock(JobRateLimitAlgorithm.class);
+        AtomicReference<CDCImporter> importerHolder = new AtomicReference<>();
+        PipelineSink sink = mockSinkWithAck(importerHolder, true);
+        CDCImporter importer = new CDCImporter(pairs, 2, 1L, sink, true, rateLimitAlgorithm);
+        importerHolder.set(importer);
+        importer.start();
+        verify(rateLimitAlgorithm).intercept(PipelineSQLOperationType.INSERT, 1);
+        assertThat(captureAckRecords(channelTwo, 2).size(), is(2));
+        assertThat(captureAckRecords(channelFour, 1).size(), is(1));
+        verify(channelThree, never()).ack(any());
+        importer.ack("missing_ack");
+        assertNull(CDCImporterManager.getImporter(importer.getImporterId()));
+    }
+    
+    private DataRecord createDataRecord(final long csn) {
+        DataRecord result = new DataRecord(PipelineSQLOperationType.INSERT, "t_order", new IngestPlaceholderPosition(), 1);
+        result.setCsn(csn);
+        return result;
+    }
+    
+    private FinishedRecord createFinishedRecord() {
+        return new FinishedRecord(new IngestPlaceholderPosition());
+    }
+    
+    private List<Record> asList(final Record... records) {
+        return new LinkedList<>(Arrays.asList(records));
+    }
+    
+    private PipelineChannel mockNonSortingChannel(final List<List<Record>> fetchResults) {
+        PipelineChannel result = mock(PipelineChannel.class);
+        Queue<List<Record>> fetchQueue = new LinkedList<>(fetchResults);
+        when(result.fetch(anyInt(), anyLong())).thenAnswer(invocation -> fetchQueue.isEmpty() ? Collections.emptyList() : fetchQueue.poll());
+        when(result.peek()).thenReturn(Collections.emptyList());
+        when(result.poll()).thenReturn(Collections.emptyList());
+        return result;
+    }
+    
+    private PipelineChannel mockSortingChannel(final Queue<List<Record>> transactions) {
+        PipelineChannel result = mock(PipelineChannel.class);
+        when(result.fetch(anyInt(), anyLong())).thenReturn(Collections.emptyList());
+        when(result.peek()).thenAnswer(invocation -> {
+            List<Record> records = transactions.peek();
+            return null == records ? Collections.emptyList() : records;
+        });
+        when(result.poll()).thenAnswer(invocation -> {
+            List<Record> records = transactions.poll();
+            return null == records ? Collections.emptyList() : records;
+        });
+        return result;
+    }
+    
+    private PipelineSink mockSimpleSink() {
+        PipelineSink result = mock(PipelineSink.class);
+        when(result.write(anyString(), anyCollection())).thenAnswer(invocation -> {
+            Collection<Record> records = invocation.getArgument(1);
+            return new PipelineJobUpdateProgress(records.size());
+        });
+        return result;
+    }
+    
+    private PipelineSink mockSinkWithAck(final AtomicReference<CDCImporter> importerHolder, final boolean stopAfterWrite) {
+        PipelineSink result = mock(PipelineSink.class);
+        when(result.write(anyString(), anyCollection())).thenAnswer(invocation -> {
+            String ackId = invocation.getArgument(0);
+            Collection<Record> records = invocation.getArgument(1);
+            importerHolder.get().ack(ackId);
+            if (stopAfterWrite) {
+                importerHolder.get().stop();
+            }
+            return new PipelineJobUpdateProgress(records.size());
+        });
+        return result;
+    }
+    
+    @SuppressWarnings({"rawtypes", "unchecked"})
+    private List<List<Record>> captureAckRecords(final PipelineChannel channel, final int expectedTimes) {
+        ArgumentCaptor<List> captor = ArgumentCaptor.forClass(List.class);
+        verify(channel, times(expectedTimes)).ack(captor.capture());
+        return captor.getAllValues().stream().map(each -> {
+            @SuppressWarnings("unchecked")
+            List<Record> records = (List<Record>) each;
+            return records;
+        }).collect(Collectors.toList());
+    }
+    
+    private List<Integer> captureProgressCounts(final PipelineJobProgressListener listener, final int expectedTimes) {
+        ArgumentCaptor<PipelineJobUpdateProgress> captor = ArgumentCaptor.forClass(PipelineJobUpdateProgress.class);
+        verify(listener, times(expectedTimes)).onProgressUpdated(captor.capture());
+        return captor.getAllValues().stream().map(PipelineJobUpdateProgress::getProcessedRecordsCount).collect(Collectors.toList());
+    }
+}