[SPARK-39072][SHUFFLE]Fast fail the remaining push blocks if shuffle …

[wankunde]

…stage finalized

What changes were proposed in this pull request?

Limit the push blocks in flight and try to stop push the remaining blocks shuffle stage is finalized.

Why are the changes needed?

Map task will try to push all map outputs to external shuffle service now.
After the shuffle stage is finalized, the reduce fetch blocks RPC will be blocked if there are still many map output blocks in flight.
We could stop pushing the remaining blocks if the shuffle stage is finalized.

Does this PR introduce any user-facing change?

No

How was this patch tested?

Exists UT

[AmplabJenkins]

Can one of the admins verify this patch?

[mridulm]

+CC @otterc

[otterc]

@wankunde Can you please provide more details/logs of the problem that you are trying to solve. In specific, can you provide some logs that exhibit the below

After the shuffle stage is finalized, the reduce fetch blocks RPC will be blocked if there are still many map output blocks in flight.

ShuffleBlockPusher already uses config maxBlocksInFlightPerAddress. Why do we need a new configuration for this?

[otterc]

@wankunde Can you please provide more details/logs of the problem that you are trying to solve. In specific, can you provide some logs that exhibit the below

After the shuffle stage is finalized, the reduce fetch blocks RPC will be blocked if there are still many map output blocks in flight.

ShuffleBlockPusher already uses config maxBlocksInFlightPerAddress. Why do we need a new configuration for this?

[wankunde]

Hi @otterc , You are right, we do not need this PR because ShuffleBlockPusher already uses config maxBlocksInFlightPerAddress. But I found that some ESS received FinalizeShuffleMerge RPC after few seconds. I am not sure if it is because there are many in flight pushing blocks to those ESS.

[wankunde]

Driver Logs

22/05/07 01:00:13,218 INFO [shuffle-client-9-5] scheduler.DAGScheduler:57 : finalizeShuffleMerge success from ESS_NODE in 7763 ms  shuffleId 2 - 0

ESS logs

2022-05-07 01:00:08,891 DEBUG org.apache.spark.network.server.TransportServer: New connection accepted for remote address /DRIVER_IP:41402.
2022-05-07 01:00:12,816 INFO org.apache.spark.network.shuffle.ExternalBlockHandler: receive FinalizeShuffleMerge finalize shuffle merge
2022-05-07 01:00:12,816 INFO org.apache.spark.network.shuffle.RemoteBlockPushResolver: application_1651741622014_0557 attempt -1 shuffle 2 shuffleMerge 0: finalize shuffle merge
2022-05-07 01:00:13,217 TRACE org.apache.spark.network.server.TransportRequestHandler: Sent result RpcResponse[requestId=7587608556323056846,body=NioManagedBuffer[buf=java.nio.HeapByteBuffer[pos=0 lim=818744 cap=818744]]] to client /DRIVER_IP:41402

[wankunde]

Close this PR