[
https://issues.apache.org/jira/browse/FLINK-40389?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-40389:
-----------------------------------
Labels: pull-request-available (was: )
> 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
> Priority: Major
> Labels: pull-request-available
> 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 a drop onto the alignment cap as a direction inversion*
> ({{ParallelismAligner#invertsDirection}}). Landing exactly on the cap is the
> 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. Only downwards though, since {{relaxedDownwardFallback}} clamps up
> to {{parallelismLowerLimit}} and can land on a cap that sits above the
> current parallelism, which would turn a requested scale-down into a scale-up:
> {code:java}
> if (candidate == upperBoundForAlignment(ctx) && candidate <
> ctx.getCurrentParallelism()) {
> 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}
> The cap is a no-op for keyed vertices. {{JobVertexScaler#scale}} clamps the
> target to {{min(maxParallelism, parallelismUpperLimit)}} before calling the
> aligner, so {{target > upperBoundForAlignment}} is only reachable when {{N}}
> is the binding term, and for a keyed vertex {{N == maxParallelism}}. Sources
> are the only case where {{N}} is independent of the clamp.
> Note that the second change only affects scale-ups. On a scale-down the
> candidate is already below the current parallelism, so it was never read as
> an inversion, which is why the legacy modes handled the scale-down rows of
> the table correctly to begin with.
> h1. Verification
> The full {{flink-autoscaler}} suite passes: 289 tests, 257 pre-existing plus
> 32 added. Each new test was confirmed to fail when its corresponding fix is
> reverted in isolation:
> ||reverted||failing tests||
> |Built-in clamp|the end-to-end {{scale()}} case under {{BALANCED}}, plus 8
> aligner cases covering both built-in modes at targets 180, 50 and 16|
> |Inversion exemption|the end-to-end {{scale()}} case under the legacy mode,
> plus 4 aligner cases covering both legacy modes on the scale-up|
> Coverage is a mode x direction matrix over all four aligning modes
> ({{BALANCED}}, {{EVENLY_SPREAD}}, and both legacy modes), each combination a
> separate parameterized invocation so no leg can be masked by an earlier
> assertion:
> * target above the cap on a scale-up (180) and on a scale-down (50, 16), and
> a target already on the cap (15)
> * {{OFF}} keeping the computed target in both directions
> * a keyed vertex ({{numSourcePartitions == 0}}) capped at its key group count
> * the legacy {{SCALING_LIMITED}} event emitted for every capped target
> * the pre-existing FLINK-39299 inversions still blocked: the scale-up case at
> 22, a scale-down whose fallback is clamped onto the cap, and a scale-down
> whose fallback is clamped below the cap
> {{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)
