[feature][connector][fake] Support mutil splits for fake source connector #2974

liugddx · 2022-10-02T23:04:14Z

close #2961

Purpose of this pull request

feature Support more than splits and parallelism for fake connector.

Check list

Code changed are covered with tests, or it does not need tests for reason:
If any new Jar binary package adding in your PR, please add License Notice according
New License Guide
If necessary, please update the documentation to describe the new feature. https://github.com/apache/incubator-seatunnel/tree/dev/docs

close apache#2961

TyrantLucifer · 2022-10-03T05:26:29Z

...ke/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceReader.java


-    public FakeSourceReader(SingleSplitReaderContext context, FakeDataGenerator randomData) {
+    public FakeSourceReader(SourceReader.Context context, FakeDataGenerator randomData) {


Suggested change

public FakeSourceReader(SourceReader.Context context, FakeDataGenerator randomData) {

public FakeSourceReader(SourceReader.Context context, FakeDataGenerator fakeDataGenerator) {

Rename randomData to fakeDataGenerator?

TyrantLucifer · 2022-10-03T05:29:50Z

...in/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceSplitEnumerator.java

+    private final Map<Integer, Set<FakeSourceSplit>> pendingSplits;
+    private final Integer totalRowNum;
+
+    public FakeSourceSplitEnumerator(SourceSplitEnumerator.Context<FakeSourceSplit> enumeratorContext, Integer totalRowNum) {


Suggested change

public FakeSourceSplitEnumerator(SourceSplitEnumerator.Context<FakeSourceSplit> enumeratorContext, Integer totalRowNum) {

public FakeSourceSplitEnumerator(SourceSplitEnumerator.Context<FakeSourceSplit> enumeratorContext, int totalRowNum) {

According to the review comment in FakeSourceSplitEnumerator, this parameter maybe need to be removed.

TyrantLucifer · 2022-10-03T05:30:00Z

...in/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceSplitEnumerator.java

+    private static final Logger LOG = LoggerFactory.getLogger(FakeSourceSplitEnumerator.class);
+    private final SourceSplitEnumerator.Context<FakeSourceSplit> enumeratorContext;
+    private final Map<Integer, Set<FakeSourceSplit>> pendingSplits;
+    private final Integer totalRowNum;


Suggested change

private final Integer totalRowNum;

private final int totalRowNum;

According to the review comment in FakeSourceSplitEnumerator, this parameter maybe need to be removed.

TyrantLucifer · 2022-10-03T05:33:59Z

...ake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceSplit.java

+@Data
+@AllArgsConstructor
+public class FakeSourceSplit implements SourceSplit {
+    private Integer rowNum;


Suggested change

private Integer rowNum;

private int rowNum;

According to the review comment in FakeSourceSplitEnumerator, this parameter maybe need to be removed.

...ake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceSplit.java

TyrantLucifer · 2022-10-03T05:42:56Z

...in/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceSplitEnumerator.java

+        LOG.info("Starting to calculate splits.");
+        int numReaders = enumeratorContext.currentParallelism();
+
+        if (null != totalRowNum) {


How about using the parameter row.size to control the row num of each reader not the total number? The disadvantage of slicing now is that the user needs to calculate how many data will be generated by each parallel task during configuration, our original intention is to make user configuration simpler, the user only needs to configure how many data will be generated by each slice, so that the slice only needs one parameter id, then only need to use the id to take the balance of the parallelism when the slice is assigned.

I think fake should be simple,so a simple sharding approach is needed here, and the configuration file does not need to be changed.

TyrantLucifer · 2022-10-03T05:48:25Z

...e/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeDataGenerator.java

@@ -57,9 +58,9 @@ private SeaTunnelRow randomRow() {
        return new SeaTunnelRow(randomRow.toArray());
    }

-    public List<SeaTunnelRow> generateFakedRows() {
+    public List<SeaTunnelRow> generateFakedRows(int rowNum) {


According to the review comment in FakeSourceSplitEnumerator, this method maybe need to revert because every split has the same row num.

liugddx · 2022-10-05T07:05:30Z

The reason why CI error is #2984 , please review. @TyrantLucifer @EricJoy2048

TyrantLucifer · 2022-10-05T07:53:00Z

...ake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceSplit.java

+@Data
+@AllArgsConstructor
+public class FakeSourceSplit implements SourceSplit {
+    private Integer rowNum;


...ake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceSplit.java

TyrantLucifer · 2022-10-05T07:55:10Z

...e/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeDataGenerator.java

@@ -57,9 +58,9 @@ private SeaTunnelRow randomRow() {
        return new SeaTunnelRow(randomRow.toArray());
    }

-    public List<SeaTunnelRow> generateFakedRows() {
+    public List<SeaTunnelRow> generateFakedRows(int rowNum) {


TyrantLucifer · 2022-10-05T07:56:10Z

...tor-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSource.java

@@ -51,7 +51,17 @@ public SeaTunnelRowType getProducedType() {
    }

    @Override
-    public AbstractSingleSplitReader<SeaTunnelRow> createReader(SingleSplitReaderContext readerContext) throws Exception {
+    public SourceSplitEnumerator<FakeSourceSplit, FakeSourceState> createEnumerator(SourceSplitEnumerator.Context<FakeSourceSplit> enumeratorContext) throws Exception {
+        return new FakeSourceSplitEnumerator(enumeratorContext, fakeConfig.getRowNum());


Suggested change

return new FakeSourceSplitEnumerator(enumeratorContext, fakeConfig.getRowNum());

return new FakeSourceSplitEnumerator(enumeratorContext, fakeConfig);

TyrantLucifer · 2022-10-05T07:56:26Z

...tor-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSource.java

+
+    @Override
+    public SourceSplitEnumerator<FakeSourceSplit, FakeSourceState> restoreEnumerator(SourceSplitEnumerator.Context<FakeSourceSplit> enumeratorContext, FakeSourceState checkpointState) throws Exception {
+        return new FakeSourceSplitEnumerator(enumeratorContext, fakeConfig.getRowNum());


Suggested change

return new FakeSourceSplitEnumerator(enumeratorContext, fakeConfig.getRowNum());

return new FakeSourceSplitEnumerator(enumeratorContext, fakeConfig);

TyrantLucifer · 2022-10-05T07:56:47Z

...ke/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceReader.java

+            FakeSourceSplit split = splits.poll();
+            if (null != split) {
+                // Generate a random number of rows to emit.
+                List<SeaTunnelRow> seaTunnelRows = fakeDataGenerator.generateFakedRows(split.getRowNum());


Suggested change

List<SeaTunnelRow> seaTunnelRows = fakeDataGenerator.generateFakedRows(split.getRowNum());

List<SeaTunnelRow> seaTunnelRows = fakeDataGenerator.generateFakedRows();

TyrantLucifer · 2022-10-05T07:58:19Z

...in/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceSplitEnumerator.java

+        LOG.info("Starting to calculate splits.");
+        int numReaders = enumeratorContext.currentParallelism();
+        for (int i = 0; i < numReaders; i++) {
+            allSplit.add(new FakeSourceSplit(rowNum, i));


Suggested change

allSplit.add(new FakeSourceSplit(rowNum, i));

allSplit.add(new FakeSourceSplit(i));

TyrantLucifer · 2022-10-05T07:59:58Z

...c/test/java/org.apache.seatunnel.connectors.seatunnel.fake.source/FakeDataGeneratorTest.java

@@ -46,7 +47,7 @@ public void testComplexSchemaParse(String conf) throws FileNotFoundException, UR
        SeaTunnelRowType seaTunnelRowType = seaTunnelSchema.getSeaTunnelRowType();
        FakeConfig fakeConfig = FakeConfig.buildWithConfig(testConfig);
        FakeDataGenerator fakeDataGenerator = new FakeDataGenerator(seaTunnelSchema, fakeConfig);
-        List<SeaTunnelRow> seaTunnelRows = fakeDataGenerator.generateFakedRows();
+        List<SeaTunnelRow> seaTunnelRows = fakeDataGenerator.generateFakedRows(fakeConfig.getRowNum());


Suggested change

List<SeaTunnelRow> seaTunnelRows = fakeDataGenerator.generateFakedRows(fakeConfig.getRowNum());

List<SeaTunnelRow> seaTunnelRows = fakeDataGenerator.generateFakedRows(fakeConfig);

liugddx · 2022-10-05T09:06:37Z

please review again @TyrantLucifer

TyrantLucifer · 2022-10-05T09:20:48Z

...ke/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceReader.java


-    public FakeSourceReader(SingleSplitReaderContext context, FakeDataGenerator randomData) {
+    public FakeSourceReader(SourceReader.Context context, FakeDataGenerator randomData) {


Rename randomData to fakeDataGenerator?

TyrantLucifer · 2022-10-05T09:21:28Z

...ake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceSplit.java

+
+    @Override
+    public String splitId() {
+        return splitId.toString();


Suggested change

return splitId.toString();

return String.valueOf(splitId);

… improve-supportparallelism

…lelism

TyrantLucifer · 2022-10-05T15:35:47Z

...ke/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceReader.java

+    @Override
+    public void handleNoMoreSplits() {
+        noMoreSplit = true;
+


Remove useless blank line.

TyrantLucifer

+1, cc @EricJoy2048

...tor-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/config/FakeConfig.java

EricJoy2048 · 2022-10-06T13:34:01Z

Please fix the CI error

liugddx · 2022-10-06T13:39:44Z

Please fix the CI error

Error: JobMasterTest.testHandleCheckpointTimeout:128 » ConditionTimeout Assertion con...

Maybe time setting is too short?

EricJoy2048 · 2022-10-07T05:44:26Z

Please fix the CI error

CI error will be fixed by #3009. I will retry the CI after #3009 merged.

EricJoy2048 · 2022-10-07T06:26:52Z

Please fix the CI error

CI error will be fixed by #3009. I will retry the CI after #3009 merged.

You need merge the dev branch to your branch first and then push it .

liugddx · 2022-10-08T05:12:58Z

rerun CI Thanks. @EricJoy2048

EricJoy2048 · 2022-10-08T10:24:53Z

Please wait #3019 merged. The CI error because #3019.

…lelism

EricJoy2048 · 2022-10-09T14:43:26Z

Some bug we must fix to resolve the CI problems. Please wait for us.

…lelism

liugddx · 2022-10-10T13:18:33Z

rerun CI please. @EricJoy2048

EricJoy2048

LGTM

liugddx · 2022-10-12T07:36:46Z

@ashulin help to review .

hailin0 · 2022-10-13T03:52:15Z

...ke/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceReader.java


+    private final SourceReader.Context context;
+    private final Deque<FakeSourceSplit> splits = new LinkedList<>();


Suggested change

private final Deque<FakeSourceSplit> splits = new LinkedList<>();

private final Queue<FakeSourceSplit> splits = new LinkedList<>();

unused deque api?

unused deque api?

cc @ashulin

...ke/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceReader.java

…e/seatunnel/connectors/seatunnel/fake/source/FakeSourceReader.java Co-authored-by: hailin0 <hailin088@gmail.com>

… improve-supportparallelism # Conflicts: # seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSource.java # seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceReader.java # seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceSplitEnumerator.java

…lelism

EricJoy2048

+1

hailin0 · 2022-10-13T14:50:35Z

+1

feature Support more than splits and parallelism for fake connector

1357050

close apache#2961

liugddx marked this pull request as draft October 2, 2022 23:04

test parallelism

d00dec8

liugddx marked this pull request as ready for review October 2, 2022 23:41

TyrantLucifer reviewed Oct 3, 2022

View reviewed changes

liugddx closed this Oct 4, 2022

liugddx deleted the improve-supportparallelism branch October 4, 2022 08:07

liugddx restored the improve-supportparallelism branch October 4, 2022 08:10

liugddx reopened this Oct 4, 2022

liugddx added 2 commits October 4, 2022 16:21

Not practical average

3b41b44

fix a bug

ba3a865

TyrantLucifer reviewed Oct 5, 2022

View reviewed changes

remove split rowNum

6d4549d

[Feature][Connector-V2] Improve class attribute

2d910be

TyrantLucifer reviewed Oct 5, 2022

View reviewed changes

liugddx added 4 commits October 5, 2022 18:27

fine-tuning

41b163e

Merge remote-tracking branch 'origin/improve-supportparallelism' into…

bb0d1df

… improve-supportparallelism

Merge remote-tracking branch 'upstream/dev' into improve-supportparal…

81abbad

…lelism

Merge remote-tracking branch 'upstream/dev' into improve-supportparal…

1f716b4

…lelism

TyrantLucifer changed the title ~~feature Support more than splits and parallelism for fake connector~~ [Feature][Connector-V2][FakeSource] Support mutil splits for fake source connector Oct 5, 2022

TyrantLucifer reviewed Oct 5, 2022

View reviewed changes

Remove useless blank line.

a3092b3

TyrantLucifer previously approved these changes Oct 6, 2022

View reviewed changes

EricJoy2048 reviewed Oct 6, 2022

View reviewed changes

...tor-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/config/FakeConfig.java Show resolved Hide resolved

liugddx requested review from EricJoy2048 and TyrantLucifer and removed request for EricJoy2048 and TyrantLucifer October 8, 2022 04:59

Merge remote-tracking branch 'upstream/dev' into improve-supportparal…

542b3ae

…lelism

Merge remote-tracking branch 'upstream/dev' into improve-supportparal…

9c4add3

…lelism

support STREAMING

b991a6b

EricJoy2048 previously approved these changes Oct 12, 2022

View reviewed changes

liugddx and others added 2 commits October 12, 2022 23:26

[feature][connector][fake] Support mutil-splits for fake source

40da144

[connector][fake] Make sure the bounded flow is correct

c2e8e0b

ashulin dismissed EricJoy2048’s stale review via c2e8e0b October 12, 2022 16:23

ashulin force-pushed the improve-supportparallelism branch from b991a6b to c2e8e0b Compare October 12, 2022 16:23

ashulin changed the title ~~[Feature][Connector-V2][FakeSource] Support mutil splits for fake source connector~~ [feature][connector][fake] Support mutil splits for fake source connector Oct 12, 2022

ashulin previously approved these changes Oct 13, 2022

View reviewed changes

hailin0 reviewed Oct 13, 2022

View reviewed changes

liugddx dismissed ashulin’s stale review via adde6d4 October 13, 2022 04:24

liugddx and others added 3 commits October 13, 2022 12:24

Update seatunnel-connectors-v2/connector-fake/src/main/java/org/apach…

adde6d4

…e/seatunnel/connectors/seatunnel/fake/source/FakeSourceReader.java Co-authored-by: hailin0 <hailin088@gmail.com>

Merge remote-tracking branch 'upstream/dev' into improve-supportparal…

b8cc6fa

…lelism

EricJoy2048 approved these changes Oct 13, 2022

View reviewed changes

Hisoka-X approved these changes Oct 14, 2022

View reviewed changes

Hisoka-X merged commit c28c44b into apache:dev Oct 14, 2022

liugddx deleted the improve-supportparallelism branch October 14, 2022 02:42


		public FakeSourceReader(SingleSplitReaderContext context, FakeDataGenerator randomData) {
		public FakeSourceReader(SourceReader.Context context, FakeDataGenerator randomData) {

	public FakeSourceSplitEnumerator(SourceSplitEnumerator.Context<FakeSourceSplit> enumeratorContext, Integer totalRowNum) {
	public FakeSourceSplitEnumerator(SourceSplitEnumerator.Context<FakeSourceSplit> enumeratorContext, int totalRowNum) {

	private final Integer totalRowNum;
	private final int totalRowNum;

	return new FakeSourceSplitEnumerator(enumeratorContext, fakeConfig.getRowNum());
	return new FakeSourceSplitEnumerator(enumeratorContext, fakeConfig);

	List<SeaTunnelRow> seaTunnelRows = fakeDataGenerator.generateFakedRows(split.getRowNum());
	List<SeaTunnelRow> seaTunnelRows = fakeDataGenerator.generateFakedRows();

	allSplit.add(new FakeSourceSplit(rowNum, i));
	allSplit.add(new FakeSourceSplit(i));


		private final SourceReader.Context context;
		private final Deque<FakeSourceSplit> splits = new LinkedList<>();

	private final Deque<FakeSourceSplit> splits = new LinkedList<>();
	private final Queue<FakeSourceSplit> splits = new LinkedList<>();

[feature][connector][fake] Support mutil splits for fake source connector #2974

[feature][connector][fake] Support mutil splits for fake source connector #2974

Conversation

liugddx commented Oct 2, 2022

Purpose of this pull request

Check list

Choose a reason for hiding this comment

TyrantLucifer Oct 5, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

liugddx commented Oct 5, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

liugddx commented Oct 5, 2022

TyrantLucifer Oct 5, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

TyrantLucifer left a comment

Choose a reason for hiding this comment

EricJoy2048 commented Oct 6, 2022

liugddx commented Oct 6, 2022

EricJoy2048 commented Oct 7, 2022 • edited Loading

EricJoy2048 commented Oct 7, 2022

liugddx commented Oct 8, 2022

EricJoy2048 commented Oct 8, 2022

EricJoy2048 commented Oct 9, 2022

liugddx commented Oct 10, 2022

EricJoy2048 left a comment

Choose a reason for hiding this comment

liugddx commented Oct 12, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

EricJoy2048 left a comment

Choose a reason for hiding this comment

hailin0 commented Oct 13, 2022

TyrantLucifer Oct 5, 2022 •

edited

Loading

TyrantLucifer Oct 5, 2022 •

edited

Loading

EricJoy2048 commented Oct 7, 2022 •

edited

Loading