From 571a27a6a74199f2e35311988ec99a9992ceedf8 Mon Sep 17 00:00:00 2001 From: Holden Karau Date: Mon, 31 Aug 2026 07:49:01 +0000 Subject: [PATCH 1/5] [CORE] Skip over-long event log lines during History Server replay Replay materializes each event log line as a String with no length bound, so a corrupt or unexpectedly large log can exhaust the memory of the process replaying it. Cap lines at spark.history.fs.eventLog.maxLineLength (default 512m, <= 0 disables): longer lines are drained and skipped with a warning while the remaining events still replay. Co-authored-by: Cursor Co-Authored-By: Holden Karau --- .../history/EventLogFileCompactor.scala | 2 +- .../deploy/history/FsHistoryProvider.scala | 4 +- .../spark/internal/config/History.scala | 10 ++ .../spark/scheduler/ReplayListenerBus.scala | 100 +++++++++++++++++- .../spark/scheduler/ReplayListenerSuite.scala | 31 ++++++ 5 files changed, 140 insertions(+), 7 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/deploy/history/EventLogFileCompactor.scala b/core/src/main/scala/org/apache/spark/deploy/history/EventLogFileCompactor.scala index e0227641fc0eb..8d9be3877c78b 100644 --- a/core/src/main/scala/org/apache/spark/deploy/history/EventLogFileCompactor.scala +++ b/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 diff --git a/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala b/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala index 748aec37570d7..2a345c92d7a9f 100644 --- a/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala +++ b/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) diff --git a/core/src/main/scala/org/apache/spark/internal/config/History.scala b/core/src/main/scala/org/apache/spark/internal/config/History.scala index 570113d145a14..efcb5474fdad1 100644 --- a/core/src/main/scala/org/apache/spark/internal/config/History.scala +++ b/core/src/main/scala/org/apache/spark/internal/config/History.scala @@ -165,6 +165,16 @@ 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.") + .version("4.4.0") + .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. " + diff --git a/core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala b/core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala index 2e6cfa98ff373..7c8fded66abf3 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala +++ b/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 diff --git a/core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala index 1f71258bf8679..f4652c23f63e2 100644 --- a/core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala +++ b/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 From b01abf740baab0ff9bc3605e2c111f363e1b55d8 Mon Sep 17 00:00:00 2001 From: Holden Karau Date: Tue, 1 Sep 2026 04:59:21 +0000 Subject: [PATCH 2/5] [CORE][FOLLOWUP] Declare a binding policy for the replay line-length config Every new config entry needs one. A History Server replay setting cannot change how a view or UDF body resolves, so NOT_APPLICABLE. Co-Authored-By: Claude Opus 5 (1M context) Co-Authored-By: Holden Karau --- .../main/scala/org/apache/spark/internal/config/History.scala | 1 + 1 file changed, 1 insertion(+) diff --git a/core/src/main/scala/org/apache/spark/internal/config/History.scala b/core/src/main/scala/org/apache/spark/internal/config/History.scala index efcb5474fdad1..4f90afbc388af 100644 --- a/core/src/main/scala/org/apache/spark/internal/config/History.scala +++ b/core/src/main/scala/org/apache/spark/internal/config/History.scala @@ -172,6 +172,7 @@ private[spark] object History { "memory replay can use when an event log is corrupt or unexpectedly large. Setting " + "this to 0 or a negative value disables the limit.") .version("4.4.0") + .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE) .bytesConf(ByteUnit.BYTE) .createWithDefaultString("512m") From 4aa2fcc3542b3c954be5d41d3ba5725127fddaf3 Mon Sep 17 00:00:00 2001 From: Holden Karau Date: Tue, 1 Sep 2026 18:14:46 +0000 Subject: [PATCH 3/5] [CORE][FOLLOWUP] Declare the introduced-in version the config actually ships in The config was marked 4.4.0, which is only true if this lands on the 4.x line and nowhere else. It is being backported to the maintenance branches, so the earliest release a user can get it in is 4.1.4, and that is what the version field is for -- 3.5 and 4.0 name their own next patch, everything from 4.1 up names 4.1.4. Same shape as the UDT loading configs, which read 4.1.3 on master for the same reason. Co-Authored-By: Claude Opus 5 (1M context) Co-Authored-By: Holden Karau --- .../main/scala/org/apache/spark/internal/config/History.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/src/main/scala/org/apache/spark/internal/config/History.scala b/core/src/main/scala/org/apache/spark/internal/config/History.scala index 4f90afbc388af..ed91413dd42bd 100644 --- a/core/src/main/scala/org/apache/spark/internal/config/History.scala +++ b/core/src/main/scala/org/apache/spark/internal/config/History.scala @@ -171,7 +171,7 @@ private[spark] object History { "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.") - .version("4.4.0") + .version("4.1.4") .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE) .bytesConf(ByteUnit.BYTE) .createWithDefaultString("512m") From 02eb343eedd7e1d15de14719c0aa2c51a2438725 Mon Sep 17 00:00:00 2001 From: Holden Karau Date: Wed, 2 Sep 2026 18:09:52 +0000 Subject: [PATCH 4/5] [CORE][FOLLOWUP] Say which releases carry the config, not just one version A single version tells an operator nothing about a backported config: it does not say which maintenance releases have it. Name the whole set in the doc, and declare the version this branch actually first ships in. Co-Authored-By: Claude Opus 5 (1M context) Co-Authored-By: Holden Karau --- .../scala/org/apache/spark/internal/config/History.scala | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/internal/config/History.scala b/core/src/main/scala/org/apache/spark/internal/config/History.scala index ed91413dd42bd..7353481b6582f 100644 --- a/core/src/main/scala/org/apache/spark/internal/config/History.scala +++ b/core/src/main/scala/org/apache/spark/internal/config/History.scala @@ -170,8 +170,10 @@ private[spark] object History { .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.") - .version("4.1.4") + "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") From c9b6cb1ef28462faefbc018c69faa105d91ed557 Mon Sep 17 00:00:00 2001 From: Holden Karau Date: Thu, 3 Sep 2026 00:41:38 +0000 Subject: [PATCH 5/5] [CORE][FOLLOWUP] Document the config in the operator table The version note only reached the config doc, so nobody reading the docs tables saw it. Add the row. Co-Authored-By: Claude Opus 5 (1M context) Co-Authored-By: Holden Karau --- docs/monitoring.md | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/docs/monitoring.md b/docs/monitoring.md index 7b8d61c525f34..6104de5313f4c 100644 --- a/docs/monitoring.md +++ b/docs/monitoring.md @@ -438,6 +438,19 @@ Security options for the Spark History Server are covered more detail in the 3.0.0 + + spark.history.fs.eventLog.maxLineLength + 512m + + 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.
+ 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. + + 4.3.0 + spark.history.fs.eventLog.rolling.maxFilesToRetain Int.MaxValue