Zhilong Hong created FLINK-21331:
------------------------------------
Summary: Optimize calculating tasks to restart in RestartPipelinedRegionFailoverStrategy
Key: FLINK-21331
URL:
https://issues.apache.org/jira/browse/FLINK-21331 Project: Flink
Issue Type: Sub-task
Components: Runtime / Coordination
Reporter: Zhilong Hong
Fix For: 1.13.0
RestartPipelinedRegionFailoverStrategy is used to calculate the tasks to restart when a task failure occurs. It contains two parts: firstly calculate the regions to restart; then add all the tasks in these regions to the restarting queue.
The bottleneck is mainly in the first part. This part traverses all the upstream and downstream regions of the failed region to determine whether they should be restarted or not.
For the current failed region, if its consumed result partition is not available, the owner, i.e., the upstream region should restart. Also, since the failed region needs to restart, its result partition won't be available, all the downstream regions need to restart, too.
1. Calculating the upstream regions that should restart
The current implementation is:
{code:java}
for each SchedulingExecutionVertex in current visited SchedulingPipelinedRegion:
for each consumed SchedulingResultPartition of the SchedulingExecutionVertex:
if the result partition is not available:
add the producer region to the restart queue
{code}
Based on FLINK-21328, the consumed result partition of a vertex is already grouped. Here we can use a HashSet to record the visited result partition group. For vertices connected with all-to-all edges, they will only need to traverse the group once. This decreases the time complexity from O(N^2) to O(N).
2. Calculating the downstream regions that should restart
The current implementation is:
{code:java}
for each SchedulingExecutionVertex in current visited SchedulingPipelinedRegion:
for each produced SchedulingResultPartition of the SchedulingExecutionVertex:
for each consumer SchedulingExecutionVertex of the produced SchedulingResultPartition:
if the region containing the consumer SchedulingExecutionVertex is not visited:
add the region to the restart queue
{code}
Since the count of the produced result partitions of a vertex equals the count of output JobEdges, the time complexity of this procedure is actually O(N^2). As the consumer vertices of a result partition are already grouped, we can use a HashSet to record the visited ConsumerVertexGroup. The time complexity decreases from O(N^2) to O(N).
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
Locked
