[
https://issues.apache.org/jira/browse/FLINK-40389?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Dennis-Mircea Ciupitu updated FLINK-40389:
------------------------------------------
Description:
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.
was:
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.
> 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
> 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)
