[
https://issues.apache.org/jira/browse/FLINK-40780?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18121481#comment-18121481
]
Roman Khachatryan edited comment on FLINK-40780 at 10/1/26 8:18 AM:
--------------------------------------------------------------------
We should probably have some announcement about this issue in flink-devel ML
(once the ticket is closed).
Other users might run into this issue as well; and if we keep the defaults 'as
is' then they'll never know.
WDYT?
was (Author: roman_khachatryan):
We should have some announcement about the issue in flink-devel (once this is
closed).
Other users might run into this issue as well; and if we keep the defaults 'as
is' then they'll never know.
WDYT?
> 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
> Priority: Major
> Labels: pull-request-available
>
> 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)
