[
https://issues.apache.org/jira/browse/FLINK-40113?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Rion Williams updated FLINK-40113:
----------------------------------
Description:
h2. Summary
[FLINK-31996|https://issues.apache.org/jira/projects/FLINK/issues/FLINK-31996]
allowed operators with different max parallelism to be chained into a single
vertex, and made it the default
({{{}pipeline.operator-chaining.chain-operators-with-different-max-parallelism{}}}).
On restore, Flink validates that an operator's max parallelism matches the
savepoint's and rejects an incompatible change with a clear error but that
validation is per-operator against the vertex's single max parallelism. It does
not verify that the operators now sharing a vertex have compatible max
parallelism with each other. When they don't, the per-vertex reconciliation
silently adopts one operator's value instead of rejecting.
h2. Current behavior
- A single operator whose max parallelism changed since the savepoint →
rejected: "{_}Max parallelism mismatch … changed in a non-compatible way."{_}
(Correct.)
- Multiple operators chained into one vertex with different recorded max
parallelism into {{StateAssignmentOperation#checkParallelismPreconditions}}
whichreconciles each against the shared vertex and adopts one value; operators
whose recorded value differs are accepted under the adopted key-group count
with {*}no equivalent check{*}.
h2. The gap
{{checkParallelismPreconditions}} checks each operator against the vertex's max
parallelism (and {{canRescaleMaxParallelism}} correctly permits rescaling an
auto-derived value), but nothing checks that the operators sharing the vertex
agree with one another.
[FLINK-31996|https://issues.apache.org/jira/projects/FLINK/issues/FLINK-31996]
made this arrangement reachable by default; the restore-side validation was not
extended to cover it. The single-operator path is guarded; the shared-vertex
path is not.
h2. Why it matters
A keyed operator can then be restored under a key-group count different from
the one its state was written with, so its state is mapped through a different
{{hash % maxParallelism}} layout. The restore should reject this, consistent
with how it already rejects the single-operator case, rather than proceeding.
h2. Proposed change (validation-only, low-risk)
Add explicit validation during per-vertex reconciliation in
{{StateAssignmentOperation#checkParallelismPreconditions}}: for any vertex that
carries keyed state, verify that all operators chained into it recorded the
same maximum parallelism in the checkpoint, and fail with a clear error
otherwise. The
comparison spans all chained operators, not only the keyed ones, because a
non-keyed operator's recorded value can win the reconciliation and misroute a
keyed operator's state.
This extends the check
[FLINK-31996|https://issues.apache.org/jira/projects/FLINK/issues/FLINK-31996]
left uncovered and only rejects a configuration that is already restored
incorrectly today; behavior for valid jobs is unchanged.
h2. Reproduction Steps
Using parallelism = 1 and default configuration:
# Build a pipeline where the keyed operator stays FORWARD-connected to its
source so the two are chainable — e.g. via
{{{}DataStreamUtils.reinterpretAsKeyedStream{}}}. (A normal {{keyBy}} inserts a
hash-partition edge, which is a chain break, so the keyed operator would become
its own chain head and the mismatch would never surface.) Give the keyed
operator an explicit {{{}setMaxParallelism(64){}}}, which differs from the
source's auto-derived 128.
# Run with operator chaining disabled, process records, and trigger a
canonical savepoint.
# Restore with chaining enabled (the default) — the source and keyed operator
now chain into a single vertex, adopting the head's 128 max parallelism.
# State written at 64 key groups is read at 128; per-key state is silently
dropped with no exception raised.
A self-contained ITCase reproducing this deterministically on both the HashMap
and RocksDB state backends is available here:
[ChainingMaxParallelismStateLossITCase|https://gist.github.com/rionmonster/ea8bcb8bbd22cb2921bdaec878cc41ca]
{*}Note{*}: the keyed operator must remain FORWARD-connected to its source
(e.g. via {{{}reinterpretAsKeyedStream{}}}) for the two to chain; a plain
{{keyBy}} inserts a chain-breaking hash edge.
was:
h2. Summary
[FLINK-31996|https://issues.apache.org/jira/projects/FLINK/issues/FLINK-31996]
allowed operators with different max parallelism to be chained into a single
vertex, and made it the default
({{{}pipeline.operator-chaining.chain-operators-with-different-max-parallelism{}}}).
On restore, Flink validates that an operator's max parallelism matches the
savepoint's and rejects an incompatible change with a clear error but that
validation is per-operator against the vertex's single max parallelism. It does
not verify that the operators now sharing a vertex have compatible max
parallelism with each other. When they don't, the per-vertex reconciliation
silently adopts one operator's value instead of rejecting.
h2. Current behavior
- A single operator whose max parallelism changed since the savepoint →
rejected: "{_}Max parallelism mismatch … changed in a non-compatible way."{_}
(Correct.)
- Multiple operators chained into one vertex with different recorded max
parallelism into {{StateAssignmentOperation#checkParallelismPreconditions}}
whichreconciles each against the shared vertex and adopts one value; operators
whose recorded value differs are accepted under the adopted key-group count
with {*}no equivalent check{*}.
h2. The gap
{{checkParallelismPreconditions}} checks each operator against the vertex's max
parallelism (and {{canRescaleMaxParallelism}} correctly permits rescaling an
auto-derived value), but nothing checks that the operators sharing the vertex
agree with one another.
[FLINK-31996|https://issues.apache.org/jira/projects/FLINK/issues/FLINK-31996]
made this arrangement reachable by default; the restore-side validation was not
extended to cover it. The single-operator path is guarded; the shared-vertex
path is not.
h2. Why it matters
A keyed operator can then be restored under a key-group count different from
the one its state was written with, so its state is mapped through a different
{{hash % maxParallelism}} layout. The restore should reject this, consistent
with how it already rejects the single-operator case, rather than proceeding.
h2. Proposed change (validation-only, low-risk)
During the per-vertex reconciliation on restore
({{{}StateAssignmentOperation#checkParallelismPreconditions{}}}), add a
precondition that explicitly compares the recorded max parallelism (key-group
count) of the keyed-state operators assigned to the same vertex, and fails with
a clear error mirroring the existing single-operator mismatch message (when
they disagree), rather than silently adopting one.
This extends the check
[FLINK-31996|https://issues.apache.org/jira/projects/FLINK/issues/FLINK-31996]
left uncovered and only rejects a configuration that is already restored
incorrectly today; behavior for valid jobs is unchanged.
h2. Reproduction Steps
Using parallelism = 1 and default configuration:
# Build a pipeline where the keyed operator stays FORWARD-connected to its
source so the two are chainable — e.g. via
{{{}DataStreamUtils.reinterpretAsKeyedStream{}}}. (A normal {{keyBy}} inserts a
hash-partition edge, which is a chain break, so the keyed operator would become
its own chain head and the mismatch would never surface.) Give the keyed
operator an explicit {{{}setMaxParallelism(64){}}}, which differs from the
source's auto-derived 128.
# Run with operator chaining disabled, process records, and trigger a
canonical savepoint.
# Restore with chaining enabled (the default) — the source and keyed operator
now chain into a single vertex, adopting the head's 128 max parallelism.
# State written at 64 key groups is read at 128; per-key state is silently
dropped with no exception raised.
A self-contained ITCase reproducing this deterministically on both the HashMap
and RocksDB state backends is available here:
[ChainingMaxParallelismStateLossITCase|https://gist.github.com/rionmonster/ea8bcb8bbd22cb2921bdaec878cc41ca]
{*}Note{*}: the keyed operator must remain FORWARD-connected to its source
(e.g. via {{{}reinterpretAsKeyedStream{}}}) for the two to chain; a plain
{{keyBy}} inserts a chain-breaking hash edge.
> State restore does not validate max-parallelism agreement between operators
> sharing a vertex
> --------------------------------------------------------------------------------------------
>
> Key: FLINK-40113
> URL: https://issues.apache.org/jira/browse/FLINK-40113
> Project: Flink
> Issue Type: Improvement
> Components: Runtime / Checkpointing, Runtime / State Backends
> Affects Versions: 2.0.0, 1.18.0, 1.19.0, 1.20.0, 2.1.0, 2.2.0, 2.3.0, 2.4.0
> Reporter: Rion Williams
> Priority: Major
>
> h2. Summary
> [FLINK-31996|https://issues.apache.org/jira/projects/FLINK/issues/FLINK-31996]
> allowed operators with different max parallelism to be chained into a single
> vertex, and made it the default
> ({{{}pipeline.operator-chaining.chain-operators-with-different-max-parallelism{}}}).
> On restore, Flink validates that an operator's max parallelism matches the
> savepoint's and rejects an incompatible change with a clear error but that
> validation is per-operator against the vertex's single max parallelism. It
> does not verify that the operators now sharing a vertex have compatible max
> parallelism with each other. When they don't, the per-vertex reconciliation
> silently adopts one operator's value instead of rejecting.
> h2. Current behavior
> - A single operator whose max parallelism changed since the savepoint →
> rejected: "{_}Max parallelism mismatch … changed in a non-compatible way."{_}
> (Correct.)
> - Multiple operators chained into one vertex with different recorded max
> parallelism into {{StateAssignmentOperation#checkParallelismPreconditions}}
> whichreconciles each against the shared vertex and adopts one value;
> operators whose recorded value differs are accepted under the adopted
> key-group count with {*}no equivalent check{*}.
> h2. The gap
> {{checkParallelismPreconditions}} checks each operator against the vertex's
> max parallelism (and {{canRescaleMaxParallelism}} correctly permits rescaling
> an auto-derived value), but nothing checks that the operators sharing the
> vertex agree with one another.
> [FLINK-31996|https://issues.apache.org/jira/projects/FLINK/issues/FLINK-31996]
> made this arrangement reachable by default; the restore-side validation was
> not extended to cover it. The single-operator path is guarded; the
> shared-vertex path is not.
> h2. Why it matters
> A keyed operator can then be restored under a key-group count different from
> the one its state was written with, so its state is mapped through a
> different {{hash % maxParallelism}} layout. The restore should reject this,
> consistent with how it already rejects the single-operator case, rather than
> proceeding.
> h2. Proposed change (validation-only, low-risk)
> Add explicit validation during per-vertex reconciliation in
> {{StateAssignmentOperation#checkParallelismPreconditions}}: for any vertex
> that carries keyed state, verify that all operators chained into it recorded
> the same maximum parallelism in the checkpoint, and fail with a clear error
> otherwise. The
> comparison spans all chained operators, not only the keyed ones, because a
> non-keyed operator's recorded value can win the reconciliation and misroute a
> keyed operator's state.
> This extends the check
> [FLINK-31996|https://issues.apache.org/jira/projects/FLINK/issues/FLINK-31996]
> left uncovered and only rejects a configuration that is already restored
> incorrectly today; behavior for valid jobs is unchanged.
> h2. Reproduction Steps
> Using parallelism = 1 and default configuration:
> # Build a pipeline where the keyed operator stays FORWARD-connected to its
> source so the two are chainable — e.g. via
> {{{}DataStreamUtils.reinterpretAsKeyedStream{}}}. (A normal {{keyBy}} inserts
> a hash-partition edge, which is a chain break, so the keyed operator would
> become its own chain head and the mismatch would never surface.) Give the
> keyed operator an explicit {{{}setMaxParallelism(64){}}}, which differs from
> the source's auto-derived 128.
> # Run with operator chaining disabled, process records, and trigger a
> canonical savepoint.
> # Restore with chaining enabled (the default) — the source and keyed
> operator now chain into a single vertex, adopting the head's 128 max
> parallelism.
> # State written at 64 key groups is read at 128; per-key state is silently
> dropped with no exception raised.
> A self-contained ITCase reproducing this deterministically on both the
> HashMap and RocksDB state backends is available here:
> [ChainingMaxParallelismStateLossITCase|https://gist.github.com/rionmonster/ea8bcb8bbd22cb2921bdaec878cc41ca]
> {*}Note{*}: the keyed operator must remain FORWARD-connected to its source
> (e.g. via {{{}reinterpretAsKeyedStream{}}}) for the two to chain; a plain
> {{keyBy}} inserts a chain-breaking hash edge.
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
