[SPARK-59407][UI][HISTORY] Improve compressed line handling in Spark History server

core/src/main/scala/org/apache/spark/deploy/history/EventLogFileCompactor.scala

@@ -113,7 +113,7 @@ class EventLogFileCompactor(
    * them via replaying events in given files.
    */
   private def initializeBuilders(fs: FileSystem, files: Seq[Path]): Seq[EventFilterBuilder] = {
-    val bus = new ReplayListenerBus()
+    val bus = new ReplayListenerBus(ReplayListenerBus.maxLineLength(sparkConf))
 
     val builders = ServiceLoader.load(classOf[EventFilterBuilder],
       Utils.getContextOrSparkClassLoader).asScala.toSeq

core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala

@@ -1075,7 +1075,7 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
     val shouldHalt = enableOptimizations &&
       ((!appCompleted && fastInProgressParsing) || reparseChunkSize > 0)
 
-    val bus = new ReplayListenerBus()
+    val bus = new ReplayListenerBus(ReplayListenerBus.maxLineLength(conf))
     val listener = new AppListingListener(reader, clock, shouldHalt, this)
     bus.addListener(listener)
 
@@ -1427,7 +1427,7 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
     // to parse the event logs in the SHS.
     val replayConf = conf.clone().set(ASYNC_TRACKING_ENABLED, false)
     val trackingStore = new ElementTrackingStore(store, replayConf)
-    val replayBus = new ReplayListenerBus()
+    val replayBus = new ReplayListenerBus(ReplayListenerBus.maxLineLength(conf))
     val listener = new AppStatusListener(trackingStore, replayConf, false,
       lastUpdateTime = Some(lastUpdated))
     replayBus.addListener(listener)

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

@@ -165,6 +165,19 @@ private[spark] object History {
       .bytesConf(ByteUnit.BYTE)
       .createWithDefaultString("1m")
 
+  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. " +
+        "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")
+
   private[spark] val EVENT_LOG_ROLLING_MAX_FILES_TO_RETAIN =
     ConfigBuilder("spark.history.fs.eventLog.rolling.maxFilesToRetain")
       .doc("The maximum number of event log files which will be retained as non-compacted. " +

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

@@ -17,22 +17,32 @@
 
 package org.apache.spark.scheduler
 
-import java.io.{EOFException, InputStream, IOException}
+import java.io.{BufferedReader, EOFException, InputStream, InputStreamReader, IOException}
+import java.nio.charset.{CodingErrorAction, StandardCharsets}
 
-import scala.io.{Codec, Source}
+import scala.annotation.tailrec
 
 import com.fasterxml.jackson.core.JsonParseException
 import com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException
 
+import org.apache.spark.SparkConf
 import org.apache.spark.internal.Logging
 import org.apache.spark.internal.LogKeys._
+import org.apache.spark.internal.config.History
 import org.apache.spark.scheduler.ReplayListenerBus._
 import org.apache.spark.util.JsonProtocol
 
 /**
  * A SparkListenerBus that can be used to replay events from serialized event data.
+ *
+ * @param maxLineLength Maximum number of characters 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.
  */
-private[spark] class ReplayListenerBus extends SparkListenerBus with Logging {
+private[spark] class ReplayListenerBus(
+    maxLineLength: Int = ReplayListenerBus.DEFAULT_MAX_LINE_LENGTH)
+  extends SparkListenerBus with Logging {
 
   /**
    * Replay each event in the order maintained in the given stream. The stream is expected to
@@ -56,10 +66,79 @@ private[spark] class ReplayListenerBus extends SparkListenerBus with Logging {
       sourceName: String,
       maybeTruncated: Boolean = false,
       eventsFilter: ReplayEventsFilter = SELECT_ALL_FILTER): Boolean = {
-    val lines = Source.fromInputStream(logData)(Codec.UTF8).getLines()
+    val lines = boundedLines(logData, sourceName)
     replay(lines, sourceName, maybeTruncated, eventsFilter)
   }
 
+  /**
+   * Reads '\n'-terminated lines like Source.getLines(), but never materializes more than
+   * [[maxLineLength]] characters of a single line. 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] = {
+    // 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 = _
+      private var lineFetched = false
+      private var warned = false
+
+      override def hasNext: Boolean = {
+        if (!lineFetched) {
+          nextLine = fetchLine()
+          lineFetched = true
+        }
+        nextLine != null
+      }
+
+      override def next(): String = {
+        if (!hasNext) {
+          throw new NoSuchElementException("No more lines")
+        }
+        val line = nextLine
+        nextLine = null
+        lineFetched = false
+        line
+      }
+
+      @tailrec private def fetchLine(): String = {
+        val sb = new java.lang.StringBuilder()
+        var overLong = false
+        var c = reader.read()
+        if (c == -1) {
+          null
+        } else {
+          while (c != -1 && c != '\n') {
+            if (sb.length() < maxLineLength) {
+              sb.append(c.toChar)
+            } else {
+              overLong = true
+            }
+            c = reader.read()
+          }
+          if (overLong) {
+            if (!warned) {
+              logWarning(log"Skipped event log lines longer than " +
+                log"${MDC(MAX_SIZE, maxLineLength)} characters in " +
+                log"${MDC(FILE_NAME, sourceName)}")
+              warned = true
+            }
+            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
+          }
+        }
+      }
+    }
+  }
+
   /**
    * Overloaded variant of [[replay()]] which accepts an iterator of lines instead of an
    * [[InputStream]]. Exposed for use by custom ApplicationHistoryProvider implementations.
@@ -146,6 +225,19 @@ 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
+
+  /** Resolves the replay line-length cap from configuration; <= 0 disables the cap. */
+  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
+  }
+
   type ReplayEventsFilter = (String) => Boolean
 
   // utility filter that selects all event logs during replay

core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala

@@ -78,6 +78,37 @@ class ReplayListenerSuite extends SparkFunSuite with BeforeAndAfter with LocalSp
     assert(eventMonster.loggedEvents(1) === JsonProtocol.sparkEventToJsonString(applicationEnd))
   }
 
+  test("Over-long event log lines are skipped instead of materialized") {
+    val logFilePath = getFilePath(testDir, "events.txt")
+    val fstream = fileSystem.create(logFilePath)
+    val fwriter = new OutputStreamWriter(fstream, StandardCharsets.UTF_8)
+    val applicationStart = SparkListenerApplicationStart("Greatest App (N)ever", None,
+      125L, "Mickey", None)
+    val applicationEnd = SparkListenerApplicationEnd(1000L)
+    val maxLineLength = 8 * 1024
+    val hugeLine = "x" * (maxLineLength + 1024)
+    Utils.tryWithResource(new PrintWriter(fwriter)) { writer =>
+      // scalastyle:off println
+      writer.println(JsonProtocol.sparkEventToJsonString(applicationStart))
+      writer.println(hugeLine)
+      writer.println(JsonProtocol.sparkEventToJsonString(applicationEnd))
+      // scalastyle:on println
+    }
+
+    val logData = fileSystem.open(logFilePath)
+    val eventMonster = new EventBufferingListener
+    try {
+      val replayer = new ReplayListenerBus(maxLineLength = maxLineLength)
+      replayer.addListener(eventMonster)
+      assert(replayer.replay(logData, logFilePath.toString))
+    } finally {
+      logData.close()
+    }
+    assert(eventMonster.loggedEvents.size === 2)
+    assert(eventMonster.loggedEvents(0) === JsonProtocol.sparkEventToJsonString(applicationStart))
+    assert(eventMonster.loggedEvents(1) === JsonProtocol.sparkEventToJsonString(applicationEnd))
+  }
+
   /**
    * Test replaying compressed spark history file that internally throws an EOFException.  To
    * avoid sensitivity to the compression specifics the test forces an EOFException to occur

docs/monitoring.md

@@ -438,6 +438,19 @@ Security options for the Spark History Server are covered more detail in the
     </td>
     <td>3.0.0</td>
   </tr>
+  <tr>
+    <td>spark.history.fs.eventLog.maxLineLength</td>
+    <td>512m</td>
+    <td>
+      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, which bounds 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.<br/>
+      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.
+    </td>
+    <td>4.3.0</td>
+  </tr>
   <tr>
     <td>spark.history.fs.eventLog.rolling.maxFilesToRetain</td>
     <td>Int.MaxValue</td>