Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

The Flink Autoscaler computes a target parallelism for each vertex and then aligns it to a divisor of the vertex's key-group count (for keyBy / HASH inputs) or its source-partition count, so that subtasks receive an even share of key groups or partitions and data skew is avoided.

The current alignment has four limitations.

1. Misleading naming: The EVENLY_SPREAD value does not strictly enforce divisor alignment despite its name.
2. Hard-coded in the scaler: The applicability gate and the search are inlined in JobVertexScaler, which makes the behavior hard to read and to change.
3. Blocking-only outcome: When no aligned value is found, the only behavior is to hold at the current parallelism and emit a SCALING_LIMITED event, even when scaling to the computed target would be fine.
4. Not extensible: Adding a strategy means editing core search logic, and operators have no way to supply their own alignment for cases the built-ins do not cover, such as custom partitioning.

The goal is to make alignment a small set of named, composable modes behind one configuration key, and to make it extensible through a discoverable plugin, without complicating the everyday configuration surface.

Proposed Changes

Alignment becomes a small set of named modes selected by one configuration key, plus a plugin SPI for custom modes. Each mode owns its own search loop and decides whether it applies to a given vertex.

Built-in modes

A single mode value, selected by name through scaling.alignment.mode. Three built-in values cover the everyday cases.

The new modes never block a scale. When no value is accepted in the search region they keep the autoscaler's computed target rather than holding at the current parallelism.

The alignment search

Each built-in mode scans a single region per scaling direction.

• Scale-up scans the above-target region, from the target up to the alignment cap.
• Scale-down scans the within-range region, between the target and the current parallelism.

BALANCED accepts the first parallelism that reduces per-subtask load, snapping to an exact divisor when one is within reach. EVENLY_SPREAD accepts only an exact divisor. OFF does not search. When the region yields nothing, the mode returns the computed target.

The split between the two regions is deliberate, and it shapes the default behavior. On a scale-up the search sits above the target, so the default BALANCED mode is as aggressive as the bounds allow in raising parallelism to clear the higher load quickly. On a scale-down the search stays within the range between the target and the current parallelism, so it relaxes the reduction rather than pushing for the smallest footprint. The default therefore favors responsiveness to load spikes over resource reduction.

Figure 1 shows the regions on the parallelism number line and the acceptance rule for each mode, plus how the new keep-the-target outcome differs from the deprecated blocking outcome.

Applicability gate

Each mode decides whether it applies to a vertex. The built-in modes apply to keyBy (hash) vertices and to source vertices that report a partition count, which the Kafka and Pulsar sources do by default. A custom mode can widen this, for example to align custom-partitioned vertices. When a mode does not apply, the autoscaler keeps the computed target.

Custom modes through a plugin SPI

