polyzos commented on code in PR #29662:
URL: https://github.com/apache/flink/pull/29662#discussion_r4548483360
##########
flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/AdaptiveGraphManager.java:
##########
@@ -447,6 +473,88 @@ private void
connectToFinishedUpStreamVertex(JobVertexBuildContext jobVertexBuil
}
}
+ /**
+ * An overridden parallelism decides the forward group of its vertex like
a configured one. A
+ * group that is still undecided takes the value. A group that was decided
to another value is
+ * re-decided when the other members only carried that value by
propagation; a member that the
+ * user configured to another value or that already runs as a job vertex
is a conflict, as a
+ * differently configured member is for the full generator.
+ */
+ private void applyParallelismOverridesToForwardGroups(
+ JobVertexBuildContext jobVertexBuildContext) {
+ for (JobVertex jobVertex :
jobVertexBuildContext.getJobVerticesInOrder().values()) {
+ if
(!jobVertexBuildContext.getParallelismOverrides().containsKey(jobVertex.getID()))
{
+ continue;
+ }
+ // createJobVertex has already applied and validated the override
+ final int overriddenParallelism = jobVertex.getParallelism();
+ final StreamNodeForwardGroup forwardGroup =
+ getStreamNodeForwardGroupByVertexId(jobVertex.getID());
+ if (forwardGroup == null) {
+ continue;
+ }
+ if (forwardGroup.isMaxParallelismDecided()
+ && overriddenParallelism >
forwardGroup.getMaxParallelism()) {
+ throw new IllegalStateException(
+ String.format(
+ "The parallelism override %d for job vertex %s
(%s) exceeds the max"
+ + " parallelism %d of its forward
group.",
+ overriddenParallelism,
+ jobVertex.getName(),
+ jobVertex.getID(),
+ forwardGroup.getMaxParallelism()));
+ }
+ if (!forwardGroup.isParallelismDecided()) {
+ forwardGroup.setParallelism(overriddenParallelism);
+ continue;
+ }
+ if (forwardGroup.getParallelism() == overriddenParallelism) {
+ continue;
+ }
Review Comment:
Valid for adaptive batch. Treating `-1` as no constraint here, or rejecting
it for the incremental generator, would fix it.
##########
flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/AdaptiveGraphManager.java:
##########
@@ -447,6 +473,88 @@ private void
connectToFinishedUpStreamVertex(JobVertexBuildContext jobVertexBuil
}
}
+ /**
+ * An overridden parallelism decides the forward group of its vertex like
a configured one. A
+ * group that is still undecided takes the value. A group that was decided
to another value is
+ * re-decided when the other members only carried that value by
propagation; a member that the
+ * user configured to another value or that already runs as a job vertex
is a conflict, as a
+ * differently configured member is for the full generator.
+ */
+ private void applyParallelismOverridesToForwardGroups(
+ JobVertexBuildContext jobVertexBuildContext) {
+ for (JobVertex jobVertex :
jobVertexBuildContext.getJobVerticesInOrder().values()) {
+ if
(!jobVertexBuildContext.getParallelismOverrides().containsKey(jobVertex.getID()))
{
+ continue;
+ }
+ // createJobVertex has already applied and validated the override
+ final int overriddenParallelism = jobVertex.getParallelism();
+ final StreamNodeForwardGroup forwardGroup =
+ getStreamNodeForwardGroupByVertexId(jobVertex.getID());
+ if (forwardGroup == null) {
+ continue;
+ }
+ if (forwardGroup.isMaxParallelismDecided()
+ && overriddenParallelism >
forwardGroup.getMaxParallelism()) {
+ throw new IllegalStateException(
+ String.format(
+ "The parallelism override %d for job vertex %s
(%s) exceeds the max"
+ + " parallelism %d of its forward
group.",
+ overriddenParallelism,
+ jobVertex.getName(),
+ jobVertex.getID(),
+ forwardGroup.getMaxParallelism()));
+ }
+ if (!forwardGroup.isParallelismDecided()) {
+ forwardGroup.setParallelism(overriddenParallelism);
+ continue;
+ }
+ if (forwardGroup.getParallelism() == overriddenParallelism) {
+ continue;
+ }
+ final Set<Integer> chainedNodeIds =
+ new
HashSet<>(jobVertexToChainedStreamNodeIdsMap.get(jobVertex.getID()));
+ final Set<StreamNode> members = new HashSet<>();
+ for (Integer memberId : forwardGroup.getVertexIds()) {
+ members.add(streamGraph.getStreamNode(memberId));
+ if (chainedNodeIds.contains(memberId)) {
+ continue;
+ }
+ final Integer userConfigured =
userConfiguredParallelisms.get(memberId);
+ if (frozenNodeToStartNodeMap.containsKey(memberId)
+ || (userConfigured != null && userConfigured !=
overriddenParallelism)) {
Review Comment:
Valid, and the case users are most likely to hit. Taking the pending
members' overrides into account before declaring a conflict should fix it.
##########
flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java:
##########
@@ -1166,6 +1190,17 @@ public static JobVertex createJobVertex(
jobVertex.setMaxParallelism(streamNode.getMaxParallelism());
+ if (JobVertexParallelismOverrides.apply(
+ jobVertex, jobVertexBuildContext.getParallelismOverrides())) {
+ parallelism = jobVertex.getParallelism();
Review Comment:
Valid. Looking up the consumer's override when the producer's outputs are
created would fix it; otherwise the documented limit should be the configured
parallelism.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
