Copilot commented on code in PR #29662:
URL: https://github.com/apache/flink/pull/29662#discussion_r4522988631
##########
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:
With chaining disabled, a source and its `forward()` map configured to 2
fail when only the map is overridden to `-1`: this comparison treats the
sentinel as a conflicting size and the subsequent check rejects the frozen
source. Treat `PARALLELISM_DEFAULT` as an unset constraint, inheriting the
group's remaining configured or running parallelism while allowing groups
without such constraints to become undecided.
##########
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:
With chaining disabled, a source and its `forward()` map both configured to
2 cannot both be overridden to 4: creating the source rejects the map's
original value of 2 before its override is applied. Resolve the effective
overrides of pending forward-group members before declaring a conflict, and add
a regression test for consistently overriding the whole group.
##########
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:
In incremental generation, a keyed map explicitly configured with
parallelism 1 and max parallelism 8 fails when overridden to 4: upstream
partition sizing reads its old parallelism and creates only one subpartition
before this override is applied. This fails even below the configured maximum,
not just above the documented sizing limit. Make the effective consumer
override available before upstream partition sizing and add coverage for
explicitly configured targets.
--
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]
