[FLINK-38770][runtime] Apply jobvertex parallelism overrides to StreamGraph submissions by MartijnVisser · Pull Request #29662 · apache/flink

MartijnVisser · 2026-10-28T17:16:41Z

What is the purpose of the change

pipeline.jobvertex-parallelism-overrides is applied by the Dispatcher only to a submitted JobGraph. Since FLINK-36065 clients submit a StreamGraph, so on 2.x the option has no effect; the operator autoscaler writes its decisions into it. Verified with a probe job: 1.20.5 applies it in session and application mode, 2.3.0 does not; git bisect names 64cbcd2.

The Dispatcher writes the resolved map into the submitted plan's job configuration. A StreamGraph persisted in HA before this fix recovers without overrides, as a JobGraph does today.

The adaptive batch scheduler logs a warning and does not apply the map to a StreamGraph it generates incrementally; fixing that is a follow-up.

Supersedes #27334, which moved the overrides into the scheduler factories; the diagnosis and the env.execute() test idea are Barak's.

Brief change log

Resolve the map once in the Dispatcher (JobVertexParallelismOverrides) and store it in a StreamGraph's job configuration
StreamingJobGraphGenerator.createJobVertex applies it after the vertex id is computed
Check the values of the job's own entries at submission for both plan types; reject an override above a configured max parallelism with a message that names the option
Adaptive batch: warn instead of applying the map to an incrementally generated StreamGraph
Option description updated, docs regenerated

Verifying this change

This change added tests and can be verified as follows:

Commit 1 adds the tests listed in its message; they fail on master and pass with commit 2
ForwardEdgeParallelismMismatchOverridesITCase now also submits StreamGraphs
Probe job on a dist built from this branch (uid on the map, override to 4): GET /jobs/<id> shows 4 and the job master logs Changing job vertex ... parallelism from 1 to 4, in session and application mode, as on 1.20.5

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): no
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? no
If yes, how is the feature documented? not applicable

Was generative AI tooling used to co-author this PR?

Yes (please specify the tool below)

Generated-by: Claude Code (Claude Fable 5.1, Claude Opus 5.5)

flinkbot · 2026-10-28T17:28:43Z

CI report:

77b3fcb Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

Copilot

🟡 Changes recommended

Valid parallelism overrides still fail because forward-group validation mishandles pending overrides and default sentinels, while incremental partition sizing uses stale consumer parallelism.

3 open issues

What changed in this PR

This PR restores job-vertex parallelism overrides for StreamGraph submissions, addressing the submission-path change that left autoscaler settings ineffective in Flink 2.x.

Changes:

Resolve cluster and job overrides in the Dispatcher and preserve them in submitted StreamGraphs.
Apply overrides during graph generation, reconcile adaptive forward groups, and improve diagnostics.
Add regression coverage for submission paths, schedulers, failover, and savepoint restore; update configuration documentation.

File	Description
flink-tests/src/test/java/org/apache/flink/test/scheduling/ParallelismOverridesTestUtils.java	Shared cluster setup and vertex-ID helpers.
flink-tests/src/test/java/org/apache/flink/test/scheduling/ParallelismOverridesJMFailoverITCase.java	Tests overrides across JobMaster failover.
flink-tests/src/test/java/org/apache/flink/test/scheduling/ParallelismOverridesITCase.java	Tests configuration precedence and forward groups across schedulers.
flink-tests/src/test/java/org/apache/flink/test/scheduling/ExecutionPlanSubmissionParityITCase.java	Compares submission paths and captures a sizing limitation.
flink-tests/src/test/java/org/apache/flink/test/checkpointing/ParallelismOverridesSavepointITCase.java	Tests savepoint restoration with overridden parallelism.
flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorParallelismOverridesTest.java	Exercises the full graph generator.
flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/ParallelismOverridesGeneratorTestBase.java	Shared topology-preservation and validation tests.
flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/AdaptiveGraphManagerParallelismOverridesTest.java	Exercises incremental graph generation.
flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobVertexParallelismOverridesTest.java	Tests override merging, parsing, and application.
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherParallelismOverridesTest.java	Tests resolution, persistence, and recovery behavior.
flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/util/JobVertexBuildContext.java	Carries overrides during vertex construction.
flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java	Applies overrides when vertices are created.
flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/AdaptiveGraphManager.java	Reconciles overrides with adaptive forward groups.
flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobVertexParallelismOverrides.java	Centralizes override resolution and application.
flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/forwardgroup/StreamNodeForwardGroup.java	Adds conflicting-parallelism diagnostics.
flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/forwardgroup/JobVertexForwardGroup.java	Adds conflicting-parallelism diagnostics.
flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/VertexInputInfoComputationUtils.java	Clarifies insufficient-subpartition errors.
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java	Resolves overrides for both submitted plan types.
flink-core/src/main/java/org/apache/flink/configuration/PipelineOptions.java	Documents precedence, supported plans, and constraints.
docs/layouts/shortcodes/generated/pipeline_configuration.html	Updates the generated configuration reference.

