[FLINK-39938] FLIP-586: Composable Parallelism Alignment Modes for Flink Autoscaler

[Dennis-Mircea]

What is the purpose of the change

The autoscaler currently hard-codes how it aligns a vertex's computed parallelism to the number of key groups or source partitions, using a single boolean adjust mode and an inline two-phase search inside JobVertexScaler. This change replaces that with a small set of named, composable alignment modes and an extensible plugin SPI, so the behavior is easy to reason about and operators can plug in their own alignment strategy without forking the autoscaler. Deployments that pinned the deprecated adjust mode keep their exact previous behavior, including the ScalingLimited event.

Brief change log

• Added a new org.apache.flink.autoscaler.alignment package: the @Experimental AlignmentMode SPI (an isApplicable gate plus an align method), AlignmentContext (a plain immutable inputs object exposing parallelism, partitions, ship strategies, the autoscaler context, evaluated metrics, topology, and a per-mode configuration), BuiltInAlignmentMode (BALANCED default, EVENLY_SPREAD, OFF), the deprecated KeyGroupOrPartitionsAdjustMode that preserves the old blocking behavior, and ParallelismAligner which holds the shared per-direction region search, the terminal fallbacks, mode resolution, and event emission.
• Reworked JobVertexScaler so it no longer hard-codes the source/keyBy gate or the two-phase search. It now computes the clamped target parallelism and delegates to ParallelismAligner, which resolves the configured mode and applies it (or keeps the target when the mode does not apply).
• Added the scaling.alignment.mode option (default BALANCED), plus the per-mode scaling.alignment.mode.<name>.class and scaling.alignment.mode.<name>.<parameter> keys. The old scaling.key-group.partitions.adjust.mode is deprecated, retained for backward compatibility, and superseded by the new option.
• Made custom modes discoverable as plugins, mirroring the scaling-executor plugin SPI: ServiceLoader in the standalone autoscaler and Flink's PluginManager in the operator, selected by name and class.
• Threaded the discovered modes from startup (AutoscalerFactory in the operator, StandaloneAutoscalerEntrypoint in standalone) through ScalingExecutor into JobVertexScaler.
• Added an AlignmentMode plugin section to plugins.md and its Chinese mirror, and regenerated the autoscaler configuration reference.

Verifying this change

This change added tests and can be verified as follows:

• AlignmentModeTest covers the built-in single-region search per scaling direction, the keep-the-target fallback, the legacy blocking behavior with its ScalingLimited event, applicability gating for non source / non keyBy vertices, and mode resolution (built-in by name, custom by class, deprecated key migration, and the default).
• JobVertexScalerTest pins the deprecated legacy mode so every existing tuple reproduces the previous behavior exactly.
• Discovery is verified in both runtimes: the standalone test loads a custom mode via ServiceLoader, and the operator test packages a custom mode into a plugin JAR and loads it via PluginManager.
• The regenerated configuration reference is checked by ConfigOptionsDocsCompletenessITCase.

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 changes to the CustomResourceDescriptors: no
• Core observer or reconciler logic that is regularly executed: yes (the autoscaler scaling decision path, with the previous behavior preserved for existing configurations)

Documentation

• Does this pull request introduce a new feature? yes
• If yes, how is the feature documented? docs and JavaDocs

[gyfora]

Looks nice overall, but. a lot of code and new interfaces so let's take it a step at a time :)