fixup! Make FallbackStorage.read return a truly lazy ManagedBuffer

Author: EnricoMi

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

@@ -23,7 +23,6 @@ import java.nio.ByteBuffer
 import scala.concurrent.Future
 import scala.reflect.ClassTag
 
-import com.google.common.io.ByteStreams
 import io.netty.buffer.Unpooled
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.{FileSystem, Path}
@@ -34,7 +33,7 @@ import org.apache.spark.internal.Logging
 import org.apache.spark.internal.LogKeys._
 import org.apache.spark.internal.config.{STORAGE_DECOMMISSION_FALLBACK_STORAGE_CLEANUP, STORAGE_DECOMMISSION_FALLBACK_STORAGE_PATH}
 import org.apache.spark.network.buffer.ManagedBuffer
-import org.apache.spark.network.util.JavaUtils
+import org.apache.spark.network.util.{JavaUtils, LimitedInputStream}
 import org.apache.spark.rpc.{RpcAddress, RpcEndpointRef, RpcTimeout}
 import org.apache.spark.shuffle.{IndexShuffleBlockResolver, ShuffleBlockInfo}
 import org.apache.spark.shuffle.IndexShuffleBlockResolver.NOOP_REDUCE_ID
@@ -142,7 +141,7 @@ private[storage] class FallbackStorageManagedBuffer (
     try {
       val in = fallbackFileSystem.open(dataFile)
       in.seek(offset)
-      ByteStreams.limit(in, blockSize)
+      new LimitedInputStream(in, blockSize)
     } finally {
       logDebug(s"Took ${(System.nanoTime() - startTimeNs) / (1000 * 1000)}ms")
     }