Thanks @joexjx . But in seatunnel, we recommend configuring parallelism in env rather than source configuration. This helps improve configuration consistency across different engines. Maybe you can help update the documentation to make the description clearer.

I understand. However, this code does contain a very simple and obvious error. envOption is a HashMap that already includes EnvCommonOptions.PARALLELISM.key(). The .option(EnvCommonOptions.PARALLELISM.key(), parallelism) call is being overwritten by .options(envOption), which is why the properly configured parallelism (determined earlier based on whether it comes from env or source) isn't taking effect.

Old Code

New Code

I removed .option(EnvCommonOptions.PARALLELISM.key(), ...) and updated envOptions with EnvCommonOptions.PARALLELISM in advance. This ensures that the parallelism value (which was already determined earlier based on whether it should be taken from env or source) takes effect correctly.

The current test case is a bit overly complex. We just need to ensure that the DataSet returned by the execute method of SourceExecuteProcessor has the expected parallelism.

SourceExecuteProcessor processor = new SourceExecuteProcessor();
... init

Assertions.equals(10, processor.execute(xxx).get(0).getDataset().rdd().getNumPartitions());

@Hisoka-X Thank you for the guidance, I have added a simpler unit test.