Modified BasicBlockFetchIterator to fail fast when local fetch error …

…has been occurred
apache · Jul 29, 2014 · b7b8250 · b7b8250
1 parent a3a9be1
commit b7b8250
Showing 1 changed file with 10 additions and 5 deletions.
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockFetcherIterator.scala b/core/src/main/scala/org/apache/spark/storage/BlockFetcherIterator.scala
@@ -199,16 +199,21 @@ object BlockFetcherIterator {
       // Get the local blocks while remote blocks are being fetched. Note that it's okay to do
       // these all at once because they will just memory-map some files, so they won't consume
       // any memory that might exceed our maxBytesInFlight
-      for (id <- localBlocksToFetch) {
-        try {
+      var fetchIndex = 0
+      try {
+        for (id <- localBlocksToFetch) {
+
           // getLocalFromDisk never return None but throws BlockException
           val iter = getLocalFromDisk(id, serializer).get
           // Pass 0 as size since it's not in flight
           results.put(new FetchResult(id, 0, () => iter))
+          fetchIndex += 1
           logDebug("Got local block " + id)
-        } catch {
-          case e: Exception => {
-            logError(s"Error occurred while fetch local block $id", e)
+        }
+      } catch {
+        case e: Exception => {
+          logError(s"Error occurred while fetching local blocks", e)
+          for (id <- localBlocksToFetch.drop(fetchIndex)) {
             results.put(new FetchResult(id, -1, null))
           }
         }