[SPARK-43221][CORE][3.5] Host local block fetching should use a block status of a block stored on disk

[attilapiros]

This is a backport to branch-3.5 from master.

Thanks for @yorksity who reported this error and even provided a PR for it.
This solution very different from #40883 as BlockManagerMasterEndpoint#getLocationsAndStatus() needed some refactoring.

What changes were proposed in this pull request?

This PR fixes an error which can be manifested in the following exception:

25/02/20 09:58:31 ERROR util.Utils: [Executor task launch worker for task 61.0 in stage 67.0 (TID 9391)]: Exception encountered
java.lang.ArrayIndexOutOfBoundsException: 0
  at org.apache.spark.broadcast.TorrentBroadcast.$anonfun$readBlocks$1(TorrentBroadcast.scala:185) ~[spark-core_2.12-3.3.2.3.3.7190.5-2.jar:3.3.2.3.3.7190.5-2]
  at scala.runtime.java8.JFunction1$mcVI$sp.apply(JFunction1$mcVI$sp.java:23) ~[scala-library-2.12.15.jar:?]
  at scala.collection.immutable.List.foreach(List.scala:431) ~[scala-library-2.12.15.jar:?]
  at org.apache.spark.broadcast.TorrentBroadcast.readBlocks(TorrentBroadcast.scala:171) ~[spark-core_2.12-3.3.2.3.3.7190.5-2.jar:3.3.2.3.3.7190.5-2]                                

The PR is changing BlockManagerMasterEndpoint#getLocationsAndStatus().

The BlockManagerMasterEndpoint#getLocationsAndStatus() function is giving back an optional BlockLocationsAndStatus which consist of 3 parts:

• locations: all the locations where the block can be found (as a sequence of block manager IDs)
• status: one block status
• localDirs: optional directory paths which can be used to read block if the block is found in the disk of an executor running on the same host

The block (either RDD blocks, shuffle blocks or torrent blocks) can be stored in many executors with different storage levels: disk or memory.

This PR changing how the block status and the block manager ID for the localDirs is found to guarantee they belong together.

Why are the changes needed?

Before this PR the BlockManagerMasterEndpoint#getLocationsAndStatus() was searching for the block status (status) and the localDirs separately. The block status actually was computed as the very first one where the block can be found. This way it can easily happen this block status was representing an in-memory block (where the disk size is 0 as it is stored in the memory) but the localDirs was filled out based on a host local block instance which was stored on disk.

This situation can be very frequent but only causing problems (exceptions as above) when encryption is on (spark.io.encryption.enabled=true) as for a not encrypted block the whole file containing the block is read, see
https://github.com/apache/spark/blob/branch-3.5/core/src/main/scala/org/apache/spark/storage/BlockManager.scala#L1244

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Host local block fetching was already covered by some existing unit tests but a new unit test is provided for this exact case: "SPARK-43221: Host local block fetching should use a block status with disk size".

The number of block mangers and the order of the blocks was chosen after some experimentation as the block status order is depends on a HashSet, see:

  private val blockLocations = new JHashMap[BlockId, mutable.HashSet[BlockManagerId]]

This test was executed with the old code too to validate the issue is reproduced:

BlockManagerSuite:
OpenJDK 64-Bit Server VM warning: Sharing is only supported for boot loader classes because bootstrap classpath has been appended
- SPARK-43221: Host local block fetching should use a block status with disk size *** FAILED ***
  0 was not greater than 0 The block size must be greater than 0 for a nonempty block! (BlockManagerSuite.scala:491)
Run completed in 6 seconds, 705 milliseconds.
Total number of tests run: 1
Suites: completed 1, aborted 0
Tests: succeeded 0, failed 1, canceled 0, ignored 0, pending 0
*** 1 TEST FAILED ***

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

No.

(cherry picked from commit 997e599)

[dongjoon-hyun]

Could you fix the compilation failure, @attilapiros ?

[error] /home/runner/work/spark/spark/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:865:10: value orElse is not a member of Equals
[error] possible cause: maybe a semicolon is missing before `value orElse'?
[error]         .orElse {
[error]          ^
[error] one error found

[attilapiros]

Fixed. It was a difference between Scala 2.12 vs 2.13.
The Option#zip gives back an Iterable on Scala 2.12 instead of an Option:

# scala
Welcome to Scala 2.12.8 (OpenJDK 64-Bit Server VM, Java 1.8.0_442).
Type in expressions for evaluation. Or try :help.

scala> val option = Some(1)
option: Some[Int] = Some(1)

scala> option.zip(option)
res1: Iterable[(Int, Int)] = List((1,1))

scala> None.zip(None)
res2: Iterable[(Nothing, Nothing)] = List()

scala> res1.headOption
res3: Option[(Int, Int)] = Some((1,1))

scala> res2.headOption
res4: Option[(Nothing, Nothing)] = None

Although the documentation does not says differently: https://www.scala-lang.org/api/2.12.8/scala/Option.html#zip[B](that:scala.collection.GenIterable[B]):Option[(A,B)]

[dongjoon-hyun]

+1, LGTM (Pending CIs). Thank you, @attilapiros .

[dongjoon-hyun]

Merged to branch-3.5.