This is an automated email from the ASF dual-hosted git repository.
davidzollo 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 152c410882 [Fix][seatunnel-translation-base] Remove connector-specific
test dependencies from translation base (#11075)
152c410882 is described below
commit 152c4108825a9bf614db912033c305360fb1dd03
Author: JeremyXin <[email protected]>
AuthorDate: Mon Jun 22 18:36:13 2026 +0800
[Fix][seatunnel-translation-base] Remove connector-specific test
dependencies from translation base (#11075)
---
.../seatunnel-translation-base/pom.xml | 13 -
.../translation/source/ParallelSourceTest.java | 343 +++++++++++++--------
2 files changed, 221 insertions(+), 135 deletions(-)
diff --git a/seatunnel-translation/seatunnel-translation-base/pom.xml
b/seatunnel-translation/seatunnel-translation-base/pom.xml
index 6c0cf4359e..87636af0bb 100644
--- a/seatunnel-translation/seatunnel-translation-base/pom.xml
+++ b/seatunnel-translation/seatunnel-translation-base/pom.xml
@@ -31,18 +31,5 @@
<artifactId>seatunnel-api</artifactId>
<version>${project.version}</version>
</dependency>
- <dependency>
- <groupId>org.apache.seatunnel</groupId>
- <artifactId>connector-file-base</artifactId>
- <version>${project.version}</version>
- <scope>test</scope>
- </dependency>
-
- <dependency>
- <groupId>org.apache.seatunnel</groupId>
- <artifactId>connector-doris</artifactId>
- <version>${project.version}</version>
- <scope>test</scope>
- </dependency>
</dependencies>
</project>
diff --git
a/seatunnel-translation/seatunnel-translation-base/src/test/java/org/apache/seatunnel/translation/source/ParallelSourceTest.java
b/seatunnel-translation/seatunnel-translation-base/src/test/java/org/apache/seatunnel/translation/source/ParallelSourceTest.java
index 1a2a3b98a7..71e28ad9a6 100644
---
a/seatunnel-translation/seatunnel-translation-base/src/test/java/org/apache/seatunnel/translation/source/ParallelSourceTest.java
+++
b/seatunnel-translation/seatunnel-translation-base/src/test/java/org/apache/seatunnel/translation/source/ParallelSourceTest.java
@@ -17,178 +17,277 @@
package org.apache.seatunnel.translation.source;
-import org.apache.seatunnel.shade.com.google.common.collect.Maps;
-
-import org.apache.seatunnel.api.table.catalog.TablePath;
-import org.apache.seatunnel.connectors.doris.config.DorisSourceConfig;
-import org.apache.seatunnel.connectors.doris.rest.PartitionDefinition;
-import org.apache.seatunnel.connectors.doris.rest.RestService;
-import org.apache.seatunnel.connectors.doris.source.DorisSource;
-import org.apache.seatunnel.connectors.doris.source.DorisSourceTable;
-import org.apache.seatunnel.connectors.doris.source.reader.DorisSourceReader;
-import org.apache.seatunnel.connectors.doris.source.split.DorisSourceSplit;
-import org.apache.seatunnel.connectors.seatunnel.file.source.BaseFileSource;
-import
org.apache.seatunnel.connectors.seatunnel.file.source.split.FileSourceSplit;
-import
org.apache.seatunnel.connectors.seatunnel.file.source.split.FileSourceSplitEnumerator;
+import org.apache.seatunnel.api.source.Boundedness;
+import org.apache.seatunnel.api.source.Collector;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
+import org.apache.seatunnel.api.source.SourceReader;
+import org.apache.seatunnel.api.source.SourceSplit;
+import org.apache.seatunnel.api.source.SourceSplitEnumerator;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
-import org.mockito.ArgumentCaptor;
-import org.mockito.MockedStatic;
-import org.mockito.Mockito;
-
-import lombok.extern.slf4j.Slf4j;
+import java.io.IOException;
+import java.io.Serializable;
import java.util.ArrayList;
-import java.util.HashSet;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
-import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
-import static org.mockito.ArgumentMatchers.any;
-
-@Slf4j
-public class ParallelSourceTest {
+class ParallelSourceTest {
@Test
- void fileParallelSourceSplitEnumeratorTest() throws Exception {
- int fileSize = 15;
+ void shouldDistributeSplitsAcrossParallelReaders() throws Exception {
+ int splitCount = 15;
int parallelism = 4;
+ List<MockSplit> splits =
+ IntStream.range(0, splitCount)
+ .mapToObj(i -> new MockSplit("split-" + i))
+ .collect(Collectors.toList());
+
+ MockSource source = new MockSource(splits);
+ Map<Integer, List<MockSplit>> assignedSplits = new LinkedHashMap<>();
+
+ for (int subtaskId = 0; subtaskId < parallelism; subtaskId++) {
+ ParallelSource<String, MockSplit, Integer> parallelSource =
+ new ParallelSource<>(
+ source, null, parallelism, "parallel-source-test",
subtaskId);
+ parallelSource.open();
+ parallelSource.splitEnumerator.run();
- List<String> filePaths = new ArrayList<>();
- for (int i = 0; i < fileSize; i++) {
- filePaths.add("file" + i + ".txt");
+ assignedSplits.put(subtaskId,
parallelSource.reader.snapshotState(0L));
+ parallelSource.close();
}
- BaseFileSource baseFileSource = Mockito.spy(BaseFileSource.class);
- Set<FileSourceSplit> splitSet = new HashSet<>();
- for (int i = 0; i < parallelism; i++) {
+ int totalAssigned =
assignedSplits.values().stream().mapToInt(List::size).sum();
+ Assertions.assertEquals(splitCount, totalAssigned);
- ParallelEnumeratorContext<FileSourceSplit> context =
- Mockito.mock(ParallelEnumeratorContext.class);
+ List<String> splitIds =
+ assignedSplits.values().stream()
+ .flatMap(List::stream)
+ .map(MockSplit::splitId)
+ .sorted()
+ .collect(Collectors.toList());
+ Assertions.assertEquals(
+
splits.stream().map(MockSplit::splitId).sorted().collect(Collectors.toList()),
+ splitIds);
- Mockito.when(context.currentParallelism()).thenReturn(parallelism);
+ for (int subtaskId = 0; subtaskId < parallelism; subtaskId++) {
+ Assertions.assertEquals(
+ expectedSplitCount(subtaskId, parallelism, splitCount),
+ assignedSplits.get(subtaskId).size());
+ }
+ }
- FileSourceSplitEnumerator fileSourceSplitEnumerator =
- new FileSourceSplitEnumerator(context, filePaths);
+ @Test
+ void shouldRestoreReaderSplitsFromCheckpointState() throws Exception {
+ MockSource source = new MockSource(Collections.emptyList());
+ Map<Integer, List<byte[]>> restoredState = new HashMap<>();
+ restoredState.put(-1,
Collections.singletonList(toBytes(Integer.valueOf(7))));
+ restoredState.put(0, Collections.singletonList(toBytes(new
MockSplit("restored-split-0"))));
+
+ ParallelSource<String, MockSplit, Integer> parallelSource =
+ new ParallelSource<>(source, restoredState, 4,
"parallel-source-restore", 0);
+ parallelSource.open();
+ parallelSource.splitEnumerator.run();
+
+ Assertions.assertEquals(
+ Collections.singletonList("restored-split-0"),
+ parallelSource.reader.snapshotState(0L).stream()
+ .map(MockSplit::splitId)
+ .sorted()
+ .collect(Collectors.toList()));
+
+ MockSplitEnumerator restoredEnumerator = source.lastRestoredEnumerator;
+ Assertions.assertNotNull(restoredEnumerator);
+ Assertions.assertEquals(Integer.valueOf(7),
restoredEnumerator.restoredState);
+
+ parallelSource.close();
+ }
- Mockito.when(baseFileSource.createEnumerator(any()))
- .thenReturn(fileSourceSplitEnumerator);
+ private static int expectedSplitCount(int subtaskId, int parallelism, int
splitCount) {
+ int splitsPerReader = splitCount / parallelism;
+ int remainder = splitCount % parallelism;
+ return subtaskId < remainder ? splitsPerReader + 1 : splitsPerReader;
+ }
- ParallelSource parallelSource =
- new ParallelSource(
- baseFileSource, null, parallelism,
"parallel-source-test" + i, i);
+ private static byte[] toBytes(Serializable value) throws IOException {
+ return new
org.apache.seatunnel.api.serialization.DefaultSerializer<Serializable>()
+ .serialize(value);
+ }
- parallelSource.open();
- parallelSource.splitEnumerator.run();
+ private static final class MockSource implements SeaTunnelSource<String,
MockSplit, Integer> {
- ArgumentCaptor<Integer> subtaskId =
ArgumentCaptor.forClass(Integer.class);
- ArgumentCaptor<List> split = ArgumentCaptor.forClass(List.class);
+ private final List<MockSplit> splits;
+ private MockSplitEnumerator lastRestoredEnumerator;
- Mockito.verify(context, Mockito.times(parallelism))
- .assignSplit(subtaskId.capture(), split.capture());
+ private MockSource(List<MockSplit> splits) {
+ this.splits = splits;
+ }
- List<Integer> subTaskAllValues = subtaskId.getAllValues();
- List<List> splitAllValues = split.getAllValues();
+ @Override
+ public Boundedness getBoundedness() {
+ return Boundedness.BOUNDED;
+ }
- Assertions.assertEquals(i, subTaskAllValues.get(i));
- Assertions.assertEquals(
- allocateFiles(i, parallelism, fileSize),
splitAllValues.get(i).size());
+ @Override
+ public SourceReader<String, MockSplit>
createReader(SourceReader.Context readerContext) {
+ return new MockReader();
+ }
- splitSet.addAll(splitAllValues.get(i));
+ @Override
+ public SourceSplitEnumerator<MockSplit, Integer> createEnumerator(
+ SourceSplitEnumerator.Context<MockSplit> enumeratorContext) {
+ return new MockSplitEnumerator(enumeratorContext, splits, null);
}
- // Check that there are no duplicate file assign
- Assertions.assertEquals(splitSet.size(), fileSize);
+ @Override
+ public SourceSplitEnumerator<MockSplit, Integer> restoreEnumerator(
+ SourceSplitEnumerator.Context<MockSplit> enumeratorContext,
+ Integer checkpointState) {
+ lastRestoredEnumerator =
+ new MockSplitEnumerator(
+ enumeratorContext, Collections.emptyList(),
checkpointState);
+ return lastRestoredEnumerator;
+ }
+
+ @Override
+ public String getPluginName() {
+ return "mock-source";
+ }
}
- @Test
- public void dorisParallelSourceSplitEnumeratorTest() throws Exception {
- int parallelism = 4;
- int partitionNums = 30;
+ private static final class MockSplitEnumerator
+ implements SourceSplitEnumerator<MockSplit, Integer> {
+
+ private final SourceSplitEnumerator.Context<MockSplit> context;
+ private final Map<Integer, List<MockSplit>> pendingSplits = new
HashMap<>();
+ private final Integer restoredState;
+ private final AtomicInteger assignCount = new AtomicInteger(0);
+
+ private MockSplitEnumerator(
+ SourceSplitEnumerator.Context<MockSplit> context,
+ List<MockSplit> splits,
+ Integer restoredState) {
+ this.context = context;
+ this.restoredState = restoredState;
+ addPendingSplits(splits);
+ }
- DorisSourceConfig dorisSourceConfig =
Mockito.mock(DorisSourceConfig.class);
- DorisSourceTable dorisSourceTable =
Mockito.mock(DorisSourceTable.class);
+ @Override
+ public void open() {}
- Map<TablePath, DorisSourceTable> dorisSourceTableMap =
Maps.newHashMap();
- dorisSourceTableMap.put(new TablePath("default", null,
"default_table"), dorisSourceTable);
+ @Override
+ public void run() {
+ assignSplits(context.registeredReaders());
+ context.registeredReaders().forEach(context::signalNoMoreSplits);
+ }
- DorisSource dorisSource = new DorisSource(dorisSourceConfig,
dorisSourceTableMap);
+ private static int getSplitOwner(int assignCount, int parallelism) {
+ return assignCount % parallelism;
+ }
- MockedStatic<RestService> restServiceMockedStatic =
Mockito.mockStatic(RestService.class);
- restServiceMockedStatic
- .when(() -> RestService.findPartitions(any(), any(), any()))
- .thenReturn(buildPartitionDefinitions(partitionNums));
+ private void assignSplits(Iterable<Integer> readers) {
+ for (Integer reader : readers) {
+ List<MockSplit> assigned = pendingSplits.remove(reader);
+ if (assigned != null && !assigned.isEmpty()) {
+ context.assignSplit(reader, assigned);
+ }
+ }
+ }
- Set<DorisSourceSplit> splitSet = new HashSet<>();
- for (int i = 0; i < parallelism; i++) {
- ParallelSource parallelSource =
- new ParallelSource(
- dorisSource, null, parallelism,
"parallel-doris-source" + i, i);
- parallelSource.open();
+ private void addPendingSplits(List<MockSplit> splits) {
+ int parallelism = context.currentParallelism();
+ List<MockSplit> orderedSplits =
+ splits.stream()
+ .sorted((left, right) ->
left.splitId().compareTo(right.splitId()))
+ .collect(Collectors.toList());
+ for (MockSplit split : orderedSplits) {
+ int ownerReader = getSplitOwner(assignCount.getAndIncrement(),
parallelism);
+ pendingSplits.computeIfAbsent(ownerReader, key -> new
ArrayList<>()).add(split);
+ }
+ }
- // execute file allocation process
- parallelSource.splitEnumerator.run();
- List<DorisSourceSplit> sourceSplits =
- ((DorisSourceReader)
parallelSource.reader).snapshotState(0);
- log.info(
- "parallel source{} splits => {}",
- i + 1,
- sourceSplits.stream()
- .map(DorisSourceSplit::splitId)
- .collect(Collectors.toList()));
+ @Override
+ public void close() {}
- Assertions.assertEquals(
- allocateFiles(i, parallelism, partitionNums),
sourceSplits.size());
+ @Override
+ public void addSplitsBack(List<MockSplit> splits, int subtaskId) {
+ pendingSplits.computeIfAbsent(subtaskId, key -> new
ArrayList<>()).addAll(splits);
+ }
+
+ @Override
+ public int currentUnassignedSplitSize() {
+ return pendingSplits.values().stream().mapToInt(List::size).sum();
+ }
+
+ @Override
+ public void handleSplitRequest(int subtaskId) {}
+
+ @Override
+ public void registerReader(int subtaskId) {}
- // collect all splits
- splitSet.addAll(sourceSplits);
+ @Override
+ public Integer snapshotState(long checkpointId) {
+ return restoredState;
}
- Assertions.assertEquals(splitSet.size(), partitionNums);
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) {}
+
+ @Override
+ public void notifyCheckpointAborted(long checkpointId) {}
}
- private List<PartitionDefinition> buildPartitionDefinitions(int
partitionNUms) {
+ private static final class MockReader implements SourceReader<String,
MockSplit> {
+
+ private final List<MockSplit> splits = new ArrayList<>();
+
+ @Override
+ public void open() {}
- List<PartitionDefinition> partitions = new ArrayList<>();
+ @Override
+ public void close() {}
- String beAddressPrefix = "doris-be-";
+ @Override
+ public void pollNext(Collector<String> output) {}
- IntStream.range(0, partitionNUms)
- .forEach(
- i -> {
- PartitionDefinition partitionDefinition =
- new PartitionDefinition(
- "default",
- "default_table",
- beAddressPrefix + i,
- new HashSet<>(i),
- "QUERY_PLAN");
+ @Override
+ public List<MockSplit> snapshotState(long checkpointId) {
+ return new ArrayList<>(splits);
+ }
+
+ @Override
+ public void addSplits(List<MockSplit> splits) {
+ this.splits.addAll(splits);
+ }
- partitions.add(partitionDefinition);
- });
+ @Override
+ public void handleNoMoreSplits() {}
- return partitions;
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) {}
+
+ @Override
+ public void notifyCheckpointAborted(long checkpointId) {}
}
- /**
- * calculate the number of files assigned each time
- *
- * @param id id
- * @param parallelism parallelism
- * @param fileSize file size
- * @return
- */
- public int allocateFiles(int id, int parallelism, int fileSize) {
- int filesPerIteration = fileSize / parallelism;
- int remainder = fileSize % parallelism;
-
- if (id < remainder) {
- return filesPerIteration + 1;
- } else {
- return filesPerIteration;
+ private static final class MockSplit implements SourceSplit {
+
+ private final String splitId;
+
+ private MockSplit(String splitId) {
+ this.splitId = splitId;
+ }
+
+ @Override
+ public String splitId() {
+ return splitId;
}
}
}
