[SPARK-59804][UI] Fix UTF-8 byte limits and physical line numbers during event log replay

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

@@ -165,18 +165,25 @@ private[spark] object History {
       .bytesConf(ByteUnit.BYTE)
       .createWithDefaultString("1m")
 
+  val EVENT_LOG_MAX_LINE_LENGTH_LIMIT: Int = 512 * 1024 * 1024
+
   val EVENT_LOG_MAX_LINE_LENGTH =
     ConfigBuilder("spark.history.fs.eventLog.maxLineLength")
-      .doc("Maximum length of a single event log line during replay. Lines longer than " +
-        "this are skipped with a warning instead of being read into memory, bounding the " +
-        "memory replay can use when an event log is corrupt or unexpectedly large. Setting " +
-        "this to 0 or a negative value disables the limit. " +
+      .doc("Maximum UTF-8 byte length of a single event log line during replay, excluding " +
+        "the line ending. Longer lines are skipped with a warning, bounding the " +
+        "memory replay can use when an event log is corrupt or unexpectedly large. Buffer " +
+        "growth, UTF-16 storage and JSON parsing can require several times this limit in heap " +
+        "space. Buffer capacity grows in steps: reducing 256m to 200m or 150m may not reduce " +
+        "the retained buffer; use 128m to reach a smaller capacity. Values at or below 0, " +
+        s"or above $EVENT_LOG_MAX_LINE_LENGTH_LIMIT, use the maximum supported limit of " +
+        s"$EVENT_LOG_MAX_LINE_LENGTH_LIMIT bytes (512 MiB). This cap avoids JVM array-size " +
+        "limits but does not guarantee sufficient heap space. " +
         "Introduced in 4.3.0; also available in 3.5.10, 4.0.5, 4.1.4 and 4.2.1; and in " +
         "all versions after 4.3.0.")
       .version("4.3.0")
       .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
       .bytesConf(ByteUnit.BYTE)
-      .createWithDefaultString("512m")
+      .createWithDefaultString("256m")
 
   private[spark] val EVENT_LOG_ROLLING_MAX_FILES_TO_RETAIN =
     ConfigBuilder("spark.history.fs.eventLog.rolling.maxFilesToRetain")

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

@@ -22,7 +22,7 @@ import java.nio.charset.{CodingErrorAction, StandardCharsets}
 
 import scala.annotation.tailrec
 
-import com.fasterxml.jackson.core.JsonParseException
+import com.fasterxml.jackson.core.{JsonParseException, JsonProcessingException}
 import com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException
 
 import org.apache.spark.SparkConf
@@ -35,15 +35,21 @@ import org.apache.spark.util.JsonProtocol
 /**
  * A SparkListenerBus that can be used to replay events from serialized event data.
  *
- * @param maxLineLength Maximum number of byes (1/2 chars) of a single event log line that will be
- *                      materialized during replay. Longer lines are drained, skipped and
- *                      logged, bounding the memory replay can use when an event log is
- *                      corrupt or unexpectedly large.
+ * @param maxLineLength Maximum UTF-8 byte length of a single event log line, excluding its line
+ *                      ending. Longer lines are drained, skipped and logged, bounding the
+ *                      memory replay can use when an event log is corrupt or unexpectedly large.
+ *                      Values at or below zero or above MAX_LINE_LENGTH use MAX_LINE_LENGTH
+ *                      (512 MiB).
  */
 private[spark] class ReplayListenerBus(
     maxLineLength: Int = ReplayListenerBus.DEFAULT_MAX_LINE_LENGTH)
   extends SparkListenerBus with Logging {
 
+  def this() = this(ReplayListenerBus.DEFAULT_MAX_LINE_LENGTH)
+
+  private[scheduler] val effectiveMaxLineLength =
+    ReplayListenerBus.normalizeMaxLineLength(maxLineLength)
+
   /**
    * Replay each event in the order maintained in the given stream. The stream is expected to
    * contain one JSON-encoded SparkListenerEvent per line.
@@ -67,22 +73,26 @@ private[spark] class ReplayListenerBus(
       maybeTruncated: Boolean = false,
       eventsFilter: ReplayEventsFilter = SELECT_ALL_FILTER): Boolean = {
     val lines = boundedLines(logData, sourceName)
-    replay(lines, sourceName, maybeTruncated, eventsFilter)
+    replayEntries(lines, sourceName, maybeTruncated, eventsFilter)
   }
 
   /**
-   * Reads '\n'-terminated lines like Source.getLines(), but never materializes more than
-   * `maxLineLength` bytes of a single line. An over-long line is drained and skipped
-   * with a warning instead of being turned into a String.
+   * Reads '\n'-terminated lines and retains their original zero-based indices. The limit
+   * measures UTF-8 content bytes, excluding the line ending, rather than JVM heap usage.
+   * The character buffer is bounded by the limit plus one possible trailing CR. An over-long
+   * line is drained and skipped with a warning instead of being turned into a String.
    */
-  private def boundedLines(logData: InputStream, sourceName: String): Iterator[String] = {
+  private def boundedLines(
+      logData: InputStream,
+      sourceName: String): Iterator[(String, Int)] = {
     // Fail on malformed input like Source.getLines() does instead of replacing it.
     val decoder = StandardCharsets.UTF_8.newDecoder()
       .onMalformedInput(CodingErrorAction.REPORT)
       .onUnmappableCharacter(CodingErrorAction.REPORT)
     val reader = new BufferedReader(new InputStreamReader(logData, decoder))
-    new Iterator[String] {
-      private var nextLine: String = _
+    new Iterator[(String, Int)] {
+      private var nextLine: (String, Int) = _
+      private var lineIndex = 0
       private var lineFetched = false
       private var warned = false
 
@@ -94,7 +104,7 @@ private[spark] class ReplayListenerBus(
         nextLine != null
       }
 
-      override def next(): String = {
+      override def next(): (String, Int) = {
         if (!hasNext) {
           throw new NoSuchElementException("No more lines")
         }
@@ -104,35 +114,30 @@ private[spark] class ReplayListenerBus(
         line
       }
 
-      @tailrec private def fetchLine(): String = {
-        val sb = new java.lang.StringBuilder()
-        var overLong = false
+      @tailrec private def fetchLine(): (String, Int) = {
+        val buffer = new BoundedLineBuffer(effectiveMaxLineLength)
         var c = reader.read()
         if (c == -1) {
           null
         } else {
+          val index = lineIndex
+          lineIndex += 1
           while (c != -1 && c != '\n') {
-            if (sb.length() * 2 < maxLineLength) {
-              sb.append(c.toChar)
-            } else {
-              overLong = true
-            }
+            buffer.append(c.toChar)
             c = reader.read()
           }
-          if (overLong) {
+          val maybeLine = buffer.result()
+          if (maybeLine.isEmpty) {
             if (!warned) {
               logWarning(log"Skipped event log lines longer than " +
-                log"${MDC(MAX_SIZE, maxLineLength)} characters in " +
-                log"${MDC(FILE_NAME, sourceName)}")
+                log"${MDC(MAX_SIZE, effectiveMaxLineLength)} bytes in " +
+                log"${MDC(FILE_NAME, sourceName)}; first skipped line: ${MDC(LINE_NUM, index + 1)}")
               warned = true
             }
+            logDebug(s"Skipped event log line ${index + 1} in $sourceName")
             fetchLine()
           } else {
-            // Handle CRLF line endings like Source.getLines() does.
-            if (sb.length() > 0 && sb.charAt(sb.length() - 1) == '\r') {
-              sb.setLength(sb.length() - 1)
-            }
-            sb.toString
+            (maybeLine.get, index)
           }
         }
       }
@@ -148,15 +153,21 @@ private[spark] class ReplayListenerBus(
       sourceName: String,
       maybeTruncated: Boolean,
       eventsFilter: ReplayEventsFilter): Boolean = {
+    replayEntries(lines.zipWithIndex, sourceName, maybeTruncated, eventsFilter)
+  }
+
+  private def replayEntries(
+      lines: Iterator[(String, Int)],
+      sourceName: String,
+      maybeTruncated: Boolean,
+      eventsFilter: ReplayEventsFilter): Boolean = {
     var currentLine: String = null
     var lineNumber: Int = 0
     val unrecognizedEvents = new scala.collection.mutable.HashSet[String]
     val unrecognizedProperties = new scala.collection.mutable.HashSet[String]
 
     try {
-      val lineEntries = lines
-        .zipWithIndex
-        .filter { case (line, _) => eventsFilter(line) }
+      val lineEntries = lines.filter { case (line, _) => eventsFilter(line) }
 
       while (lineEntries.hasNext) {
         try {
@@ -202,11 +213,18 @@ private[spark] class ReplayListenerBus(
         // Just stop replay.
         false
       case _: EOFException if maybeTruncated => false
+      case jpe: JsonProcessingException =>
+        logError(log"Exception parsing Spark event log: ${MDC(PATH, sourceName)} " +
+          log"at line ${MDC(LINE_NUM, lineNumber)}")
+        throw jpe
       case ioe: IOException =>
         throw ioe
       case e: Exception =>
         logError(log"Exception parsing Spark event log: ${MDC(PATH, sourceName)}", e)
-        logError(log"Malformed line #${MDC(LINE_NUM, lineNumber)}: ${MDC(LINE, currentLine)}\n")
+        val prefix = Option(currentLine).map(_.take(1024)).orNull
+        val length = Option(currentLine).fold(0)(_.length)
+        logError(log"Malformed line #${MDC(LINE_NUM, lineNumber)}: ${MDC(LINE, prefix)} " +
+          log"(line length: ${MDC(SIZE, length)} characters; showing at most 1024)\n")
         false
     }
   }
@@ -225,21 +243,62 @@ private[spark] class HaltReplayException extends RuntimeException
 
 private[spark] object ReplayListenerBus {
 
-  /**
-   * Default per-line cap during replay: far above any legitimate event line, so replay
-   * memory stays bounded even for corrupt logs. Matches the default of
-   * spark.history.fs.eventLog.maxLineLength.
-   */
-  val DEFAULT_MAX_LINE_LENGTH: Int = 512 * 1024 * 1024
+  /** Default UTF-8 content limit shared with the history server configuration. */
+  val DEFAULT_MAX_LINE_LENGTH: Int = History.EVENT_LOG_MAX_LINE_LENGTH.defaultValue.get.toInt
+
+  // Bound StringBuilder growth so UTF-16 inflation stays below the JVM array-size limit.
+  val MAX_LINE_LENGTH: Int = History.EVENT_LOG_MAX_LINE_LENGTH_LIMIT
 
-  /** Resolves the replay line-length cap from configuration; <= 0 disables the cap. */
+  /** Resolves the byte limit, using 512 MiB for non-positive or larger values. */
   def maxLineLength(conf: SparkConf): Int = {
-    val configured = conf.get(History.EVENT_LOG_MAX_LINE_LENGTH)
-    if (configured <= 0 || configured > Int.MaxValue) Int.MaxValue else configured.toInt
+    normalizeMaxLineLength(conf.get(History.EVENT_LOG_MAX_LINE_LENGTH))
+  }
+
+  private def normalizeMaxLineLength(configured: Long): Int = {
+    if (configured <= 0 || configured > MAX_LINE_LENGTH) MAX_LINE_LENGTH else configured.toInt
   }
 
   type ReplayEventsFilter = (String) => Boolean
 
   // utility filter that selects all event logs during replay
   val SELECT_ALL_FILTER: ReplayEventsFilter = { (eventString: String) => true }
+
+  /** Accumulates a bounded prefix of one decoded line, allowing a possible trailing CR. */
+  private[scheduler] class BoundedLineBuffer(maxLineLength: Int) {
+    private val buffer = new java.lang.StringBuilder()
+    private var byteLength = 0L
+    private val bufferLimit = maxLineLength.toLong + 1
+
+    def length: Int = buffer.length()
+
+    def capacity: Int = buffer.capacity()
+
+    def append(c: Char): Unit = {
+      if (byteLength <= bufferLimit) {
+        // The strict decoder emits valid surrogate pairs: count two bytes per half.
+        byteLength += (if (c < 0x80) 1 else if (c < 0x800 || Character.isSurrogate(c)) 2 else 3)
+        // Reserve one extra byte for a possible CR in the line ending.
+        if (byteLength <= bufferLimit) {
+          buffer.append(c)
+        } else {
+          // The line will be skipped; release the retained prefix while draining.
+          buffer.setLength(0)
+          buffer.trimToSize()
+        }
+      }
+    }
+
+    def result(): Option[String] = {
+      val trailingCR = length > 0 && buffer.charAt(length - 1) == '\r'
+      val contentLength = byteLength - (if (trailingCR) 1 else 0)
+      if (contentLength > maxLineLength) {
+        None
+      } else if (trailingCR) {
+        Some(buffer.substring(0, length - 1))
+      } else {
+        Some(buffer.toString)
+      }
+    }
+  }
+
 }