Dennis-Mircea Ciupitu created FLINK-40389:
---------------------------------------------
Summary: Source parallelism is not capped at the partition count
Key: FLINK-40389
URL: https://issues.apache.org/jira/browse/FLINK-40389
Project: Flink
Issue Type: Bug
Components: Autoscaler, Kubernetes Operator
Reporter: Dennis-Mircea Ciupitu
Fix For: kubernetes-operator-1.16.0
h1. Summary
A source vertex can be scaled far beyond its partition count. Every subtask
above that count is assigned no split and stays idle, together with anything
chained onto the source, so the slots are paid for and do no work.
Two separate defects combine. FLIP-586 (FLINK-39938) removed the cap for the
new alignment modes, and the legacy modes never corrected the case where the
current parallelism is already above the partition count, because of the
direction guard added in FLINK-39299.
h1. Reproduction
Kafka source with 15 partitions, current parallelism 100, operator max
parallelism 180, {{vertex.max-parallelism}} at its default of 200. Values are
the result of {{{}JobVertexScaler#scale{}}}:
||raw target||new default (BALANCED)||legacy default (EVENLY_SPREAD)||correct||
|180 (scale up)|180|100 (scale blocked)|15|
|50 (scale down)|50|15|15|
|10 (scale down)|15|15|15|
Only the third row, where the target already sits at or below the partition
count, is handled correctly by either.
h1. Cause
h2. New modes: the cap bounds the search, not the result
{{ParallelismAligner#firstAlignedInRegion}} scans {{[target, regionEnd]}} where
{{regionEnd}} is derived from {{{}upperBoundForAlignment = min(N,
maxParallelism, parallelismUpperLimit){}}}. When the target is already above
that region the loop body never executes and the search returns 0. The built-in
modes then keep the unaligned target:
{code:java}
int aligned = ParallelismAligner.firstAlignedInRegion(ctx, acceptLoadReducing);
return aligned > 0 ? aligned : ctx.getNewParallelism();
{code}
So the region is capped at 15 while the returned value is not.
h2. Legacy modes: the direction guard blocks a legitimate correction
The legacy modes fall back to {{{}relaxedDownwardFallback{}}}, which correctly
arrives at 15, then pass it through {{{}applyBlockingFallback{}}}, whose guard
is:
{code:java}
return (isScaleUp(ctx) && candidate <= ctx.getCurrentParallelism())
|| (!isScaleUp(ctx) && candidate >= ctx.getCurrentParallelism());
{code}
FLINK-39299 added this to stop the search drifting a few steps past
{{currentParallelism}} and silently inverting a scale. It cannot distinguish
that from the case here, where the candidate is legitimately far below the
current parallelism because the partition count is a hard ceiling. With current
100 and a scale-up requested, candidate 15 looks like an inversion, so the
scale is blocked and the vertex stays at 100.
h2. Why only sources
For a keyed vertex the alignment target is the key group count, which is
{{{}maxParallelism{}}}, and {{JobVertexScaler#scale}} already clamps to
{{{}min(maxParallelism, parallelismUpperLimit){}}}. The target therefore can
never exceed the ceiling and the defect cannot occur.
A source reports {{numSourcePartitions}} separately from
{{{}maxParallelism{}}}, and nothing clamps the target to it, which is why
sources are the only affected case.
h1. Proposed fix
Two small changes, both inside the alignment package, one per defect.
*1. Do not treat the alignment cap as a direction inversion*
({{{}ParallelismAligner#invertsDirection{}}}). Landing exactly on the cap is
the key group or partition count being the real limit rather than the search
drifting, so correcting down to it is the intended outcome even when a scale-up
was requested:
{code:java}
if (candidate == upperBoundForAlignment(ctx)) {
return false;
}
{code}
*2. Never keep a target above the cap*
({{{}BuiltInAlignmentMode#alignOrKeepTarget{}}}). When the region yields
nothing, keep the computed target but bound it by the cap:
{code:java}
return Math.min(ctx.getNewParallelism(),
ParallelismAligner.upperBoundForAlignment(ctx));
{code}
Verified against the reproduction: every row returns 15 for both the legacy and
the new modes, the FLINK-39299 scale-up inversion case still blocks at 22, and
the full {{flink-autoscaler}} suite passes unchanged (257 tests).
{{OFF}} overrides {{alignParallelism}} directly and is untouched, so explicitly
disabling alignment still uses the computed target as-is.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
