[FLINK-3020][streaming] set number of task slots to maxmium parallelism in local execution

[mxm]

No description provided.

[tillrohrmann]

I think the maximum parallelism of the job should be taken if the user has not specified a different parallelism than the default one (-1). I think that's also how the batch part does it (here the LocalExecutor has a field taskManagerNumSlots which can be set). Besides that, looks good to merge.

[mxm]

True. That's how it is handled on the batch side. Not sure about this behavior though. If a user sets a default parallelism but uses operators with parallelism > defaultParallelism this would fail, right? The rational behind this is probably to maximize the parallelism for all operators and not have operators with exceptional high parallelism.

[tillrohrmann]

Yes, in such a case it would fail. Thinking about it, you're right that it would give a better user experience if the maximum degree of the job is taken instead of the default parallelism. Maybe we should change it then for the batch part as well.

[mxm]

I would also be in favor of changing the local execution to always use the maximum specified parallelism as the number of task slots. IMHO the current behavior is not intuitive. The default parallelism currently acts as a maximum parallelism in local execution.

[StephanEwen]

+1 for taking the max parallelism of all operators

[StephanEwen]

This currently means that in the default case, the slots do not respect the max parallel operator, correct?

[tillrohrmann]

defaultLocalParallelism seems to be the number of available processors. I think this is very unintuitive that the default parallelism is only taken if it equals the number of processors.

This means in case of an operator with a higher dop than the #cores, the program will only fail if the user sets the default parallelism to #cores.

[mxm]

Alright. I will push the original pull request version again which uses the max parallelism of all operators. Further, I will open a separate JIRA for the batch side change.

[mxm]

Just checked. The batch side always uses the maximum parallelism as the number of task slots (if they are not set explicitly). Till and me actually thought differently. So the proposed changes in this PR align with the batch behavior.

[tillrohrmann]

But this is only true if the taskManagerNumSlots in LocalExecutor are left untouched. Currently, this is the case but this is not enforced.

[mxm]

Exactly. But that makes sense, right? If the users explicitly sets the number of task slots, we shouldn't change the number of task slots automatically.

[tillrohrmann]

Yes I think so. I guess it's due to the difference that batch programs are executed using a PlanExecutor whereas streaming programs are directly executed by the StreamExecutionEnvironment. Consequently, there is no direct way of specifying the number of task slots for the used LocalFlinkMiniCluster other than via the parallelism of the job or the default parallelism.

[mxm]

I think we agree that we want to set the number of task slots to the maximum parallelism instead of the default one. I'll merge this later on.