fixup! Update unit tests

Author: EnricoMi

core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala

@@ -153,7 +153,8 @@ final class ShuffleBlockFetcherIterator(
    * Thread pool reading from fallback storage, first creating FallbackStorageRequest from
    * block id and map index, then materializing requests to SuccessFetchResult.
    */
-  private[this] val fallbackStorageReadPool: ThreadPoolExecutor =
+  // This is visible for testing
+  private[storage] val fallbackStorageReadPool: ThreadPoolExecutor =
     ThreadUtils.newDaemonFixedThreadPool(fallbackStorageReadThreads, "fallback-storage-read")
   private[this] val fallbackStorageReadContext: ExecutionContextExecutor =
     ExecutionContext.fromExecutor(fallbackStorageReadPool)

core/src/test/scala/org/apache/spark/storage/ShuffleBlockFetcherIteratorSuite.scala

@@ -39,6 +39,8 @@ import org.mockito.invocation.InvocationOnMock
 import org.mockito.stubbing.Answer
 import org.roaringbitmap.RoaringBitmap
 import org.scalatest.PrivateMethodTester
+import org.scalatest.concurrent.Eventually
+import org.scalatest.time.SpanSugar.convertIntToGrainOfTime
 
 import org.apache.spark.{MapOutputTracker, SparkFunSuite, TaskContext}
 import org.apache.spark.MapOutputTracker.SHUFFLE_PUSH_MAP_ID
@@ -52,7 +54,8 @@ import org.apache.spark.storage.ShuffleBlockFetcherIterator._
 import org.apache.spark.util.Utils
 
 
-class ShuffleBlockFetcherIteratorSuite extends SparkFunSuite with PrivateMethodTester {
+class ShuffleBlockFetcherIteratorSuite extends SparkFunSuite with Eventually
+  with PrivateMethodTester {
 
   private var transfer: BlockTransferService = _
   private var mapOutputTracker: MapOutputTracker = _
@@ -361,6 +364,10 @@ class ShuffleBlockFetcherIteratorSuite extends SparkFunSuite with PrivateMethodT
     verify(blockManager, times(3)).getLocalBlockData(any())
 
     // 2 fallback storage blocks fetched in initialization
+    // initialize creates futures that eventually call into getFallbackStorageBlockData
+    eventually(timeout(1.seconds), interval(10.millis)) {
+      assert(iterator.fallbackStorageReadPool.getCompletedTaskCount >= 2)
+    }
     verify(blockManager, times(2)).getFallbackStorageBlockData(any())
     // SPARK-55469: but buffer data have never been materialized
     fallbackBlocks.values.foreach { mockBuf =>
@@ -552,6 +559,10 @@ class ShuffleBlockFetcherIteratorSuite extends SparkFunSuite with PrivateMethodT
     verify(blockManager, times(1)).getLocalBlockData(any())
 
     // 1 fallback merge block fetched in initialization
+    // initialize creates futures that eventually call into getFallbackStorageBlockData
+    eventually(timeout(1.seconds), interval(10.millis)) {
+      assert(iterator.fallbackStorageReadPool.getCompletedTaskCount >= 1)
+    }
     verify(blockManager, times(1)).getFallbackStorageBlockData(any())
     // SPARK-55469: but buffer data have never been materialized
     mergedFallbackBlocks.values.foreach { mockBuf =>