[FLINK-37232][runtime] Fix for broken synchronization assumption on the AdaptiveScheduler's side introduced by FLIP-272

[ztison]

What is the purpose of the change

The PR fixes the issue with the rescale state transition short cut that was introduced in FLIP-472 where the WaitingForRequirements state is omitted and the transition goes directly from the Restarting state to the CreatingExecutionGraph state.

Brief change log

• Pass the available VertexParallelism that lead to the rescale decision to the Restarting state and check when the job is cancelled whether that parallelism has changed.

Verifying this change

This change is already covered by existing tests, such as (please describe tests).

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: no
• 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

[flinkbot]

CI report:

• 78855d9 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[XComp]

Looks good. CI is failing due to spotless.

./mvnw -pl flink-runtime spotless:apply should do the trick.

[XComp]

Thanks for going through my comments. Looks good from my end. 👍

[ztison]

@XComp Thanks for review, I addressed the comments and updated the description. PTAL.

[ztison]

@flinkbot run azure

[davidradl]

@ztison looks like you bot command did not take- we still see yesterdays CI FAILURE

[XComp]

@flinkbot run azure

[XComp]

@flinkbot run azure

[XComp]

Good job and thanks for adding the ITCase. 👍

I verified that we can reproduce the error with the ITCase and the changes not being applied:

StandaloneDispatcher [flink-pekko.actor.default-dispatcher-4] INFO  Job 98f97effc400a2ee5004f62ca5c4cea4 reached terminal state FAILED.
org.apache.flink.runtime.jobmanager.scheduler.NoResourceAvailableException: Not enough resources available for scheduling.
	at org.apache.flink.runtime.scheduler.adaptive.AdaptiveScheduler.lambda$determineParallelism$26(AdaptiveScheduler.java:1136)

I have one question, though. PTAL

[XComp]

Nothing else to add. Thanks for fixing the issue. 👍 Please provide backports for release-2.0 and release-2.0-preview1-rc1