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..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
@@ -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. " +
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
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 |