fixup! Add unit tests

Author: EnricoMi

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

@@ -22,10 +22,11 @@ import java.nio.file.Files
 import scala.concurrent.duration._
 import scala.util.Random
 
+import io.netty.buffer.ByteBuf
 import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.{FSDataInputStream, LocalFileSystem, Path, PositionedReadable, Seekable}
+import org.apache.hadoop.fs.{FileSystem, FSDataInputStream, LocalFileSystem, Path, PositionedReadable, Seekable}
 import org.mockito.{ArgumentMatchers => mc}
-import org.mockito.Mockito.{mock, never, verify, when}
+import org.mockito.Mockito.{mock, never, spy, times, verify, when}
 import org.scalatest.concurrent.Eventually.{eventually, interval, timeout}
 
 import org.apache.spark.{LocalSparkContext, SparkConf, SparkContext, SparkFunSuite, TestUtils}
@@ -155,9 +156,49 @@ class FallbackStorageSuite extends SparkFunSuite with LocalSparkContext {
     assert(fallbackStorage.exists(1, ShuffleDataBlockId(1, 2L, NOOP_REDUCE_ID).name))
 
     val readResult = FallbackStorage.read(conf, ShuffleBlockId(1, 2L, 0))
+    assert(readResult.isInstanceOf[FallbackStorageManagedBuffer])
     assert(readResult.nioByteBuffer().array().sameElements(content))
   }
 
+  test("SPARK-55469: FallbackStorageManagedBuffer reads block data lazily") {
+    withTempDir { dir =>
+      val fs = FileSystem.getLocal(new Configuration())
+      val file = new Path(dir.getAbsolutePath, "file")
+      val data = Array[Byte](1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
+      tryWithResource(fs.create(file)) { os => os.write(data) }
+
+      Seq((0, 4), (1, 2), (4, 4), (7, 2), (8, 0)).foreach { case (offset, length) =>
+        val clue = s"offset: $offset, length: $length"
+
+        // creating the managed buffer does not open the file
+        val mfs = spy(fs)
+        val buf = new FallbackStorageManagedBuffer(mfs, file, offset, length)
+        verify(mfs, never()).open(mc.any[Path]())
+        assert(buf.size() === length, clue)
+
+        // creating the input stream opens the file
+        {
+          val bytes = buf.createInputStream().readAllBytes()
+          verify(mfs, times(1)).open(mc.any[Path]())
+          assert(bytes.mkString(",") === data.slice(offset, offset + length).mkString(","), clue)
+        }
+
+        // getting a NIO ByteBuffer opens the file again
+        {
+          val bytes = buf.nioByteBuffer().array()
+          verify(mfs, times(2)).open(mc.any[Path]())
+          assert(bytes.mkString(",") === data.slice(offset, offset + length).mkString(","), clue)
+        }
+
+        // getting a NIO ByteBuffer opens the file again
+        assert(buf.convertToNetty().asInstanceOf[ByteBuf].release() === length > 0, clue)
+        verify(mfs, times(3)).open(mc.any[Path]())
+        assert(buf.convertToNettyForSsl().asInstanceOf[ByteBuf].release() === length > 0, clue)
+        verify(mfs, times(4)).open(mc.any[Path]())
+      }
+    }
+  }
+
   test("SPARK-34142: fallback storage API - cleanUp app") {
     withTempDir { dir =>
       Seq(true, false).foreach { cleanUp =>
@@ -372,6 +413,7 @@ class FallbackStorageSuite extends SparkFunSuite with LocalSparkContext {
     }
   }
 }
+
 class ReadPartialInputStream(val in: FSDataInputStream) extends InputStream
   with Seekable with PositionedReadable {
   override def read: Int = in.read