Zhilong Hong created FLINK-22077:
------------------------------------
Summary: Wrong way to calculate cross-region
ConsumedPartitionGroups in PipelinedRegionSchedulingStrategy
Key: FLINK-22077
URL: https://issues.apache.org/jira/browse/FLINK-22077
Project: Flink
Issue Type: Bug
Components: Runtime / Coordination
Affects Versions: 1.13.0
Reporter: Zhilong Hong
Fix For: 1.13.0
h3. Introduction
We implement a wrong way to calculate cross-region ConsumedPartitionGroups in
{{PipelinedRegionSchedulingStrategy}}, it slows down the procedure of
{{onExecutionStateChange}}, make the complexity from O(N) to O(N^2). Also the
semantic of cross-region is totally wrong.
h3. Details
In {{PipelinedRegionSchedulingStrategy#startScheduling}}, as expected, we need
to schedule all region with no external blocking edges, i.e., source regions.
To decrease the complexity, we choose to schedule all the regions that has no
external BLOCKING ConsumedPartitionGroups.
However, for the case illustrated in FLINK-22017, the region 2 has a
ConsumedPartitionGroup, which has both internal and external blocking
IntermediateResultPartitions. If we choose one to represent the entire
ConsumedPartitionGroup, it may choose the internal one, and make the entire
group internal. Region 2 will be scheduled.
As Region 1 is not finished, Region 2 cannot transition to running. A deadlock
may happen if resource is not enough for both two regions.
To make it right, we introduced cross-region ConsumedPartitionGroups in
FLINK-21330. The regions which has ConsumedPartitionGroups with both internal
and external blocking IntermediateResultPartitions will be recorded. When we
call {{startScheduling}}, these ConsumedPartitionGroups will be treated as
external, and region 2 will not be scheduled.
But we have to admit that the implementation of cross-region is wrong. The
ConsumedPartitionGroups that has multiple producer regions will be treated as
cross-region groups. It is not the same logic as we mentioned above. The
semantic is totally wrong. Also all the ALL-TO-ALL BLOCKING
ConsumedPartitionGroups will be treated as cross-region groups, since their
producers are in different regions. (Each producer has its own region.) This
slows down the complexity from O(N) to O(N^2) for ALL-TO-ALL BLOCKING edges.
h3. Solution
To correctly calculate the cross-region ConsumedPartitionGroups, we can just
calculate the producer regions for all ConsumedPartitionGroups, and then
iterate all the regions and its ConsumedPartitionGroups. If the
ConsumedPartitionGroup has two or more producer regions, and the regions
contains current region, it is a cross-region ConsumedPartitionGroup. This
meets the correct semantics, and make sure ALL-TO-ALL BLOCKING
ConsumedPartitionGroups will not be treated as cross-region one. This fix will
also decreases the complexity from O(N) to O(N^2). I prefer it's necessary to
add this bug-fix to release 1.13.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
