[
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)
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
With parallelism = 1 and using default config
(`chain-operators-with-different-max-parallelism = true`):
{code:java}
DataStreamSource<Long> src = env.fromSequence(1, 200); // chain head; max
parallelism auto-derived (128)
// reinterpretAsKeyedStream puts a forward (chainable) edge before the keyed
operator,
// so it can be chained as a NON-head.
DataStreamUtils.reinterpretAsKeyedStream(src, key -> key % 4)
.process(new PerKeyCounter()) // keyed ValueState<Long> counter
.uid("keyed")
.setMaxParallelism(64) // explicit — differs from the
head's 128
.addSink(...);
{code}
1. Run with `env.disableOperatorChaining()`, process some records, take a
savepoint → the keyed operator is its own vertex with max parallelism 64; its
state is written into 64 key groups.
2. Restore into the same job without disabling chaining → source and keyed
operator chain into one vertex, which adopts the head's max parallelism 128;
the operator's configured 64 is silently discarded.
3. The keyed state (written at 64 key groups) is read at 128 → per-key state
isn't found and counters restart from zero. No exception is raised.
With `setMaxParallelism(256)` (larger than the head's 128) instead, the same
setup fails the restore loudly with IllegalStateException: The key group must
belong to the backend — same cause, opposite direction.
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 left uncovered and only rejects a
configuration that is already restored incorrectly today; behavior for valid
jobs is unchanged.
h2. Reproduction
With parallelism = 1 and using default config
(`chain-operators-with-different-max-parallelism = true`):
{code:java}
DataStreamSource<Long> src = env.fromSequence(1, 200); // chain head; max
parallelism auto-derived (128)
// reinterpretAsKeyedStream puts a forward (chainable) edge before the keyed
operator,
// so it can be chained as a NON-head.
DataStreamUtils.reinterpretAsKeyedStream(src, key -> key % 4)
.process(new PerKeyCounter()) // keyed ValueState<Long> counter
.uid("keyed")
.setMaxParallelism(64) // explicit — differs from the
head's 128
.addSink(...);
{code}
1. Run with `env.disableOperatorChaining()`, process some records, take a
savepoint → the keyed operator is its own vertex with max parallelism 64; its
state is written into 64 key groups.
2. Restore into the same job without disabling chaining → source and keyed
operator chain into one vertex, which adopts the head's max parallelism 128;
the operator's configured 64 is silently discarded.
3. The keyed state (written at 64 key groups) is read at 128 → per-key state
isn't found and counters restart from zero. No exception is raised.
With `setMaxParallelism(256)` (larger than the head's 128) instead, the same
setup fails the restore loudly with IllegalStateException: The key group must
belong to the backend — same cause, opposite direction.
> 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
> 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)
> 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
> With parallelism = 1 and using default config
> (`chain-operators-with-different-max-parallelism = true`):
> {code:java}
> DataStreamSource<Long> src = env.fromSequence(1, 200); // chain head; max
> parallelism auto-derived (128)
> // reinterpretAsKeyedStream puts a forward (chainable) edge before the keyed
> operator,
> // so it can be chained as a NON-head.
> DataStreamUtils.reinterpretAsKeyedStream(src, key -> key % 4)
> .process(new PerKeyCounter()) // keyed ValueState<Long> counter
> .uid("keyed")
> .setMaxParallelism(64) // explicit — differs from the
> head's 128
> .addSink(...);
> {code}
> 1. Run with `env.disableOperatorChaining()`, process some records, take a
> savepoint → the keyed operator is its own vertex with max parallelism 64; its
> state is written into 64 key groups.
> 2. Restore into the same job without disabling chaining → source and keyed
> operator chain into one vertex, which adopts the head's max parallelism 128;
> the operator's configured 64 is silently discarded.
> 3. The keyed state (written at 64 key groups) is read at 128 → per-key state
> isn't found and counters restart from zero. No exception is raised.
> With `setMaxParallelism(256)` (larger than the head's 128) instead, the same
> setup fails the restore loudly with IllegalStateException: The key group must
> belong to the backend — same cause, opposite direction.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