Custom modes are discovered as plugins, using a ServiceLoader in the standalone autoscaler and Flink's PluginManager in the operator, and are selected by name. The implementation class is given by scaling.alignment.mode.<name>.class, and arbitrary parameters are passed to the mode through scaling.alignment.mode.<name>.<parameter>. This follows the same convention as the autoscaler's Scaling Executor Plugin SPI (FLIP-575, PR #1085). The two seams are complementary, because alignment is the per-vertex skew-avoidance step inside the parallelism computation while the Scaling Executor Plugin SPI intercepts the final scaling decisions.

Code structure

The alignment logic moves out of JobVertexScaler into a dedicated org.apache.flink.autoscaler.alignment package. JobVertexScaler computes the clamped target parallelism and delegates to ParallelismAligner, which resolves the configured mode and applies it. The package holds the AlignmentMode SPI, the AlignmentContext inputs object, the BuiltInAlignmentMode enum, the deprecated KeyGroupOrPartitionsAdjustMode, and ParallelismAligner, which owns the shared region search, the terminal fallbacks, the mode resolution, and the legacy event emission.

Public Interfaces

Configuration options

The options live under a neutral scaling.alignment namespace. The feature aligns parallelism to key groups or source partitions equally (the internal variable is numKeyGroupsOrPartitions), so the neutral name fits better than a key-group-specific one.

The mode key is a String rather than an enum, because the value can name a built-in mode (BALANCED, EVENLY_SPREAD, OFF) or a custom mode. A value that matches a built-in name selects that built-in. Any other value is treated as a custom mode name, whose class is read from the matching .class key and matched against the discovered plugins. The name is matched exactly, so a custom name only needs to avoid the three reserved built-in names.

AlignmentMode SPI (experimental)

The modes are expressed through a new @Experimental interface, org.apache.flink.autoscaler.alignment.AlignmentMode, with two methods.

• isApplicable(AlignmentContext) returns whether the mode acts on the vertex. It defaults to keyBy (hash) vertices and to source vertices that report a partition count (the Kafka and Pulsar sources do by default), and align is called only when it returns true.
• align(AlignmentContext) returns the parallelism to apply.

AlignmentContext is a plain immutable inputs object (no Lombok, so the SPI stays legible in the shipped source). It exposes the current and computed target parallelism, the partition and key-group counts, the input ship strategies, the full autoscaler context, the per-vertex evaluated metrics, the job topology, and a prefix-stripped configuration for the selected mode. Built-in modes are resolved by name directly. Custom implementations are discovered as plugins, as described above. Unlike an earlier revision of this proposal, the plugin discovery and selection ship in this FLIP rather than being deferred.

Compatibility, Deprecation, and Migration Plan

Jobs that pinned the deprecated configuration keep their exact previous behavior. The everyday default changes deliberately, as agreed during review, to the non-blocking BALANCED mode.

• Deprecation: The previous key job.autoscaler.scaling.key-group.partitions.adjust.mode and its enum (now relocated as KeyGroupOrPartitionsAdjustMode) are marked deprecated. The enum is retained so the deprecated key still parses, and it preserves the original blocking behavior together with the SCALING_LIMITED event.
• Precedence: The new key is read first. Only when it is unset is the deprecated key honored, at lower precedence. When neither is set, the new default BALANCED applies.
• No silent remap: The deprecated values are not remapped onto the new same-named modes, because the new modes never block while the legacy ones do. A pinned legacy value therefore keeps running the legacy mode, not the new one.

The only behavior change for an unset configuration is that alignment no longer holds a scale at the current parallelism when it cannot find an aligned value. It keeps the computed target instead.

Test Plan

The implementation is covered by unit tests in the flink-autoscaler, flink-autoscaler-standalone, and flink-kubernetes-operator modules at the following levels.

• Per-mode correctness: Each built-in mode's region search for both scale-up and scale-down, across a range of key-group and source-partition counts (primes, semi-primes, and highly composite numbers), so that divisor and non-divisor targets are both exercised.
• Keep-the-target behavior: When no aligned value is found, the new modes return the computed target and emit no event.
• Backward compatibility: The deprecated modes reproduce the previous blocking behavior and the SCALING_LIMITED event exactly, pinned through the existing scaler tests.
• Applicability gating: The built-in modes do not act on vertices that are neither source nor keyBy.
• Resolution and migration: Selection of a built-in by name, a custom mode by class, the precedence of the new key over the deprecated key, and the default.
• Plugin discovery: Custom modes are discovered through a ServiceLoader in the standalone autoscaler and through the PluginManager in the operator.
• Documentation completeness: The generated autoscaler configuration reference stays in sync with the options.

Rejected Alternatives

• Flat mode-times-fallback cross-product (the earlier shape). Rejected because it exposes a large, unguarded cross-product in which many combinations are redundant or no-ops, which is hard to reason about. The three named modes plus a validated advanced door keep the expressiveness without the flat surface.
• Redefining EVENLY_SPREAD in place to mean strict-divisor-only. Rejected because a user who pinned EVENLY_SPREAD would silently switch from scaling to scaling-blocked. The token is retired instead, and pinned values keep their original behavior.
• Three fixed modes with no advanced door. Rejected because it drops real expressiveness, including the OPTIMIZE_RESOURCE_UTILIZATION strategy and useful compositions such as a cheap in-range divisor first, then an upward spread. ADVANCED preserves the complete model for the users who need it.
• Shipping a custom-strategy class loader now. Rejected for this FLIP. A class-from-config loader is a new, unprecedented mechanism in the autoscaler module and re-opens the complexity concern. The @Experimental AlignmentStrategy seam ships now, and a runtime-agnostic loader can be a future FLIP that follows the META-INF/services ServiceLoader plugin convention being introduced for the autoscaler's Scaling Executor Plugin SPI (FLIP-575, PR #1085). That seam is complementary, because alignment is the per-vertex skew-avoidance step inside the parallelism computation while the Scaling Executor Plugin SPI intercepts the final scaling decisions.
• A key-group-alignment key name. Rejected because the feature aligns parallelism to source partitions as well as key groups, so the neutral scaling.alignment name is more accurate and matches the AlignmentMode and AlignmentStrategy types.
• Offering STRICT_DIVISOR as a fallback. Rejected because a fallback stricter than the primary can never change the result, so it would only add an inert option.