Rion Williams created FLINK-40113:
-------------------------------------
Summary: 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
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)
Extend the restore validation to cover the chained-operators case FLINK-31996
previously introduced: when operators sharing a vertex have incompatible max
parallelism, fail with a clear error (mirroring the existing single-operator
message) instead of silently adopting one. This only rejects a configuration
that is {*}already restored incorrectly today{*}; behavior for valid jobs is
unchanged.
Roughly, in the per-vertex reconciliation
(`StateAssignmentOperation#checkParallelismPreconditions`) we would perform
something _like_ the following:
{code:java}
// Operators assigned to one vertex must have been checkpointed with the same
max parallelism,
// or a keyed operator would be restored under a different key-group count than
its state was
// written with.
Set<Integer> maxParallelisms = operatorStates.stream() // operators carrying
keyed state
.map(OperatorState::getMaxParallelism)
.collect(Collectors.toSet());
if (maxParallelisms.size() > 1) {
throw new IllegalStateException(String.format(
"Operators chained into %s were checkpointed with different max
parallelisms %s; "
+ "restoring would change a keyed operator's key-group count.",
executionJobVertex, maxParallelisms));
}{code}
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)
