setParallelism() for addSource() in streaming

[Hung]

Hi Flink users,

I'd like to setup different parallelisms for different sources.

I found we cannot
DataStream<String> bigStream = env.addSource(
                new FooSource(properties, bigTopic)
        ).setParallelism(5)..rebalance();

DataStream<String> smallStream = env.addSource(
                new FooSource(properties, smallTopic)
        ).setParallelism(1)..rebalance();
   
        env.setParallelism(3);
//do .map(), window(), ...

But I can set the following:

DataStream<String> bigStream = env.setParallelism(5).addSource(
                new FooSource(properties, bigTopic)
        ).rebalance();

DataStream<String> smallStream = env.setParallelism(1).addSource(
                new FooSource(properties, smallTopic)
        ).rebalance();

        env.setParallelism(3);
//do .map(), window(), ...

Would it have the same effect?

Best,

Sendoh

[Jonas Gröger]

env.setParallelism(5).addSource(???) will set the default parallelism for this Job to 5 and then add the source.