Attemp to fetch failed block from FallbackStorage

Author: EnricoMi

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

@@ -32,10 +32,10 @@ import scala.util.{Failure, Success}
 import io.netty.util.internal.OutOfDirectMemoryError
 import org.roaringbitmap.RoaringBitmap
 
-import org.apache.spark.{MapOutputTracker, SparkException, TaskContext}
+import org.apache.spark.{MapOutputTracker, SparkEnv, SparkException, TaskContext}
 import org.apache.spark.MapOutputTracker.SHUFFLE_PUSH_MAP_ID
 import org.apache.spark.errors.SparkCoreErrors
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{config, Logging}
 import org.apache.spark.internal.LogKeys._
 import org.apache.spark.network.buffer.{FileSegmentManagedBuffer, ManagedBuffer}
 import org.apache.spark.network.shuffle._
@@ -973,14 +973,29 @@ final class ShuffleBlockFetcherIterator(
           }
 
         case FailureFetchResult(blockId, mapIndex, address, e) =>
+          var error = e
           var errorMsg: String = null
           if (e.isInstanceOf[OutOfDirectMemoryError]) {
             val logMessage = log"Block ${MDC(BLOCK_ID, blockId)} fetch failed after " +
               log"${MDC(MAX_ATTEMPTS, maxAttemptsOnNettyOOM)} retries due to Netty OOM"
             logError(logMessage)
             errorMsg = logMessage.message
+          } else if (
+            SparkEnv.get.conf.get(config.STORAGE_DECOMMISSION_FALLBACK_STORAGE_PATH).isDefined) {
+            try {
+              val buf = FallbackStorage.read(SparkEnv.get.conf, blockId)
+              results.put(SuccessFetchResult(blockId, mapIndex, address, buf.size(), buf,
+                isNetworkReqDone = false))
+              result = null
+              error = null
+            } catch {
+              case t: Throwable =>
+                logInfo(s"Failed to read block from fallback storage: $blockId", t)
+            }
+          }
+          if (error != null) {
+            throwFetchFailedException(blockId, mapIndex, address, error, Some(errorMsg))
           }
-          throwFetchFailedException(blockId, mapIndex, address, e, Some(errorMsg))
 
         case DeferFetchRequestResult(request) =>
           val address = request.address