[CARMEL-7356][CARMEL-3130] Better fetch fail handling and MapOutputTracker improvement (apache#113)

Author: wangyum

common/network-common/src/main/java/org/apache/spark/network/util/NettyMemoryMetrics.java

@@ -26,9 +26,7 @@
 import com.codahale.metrics.MetricRegistry;
 import com.codahale.metrics.MetricSet;
 import com.google.common.annotations.VisibleForTesting;
-import io.netty.buffer.PoolArenaMetric;
-import io.netty.buffer.PooledByteBufAllocator;
-import io.netty.buffer.PooledByteBufAllocatorMetric;
+import io.netty.buffer.*;
 
 /**
  * A Netty memory metrics class to collect metrics from Netty PooledByteBufAllocator.
@@ -136,6 +134,34 @@ private void registerArenaMetric(PoolArenaMetric arenaMetric, String arenaName)
         });
       }
     }
+
+    String chunkMetricName = MetricRegistry.name(metricPrefix, arenaName, "chunksInfo");
+    allMetrics.put(chunkMetricName, (Gauge<String>) () -> {
+      try {
+        synchronized (arenaMetric) {
+          List<PoolChunkListMetric> chunkListMetricsList = arenaMetric.chunkLists();
+          StringBuilder builder = new StringBuilder();
+          for (PoolChunkListMetric chunkListMetrics : chunkListMetricsList) {
+            builder.append(System.lineSeparator());
+            builder.append("chunkList-" + chunkListMetrics.minUsage() + "-" +
+                chunkListMetrics.maxUsage() + ":");
+            builder.append(System.lineSeparator());
+            int chunkIdx = 0;
+            for (PoolChunkMetric chunkMetric : chunkListMetrics) {
+              String chunkName = "chunk-" + chunkIdx;
+              String chunkInfo = String.format(Locale.US,"%s: chunkSize:%s, freeBytes:%s, usage:%s",
+                  chunkName, chunkMetric.chunkSize(), chunkMetric.freeBytes(), chunkMetric.usage());
+              builder.append("\t" + chunkInfo);
+              builder.append(System.lineSeparator());
+              chunkIdx++;
+            }
+          }
+          return builder.toString();
+        }
+      } catch (Exception e) {
+        return ""; // Swallow the exceptions.
+      }
+    });
   }
 
   @Override

common/network-common/src/test/java/org/apache/spark/network/util/NettyMemoryMetricsSuite.java

@@ -136,7 +136,7 @@ public void testAdditionalMetrics() throws IOException, InterruptedException {
       Assert.assertTrue(name.startsWith("shuffle-server"));
       String metricName = name.substring(name.lastIndexOf(".") + 1);
       Assert.assertTrue(metricName.equals("usedDirectMemory")
-        || metricName.equals("usedHeapMemory")
+        || metricName.equals("usedHeapMemory") || metricName.equals("chunksInfo")
         || NettyMemoryMetrics.VERBOSE_METRICS.contains(metricName));
     });
 
@@ -145,7 +145,7 @@ public void testAdditionalMetrics() throws IOException, InterruptedException {
       Assert.assertTrue(name.startsWith("shuffle-client"));
       String metricName = name.substring(name.lastIndexOf(".") + 1);
       Assert.assertTrue(metricName.equals("usedDirectMemory")
-        || metricName.equals("usedHeapMemory")
+        || metricName.equals("usedHeapMemory") || metricName.equals("chunksInfo")
         || NettyMemoryMetrics.VERBOSE_METRICS.contains(metricName));
     });
 

core/src/main/scala/org/apache/spark/MapOutputTracker.scala

@@ -117,6 +117,14 @@ private class ShuffleStatus(
    */
   private[this] var cachedSerializedMapStatus: Array[Byte] = _
 
+  /**
+   * Sometimes map statuses may serialize fail, if too many reducers are waiting map statuses
+   * serialization, it will occupy all the MessageLoop threads, and block other map statuses
+   * output requests, so cache the exception so that other reducers' request can failed fast.
+   */
+  @volatile
+  private[this] var mapStatusSerializedException: Option[Throwable] = None
+
   /**
    * Broadcast variable holding serialized map output statuses array. When [[serializedMapStatus]]
    * serializes the map statuses array it may detect that the result is too large to send in a
@@ -349,15 +357,27 @@ private class ShuffleStatus(
     withReadLock {
       if (cachedSerializedMapStatus != null) {
         result = cachedSerializedMapStatus
+      } else if (mapStatusSerializedException.isDefined) {
+        throw mapStatusSerializedException.get
       }
     }
 
     if (result == null) withWriteLock {
       if (cachedSerializedMapStatus == null) {
-        val serResult = MapOutputTracker.serializeOutputStatuses[MapStatus](
-          mapStatuses, broadcastManager, isLocal, minBroadcastSize, conf)
-        cachedSerializedMapStatus = serResult._1
-        cachedSerializedBroadcast = serResult._2
+        if (mapStatusSerializedException.isDefined) {
+          throw mapStatusSerializedException.get
+        }
+        try {
+          val serResult = MapOutputTracker.serializeOutputStatuses[MapStatus](
+            mapStatuses, broadcastManager, isLocal, minBroadcastSize, conf)
+          cachedSerializedMapStatus = serResult._1
+          cachedSerializedBroadcast = serResult._2
+          mapStatusSerializedException = None
+        } catch {
+          case NonFatal(e) =>
+            mapStatusSerializedException = Some(e)
+            throw e
+        }
       }
       // The following line has to be outside if statement since it's possible that another thread
       // initializes cachedSerializedMapStatus in-between `withReadLock` and `withWriteLock`.
@@ -929,6 +949,13 @@ private[spark] class MapOutputTrackerMaster(
     incrementEpoch()
   }
 
+  def removeOutputsOnExecutorForShuffle(execId: String, shuffleId: Int): Unit = {
+    shuffleStatuses.get(shuffleId).foreach { shuffleStatus =>
+      shuffleStatus.removeOutputsOnExecutor(execId)
+    }
+    incrementEpoch()
+  }
+
   /** Check if the given shuffle is being tracked */
   def containsShuffle(shuffleId: Int): Boolean = shuffleStatuses.contains(shuffleId)
 

core/src/main/scala/org/apache/spark/internal/config/package.scala

@@ -955,6 +955,13 @@ package object config {
       .booleanConf
       .createWithDefault(false)
 
+  private[spark] val REMOVE_EXECUTOR_ON_FETCH_FAILURE =
+    ConfigBuilder("spark.files.fetchFailure.removeExecutor")
+      .doc("Whether to remove executor on a FetchFailure.")
+      .version("3.5.0")
+      .booleanConf
+      .createWithDefault(true)
+
   private[spark] val LISTENER_BUS_EVENT_QUEUE_CAPACITY =
     ConfigBuilder("spark.scheduler.listenerbus.eventqueue.capacity")
       .doc("The default capacity for event queues. Spark will try to initialize " +

core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala

@@ -41,7 +41,7 @@ import org.apache.spark.errors.SparkCoreErrors
 import org.apache.spark.executor.{ExecutorMetrics, TaskMetrics}
 import org.apache.spark.internal.Logging
 import org.apache.spark.internal.config
-import org.apache.spark.internal.config.{JOB_GROUP_MAX_SHUFFLE_SIZE, RDD_CACHE_VISIBILITY_TRACKING_ENABLED, TASK_SUBMISSION_ASYNC}
+import org.apache.spark.internal.config.{JOB_GROUP_MAX_SHUFFLE_SIZE, RDD_CACHE_VISIBILITY_TRACKING_ENABLED, REMOVE_EXECUTOR_ON_FETCH_FAILURE, TASK_SUBMISSION_ASYNC}
 import org.apache.spark.internal.config.Tests.TEST_NO_STAGE_RETRY
 import org.apache.spark.network.shuffle.{BlockStoreClient, MergeFinalizerListener}
 import org.apache.spark.network.shuffle.protocol.MergeStatuses
@@ -220,6 +220,8 @@ private[spark] class DAGScheduler(
   /** If enabled, FetchFailed will not cause stage retry, in order to surface the problem. */
   private val disallowStageRetryForTest = sc.getConf.get(TEST_NO_STAGE_RETRY)
 
+  private val removeExecutorOnFetchFailure = sc.getConf.get(REMOVE_EXECUTOR_ON_FETCH_FAILURE)
+
   private val shouldMergeResourceProfiles = sc.getConf.get(config.RESOURCE_PROFILE_MERGE_CONFLICTS)
 
   private val groupToShuffleSize = new ConcurrentHashMap[String, AtomicLong].asScala
@@ -611,6 +613,7 @@ private[spark] class DAGScheduler(
       mapOutputTracker.registerShuffle(shuffleDep.shuffleId, rdd.partitions.length,
         shuffleDep.partitioner.numPartitions)
     }
+    log.info(s"Create MapStage:$id with shuffle ${shuffleDep.shuffleId}")
     stage
   }
 
@@ -2387,18 +2390,25 @@ private[spark] class DAGScheduler(
               // reason to believe shuffle data has been lost for the entire host).
               None
             }
-            removeExecutorAndUnregisterOutputs(
-              execId = bmAddress.executorId,
-              fileLost = true,
-              hostToUnregisterOutputs = hostToUnregisterOutputs,
-              maybeEpoch = Some(task.epoch),
-              // shuffleFileLostEpoch is ignored when a host is decommissioned because some
-              // decommissioned executors on that host might have been removed before this fetch
-              // failure and might have bumped up the shuffleFileLostEpoch. We ignore that, and
-              // proceed with unconditional removal of shuffle outputs from all executors on that
-              // host, including from those that we still haven't confirmed as lost due to heartbeat
-              // delays.
-              ignoreShuffleFileLostEpoch = isHostDecommissioned)
+            if (removeExecutorOnFetchFailure) {
+              removeExecutorAndUnregisterOutputs(
+                execId = bmAddress.executorId,
+                fileLost = true,
+                hostToUnregisterOutputs = hostToUnregisterOutputs,
+                maybeEpoch = Some(task.epoch),
+                // shuffleFileLostEpoch is ignored when a host is decommissioned because some
+                // decommissioned executors on that host might have been removed before this fetch
+                // failure and might have bumped up the shuffleFileLostEpoch. We ignore that, and
+                // proceed with unconditional removal of shuffle outputs from all executors on that
+                // host, including from those that we still haven't confirmed as lost due to
+                // heartbeat delays.
+                ignoreShuffleFileLostEpoch = isHostDecommissioned)
+            } else {
+              log.info(s"Remove all map status for the shuffle $shuffleId " +
+                s"on the executor: ${bmAddress.executorId}, because of fetch failed")
+              mapOutputTracker.removeOutputsOnExecutorForShuffle(bmAddress.executorId, shuffleId)
+              clearCacheLocs()
+            }
           }
         }
 

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

@@ -312,11 +312,18 @@ final class ShuffleBlockFetcherIterator(
             // Increment the ref count because we need to pass this to a different thread.
             // This needs to be released after use.
             buf.retain()
+            val returnBytes = buf.size()
+            val expectBytes = infoMap(blockId)._1
+            if (log.isDebugEnabled && returnBytes != expectBytes) {
+              logDebug(s"task:${context.taskAttemptId()}," +
+                s"block:$blockId, return bytes:$returnBytes," +
+                s"diff:${returnBytes - expectBytes}")
+            }
             remainingBlocks -= blockId
             blockOOMRetryCounts.remove(blockId)
             updateMergedReqsDuration(BlockId(blockId).isShuffleChunk)
             results.put(SuccessFetchResult(BlockId(blockId), infoMap(blockId)._2,
-              address, infoMap(blockId)._1, buf, remainingBlocks.isEmpty))
+              address, expectBytes, buf, remainingBlocks.isEmpty))
             logDebug("remainingBlocks: " + remainingBlocks)
             enqueueDeferredFetchRequestIfNecessary()
           } else {