_{Review effort: Balanced}

polyzos

@MartijnVisser Thanks for tackling this 🙏 It's a regression from 1.x that makes the operator autoscaler ineffective on 2.x, and resolving the map once in the Dispatcher, then applying it while vertices are created, reads well to me. I mostly looked at it from the streaming side, since that's what the autoscaler writes these overrides for.

The Copilot comments only affect adaptive batch. Streaming jobs never use the incremental generator, and the full generator applies every override before any output is sized. If fixing the batch cases grows the PR a lot, failing fast when overrides meet the incremental generator and following up separately would let the streaming fix land first.

Tests. The coverage is thorough. One case I'd add: with the adaptive scheduler, rescale a job in place over REST, then fail over the JobManager. Which wins, the override or the persisted resource requirements? That's the path the operator uses for in-place scaling on 2.x.

Backport. Since this is a regression from FLINK-36065, it'd be worth backporting to the maintained 2.x branches. We run 2.3 with this commit backported; results are in a separate comment.

polyzos · 2026-11-12T09:18:33Z

Some test results for this PR (operator 1.16.1, Flink 2.3.0, application mode, Fluss sources and sinks).

Without the fix (Flink 2.3.0):

Default scheduler: the autoscaler wrote correct overrides, each decision triggered a savepoint restart, and every vertex came back at parallelism 1.
Adaptive scheduler: the in-place rescale worked, but any redeploy (configuration change, restart nonce, suspend and resume) returned the job to its configured parallelism, and the autoscaler stayed stuck until autoscalerResetNonce was changed.

With this PR (commit c7936e92) backported onto release-2.3.0:

Default scheduler: every decision took effect, with a redeploy of about 1.5 to 2 minutes each.
Adaptive scheduler: rescaled in place, and the scaled parallelism survived a restart, a suspend and resume, and spec edits.
Steady-state autoscaling then behaved as expected: no scaling at steady load, a load step absorbed in two steps, and a backlog cleared by a scale-out.

Tested only for streaming SQL jobs, in application mode.

…Graph submissions No test submitted a StreamGraph with pipeline.jobvertex-parallelism-overrides set. These fail on master, except the control cases: - ExecutionPlanSubmissionParityITCase: the same program as StreamGraph and as JobGraph yields the same vertices at the same parallelism on the default and the adaptive scheduler, with and without overrides. - ParallelismOverridesITCase: env.execute() with the map in the cluster configuration, the job configuration and both. - StreamingJobGraphGeneratorParallelismOverridesTest: over random programs an override changes exactly the overridden vertices; ids, operator ids, chaining and edges stay the same. - ParallelismOverridesJMFailoverITCase and ParallelismOverridesSavepointITCase: the override survives a job master failover and a savepoint restore. - ParallelismOverridesRescaleRecoveryITCase: after an in-place rescale through the resource requirements, a JobManager failover keeps the rescaled parallelism instead of the override. - DispatcherParallelismOverridesTest: the Dispatcher attaches the merged map to a StreamGraph, rejects invalid entries of the job configuration and leaves a recovered plan alone. - AdaptiveExecutionHandlerFactoryTest: the adaptive batch scheduler warns that it ignores the map for an incrementally generated StreamGraph. Generated-by: Claude Code (Claude Fable 5.1, Claude Opus 5.5)

