[SPARK-51638][CORE] Fix fetching the remote disk stored RDD blocks via the external shuffle service

[attilapiros]

What changes were proposed in this pull request?

Fix remote fetching of disk stored RDD blocks via the external shuffle service when spark.shuffle.service.fetch.rdd.enabled is set.

Why are the changes needed?

After https://issues.apache.org/jira/browse/SPARK-43221 remote fetching was handled in BlockManagerMasterEndpoint#getLocationsAndStatus at one place where all the location was used along with the blockManagerInfo map but this map only includes information about the active executors which are not already killed (after for example downscaling in dynamic allocation or just killed because of a failures).

This PR extend the search to all the remote external shuffle services where the blockStatusByShuffleService map is used. That map contains block infos even for the killed executors.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

An existing unit test was extended.

Was this patch authored or co-authored using generative AI tooling?

No.

[attilapiros]

cc @dongjoon-hyun, @Ngone51

[dongjoon-hyun]

Ack, @attilapiros .

[dongjoon-hyun]

Since this is a regression, could you make an independent test case instead of adding to the existing SPARK-25888 test case, @attilapiros ?

[dongjoon-hyun]

+1 for the code change.

[attilapiros]

Since this is a regression, could you make an independent test case instead of adding to the existing SPARK-25888 test case, @attilapiros ?

We can duplicate the test but actually these checks were really missing from the old test ("SPARK-25888: using external shuffle service fetching disk persisted blocks")
+
We will double/quadruple the runtime as any test we add runs in 2 times (ExternalShuffleServiceSuite&SslExternalShuffleServiceSuite instance):

[info] ExternalShuffleServiceSuite:
[info] - SPARK-25888: using external shuffle service fetching disk persisted blocks (15 seconds, 358 milliseconds)
[info] SslExternalShuffleServiceSuite:
[info] - SPARK-25888: using external shuffle service fetching disk persisted blocks (14 seconds, 270 milliseconds)

[attilapiros]

Runtimes in my machine (within a docker container) was super bad. Without docker it is much better:

[info] ExternalShuffleServiceSuite:
[info] - SPARK-25888: using external shuffle service fetching disk persisted blocks (6 seconds, 770 milliseconds)
[info] SslExternalShuffleServiceSuite:
[info] - SPARK-25888: using external shuffle service fetching disk persisted blocks (6 seconds, 987 milliseconds)

[dongjoon-hyun]

If you really don't prefer to add a new test case with the test prefix, SPARK-51638, feel free to merge with the AS-IS status, @attilapiros . Although I'm reluctant to change SPARK-25888 test case, I trust your decision.

[attilapiros]

Let's wait for @Ngone51 to review the PR and he can be the judge in this question as well.