Rui Fan created FLINK-40780:
-------------------------------
Summary: FORWARD edge silently converted to REBALANCE on
parallelism mismatch corrupts changelog (upsert/retract) streams
Key: FLINK-40780
URL: https://issues.apache.org/jira/browse/FLINK-40780
Project: Flink
Issue Type: Bug
Components: Runtime / Coordination
Reporter: Rui Fan
Assignee: Rui Fan
h2. Problem
When a FORWARD (pointwise) edge connects two operators whose parallelism
differs, the runtime silently replaces the ForwardPartitioner with a REBALANCE
(round-robin) partitioner (introduced in FLINK-30213). This is correct for
append-only streams, but for order-sensitive changelog streams (upsert/retract)
it silently produces wrong, non-deterministic results — no error, no failing
job.
A FORWARD edge preserves the upstream partitioning, and the planner relies on
this to skip a redundant keyed exchange when the input is already partitioned
by the required key. Downgrading it to REBALANCE breaks that guarantee: records
for the same key are scattered across downstream subtasks, so a keyed operator
or upsert sink no longer sees a key's changelog in order.
h2. Example
A streaming global aggregation into an upsert sink:
INSERT INTO upsert_sink SELECT 1 AS k, COUNT(*) FROM src;
Compiled at parallelism 1, the whole path (GroupAggregate → Calc → Sink) chains
via FORWARD — correct. If the sink is later run at a higher parallelism than
the parallelism-1 aggregate (e.g. via a per-vertex parallelism override /
rescale), the edge into the sink stays FORWARD but now spans 1→N: the single
key's changelog is spread round-robin across N writers, which interleave writes
to the same key and yield a non-deterministic final value.
h2. Proposed fix
Add a config option controlling the behavior when a FORWARD edge spans
mismatched parallelism:
pipeline.forward-edge.parallelism-mismatch-mode (enum)
- REBALANCE — current behavior (downgrade to REBALANCE). Default
(backward-compatible).
- FAIL — reject the job with a clear exception.
- KEEP_FORWARD — keep the 1-to-1 mapping (order-preserving; the pre-FLINK-30213
behavior).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