…mGraph submissions The Dispatcher applied pipeline.jobvertex-parallelism-overrides only to a submitted JobGraph. Since FLINK-36065 clients submit a StreamGraph, which the job master turns into a JobGraph later, so the option had no effect. - Resolve the map once in the Dispatcher (job configuration over cluster configuration) and store it in the plan's job configuration, so a recovered plan replays it. - Apply it in StreamingJobGraphGenerator.createJobVertex, after the vertex id is computed; ids and chaining do not change. - Check the values of the job configuration at submission for both plan types, and reject an override above the max parallelism configured for the vertex. - Drop entries whose key is not a job vertex id with a warning. - The incremental generator of the adaptive batch scheduler sizes the inputs of a vertex before the vertex id exists, so it does not apply the map and the scheduler logs a warning instead. Generated-by: Claude Code (Claude Fable 5.1, Claude Opus 5.5)

…or StreamGraphs too Generated-by: Claude Code (Claude Opus 5.5)

MartijnVisser · 2026-11-12T11:59:55Z

Thanks for testing this. Adaptive batch warns instead of failing, since a map in the cluster configuration would fail unrelated batch jobs. After an in-place rescale and a failover the persisted requirements win, see ParallelismOverridesRescaleRecoveryITCase. Could you rerun the default scheduler case on the new head?

MartijnVisser force-pushed the FLINK-38770-overrides-streamgraph branch from 39fb5fd to c7936e9 Compare October 28, 2026 17:19

MartijnVisser marked this pull request as ready for review October 29, 2026 07:11

XComp self-assigned this Nov 5, 2026

XComp self-requested a review November 5, 2026 02:23

rmetzger requested a balanced review from Copilot November 10, 2026 08:56

Copilot started reviewing on behalf of rmetzger November 10, 2026 08:59

View session

Copilot AI reviewed Nov 10, 2026

View reviewed changes

github-actions Bot added the community-reviewed PR has been reviewed by the community. label Nov 10, 2026

polyzos reviewed Nov 12, 2026

View reviewed changes

MartijnVisser added 3 commits November 12, 2026 12:53

[FLINK-38770][tests] Run the forward edge mismatch overrides ITCase f…

77b3fcb

…or StreamGraphs too Generated-by: Claude Code (Claude Opus 5.5)

MartijnVisser force-pushed the FLINK-38770-overrides-streamgraph branch from c7936e9 to 77b3fcb Compare November 12, 2026 11:59

MartijnVisser requested a review from polyzos November 12, 2026 12:05

Navigation Menu

Repository navigation

[FLINK-38770][runtime] Apply jobvertex parallelism overrides to StreamGraph submissions - #29662

[FLINK-38770][runtime] Apply jobvertex parallelism overrides to StreamGraph submissions#29662
MartijnVisser wants to merge 3 commits into
apache:masterfrom
MartijnVisser:FLINK-38770-overrides-streamgraph

MartijnVisser commented Oct 28, 2026 •

edited

Loading

Uh oh!

flinkbot commented Oct 28, 2026 •

edited

Loading

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

polyzos left a comment •

edited

Loading

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

polyzos commented Nov 12, 2026 •

edited

Loading

Uh oh!

MartijnVisser commented Nov 12, 2026 •

edited

Loading

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

Conversation

MartijnVisser commented Oct 28, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Was generative AI tooling used to co-author this PR?

Uh oh!

flinkbot commented Oct 28, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

🟡 Changes recommended

Uh oh!

Uh oh!

Uh oh!

Uh oh!

polyzos left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

polyzos commented Nov 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

MartijnVisser commented Nov 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

MartijnVisser commented Oct 28, 2026 •

edited

Loading

flinkbot commented Oct 28, 2026 •

edited

Loading

polyzos left a comment •

edited

Loading

polyzos commented Nov 12, 2026 •

edited

Loading

MartijnVisser commented Nov 12, 2026 •

edited

Loading