StefanRRichter commented on code in PR #22864: URL: https://github.com/apache/flink/pull/22864#discussion_r1242128808
########## flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java: ########## @@ -1008,6 +1008,23 @@ private StreamConfig createJobVertex(Integer streamNodeId, OperatorChainInfo cha jobVertex.setInvokableClass(streamNode.getJobVertexClass()); + // Find the minimal max parallelism in the chain for operators that have a configured max + // parallelism (>0). If no operator has max parallelism configured, returns the default -1. + int chainMinimalMaxParallelism = + chainInfo.getAllChainedNodes().stream() + .mapToInt(StreamNode::getMaxParallelism) + .filter(maxParallelism -> maxParallelism > 0) + .min() + .orElse(-1); + + jobVertex.setMaxParallelism(chainMinimalMaxParallelism); + + boolean anyParallelismInChainConfigured = + chainInfo.getAllChainedNodes().stream() + .anyMatch(StreamNode::isParallelismConfigured); + + jobVertex.setParallelismConfigured(anyParallelismInChainConfigured); Review Comment: If that was true und the chain incomplete, would that not mean that the current implementation that determines if parallelism is configured would also be wrong? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org