This is an automated email from the ASF dual-hosted git repository.
viirya pushed a commit to branch branch-3.5
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.5 by this push:
new f2be14e9a51b [SPARK-59804][3.5][UI] Fix UTF-8 byte limits and physical
line numbers during event log replay
f2be14e9a51b is described below
commit f2be14e9a51b2d9bb5a3b88a4fccf3611eaad857
Author: Liang-Chi Hsieh <[email protected]>
AuthorDate: Wed Sep 30 17:30:52 2026 -0700
[SPARK-59804][3.5][UI] Fix UTF-8 byte limits and physical line numbers
during event log replay
### What changes were proposed in this pull request?
Backport #59074 (SPARK-59804, master commit
719947a5652f83a80492db50a6f1e5fe14d9e6b8) to `branch-3.5`.
Adapt the replay diagnostics to the string-based logging API in Spark 3.5,
preserving the messages and exception behavior. Add explicit filter parameter
types where Scala 2.12 cannot infer them for overloaded replay calls. Retain
the branch's `3.5.10` configuration version.
The backport includes UTF-8 byte accounting, physical-line diagnostics, the
256 MiB default and 512 MiB maximum, the restored no-argument constructor,
early release of skipped-line buffers, bounded malformed-line excerpts,
documentation, and regression tests. Independent follow-ups from the original
PR remain outside this backport.
### Why are the changes needed?
`branch-3.5` contains SPARK-59407 (8defa3c9418), so it has the same
byte-limit, diagnostic, and constructor regressions fixed by #59074. The master
patch needs logging API adaptation on this branch.
### Does this PR introduce _any_ user-facing change?
Yes, the same fixes as #59074: limits measure UTF-8 content bytes excluding
line endings; the default is 256 MiB; non-positive and above-maximum settings
resolve to 512 MiB. Replay preserves physical line numbers across skipped
lines, bounds malformed-line excerpts, and restores the JVM no-argument
constructor.
### How was this patch tested?
On branch-3.5 with Java 17.0.16 and Scala 2.12.18, using the company Maven
proxy:
```
build/sbt 'core/testOnly *ReplayListenerSuite'
build/sbt 'core/Compile/scalastyle' 'core/Test/scalastyle'
```
All 19 tests and both scalastyle checks passed. The tests cover UTF-8
boundaries, normalized constructor replay, reflective no-argument construction,
buffer release, physical-line diagnostics, bounded error excerpts, and IO
exception handling. Existing compressed/end-to-end tests retain the assertion
limitations documented in #59074; their independent follow-up remains outside
this backport.
Also checked the compiled class with `javap`: both constructors and the
Scala default-argument method are present.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-6)
Closes #59166 from viirya/codex/replay-line-limits-35.
Authored-by: Liang-Chi Hsieh <[email protected]>
Signed-off-by: Liang-Chi Hsieh <[email protected]>
---
.../org/apache/spark/internal/config/History.scala | 17 +-
.../apache/spark/scheduler/ReplayListenerBus.scala | 144 ++++++++++----
.../spark/scheduler/ReplayListenerSuite.scala | 219 +++++++++++++++++++++
docs/monitoring.md | 14 +-
4 files changed, 342 insertions(+), 52 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 6d5c5b80f354..d620dd38a734 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
@@ -122,17 +122,24 @@ 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("3.5.10")
.bytesConf(ByteUnit.BYTE)
- .createWithDefaultString("512m")
+ .createWithDefaultString("256m")
private[spark] val EVENT_LOG_ROLLING_MAX_FILES_TO_RETAIN =
ConfigBuilder("spark.history.fs.eventLog.rolling.maxFilesToRetain")
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 c38561f45a88..e715fe5969f5 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala
+++ b/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
@@ -34,15 +34,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 bytes (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.
@@ -66,22 +72,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
@@ -93,7 +103,7 @@ private[spark] class ReplayListenerBus(
nextLine != null
}
- override def next(): String = {
+ override def next(): (String, Int) = {
if (!hasNext) {
throw new NoSuchElementException("No more lines")
}
@@ -103,34 +113,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(s"Skipped event log lines longer than $maxLineLength
" +
- s"characters in $sourceName")
+ logWarning("Skipped event log lines longer than " +
+ s"$effectiveMaxLineLength bytes in " +
+ s"$sourceName; first skipped line: ${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)
}
}
}
@@ -146,15 +152,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 {
@@ -199,11 +211,18 @@ private[spark] class ReplayListenerBus(
// Just stop replay.
false
case _: EOFException if maybeTruncated => false
+ case jpe: JsonProcessingException =>
+ logError(s"Exception parsing Spark event log: $sourceName " +
+ s"at line $lineNumber")
+ throw jpe
case ioe: IOException =>
throw ioe
case e: Exception =>
logError(s"Exception parsing Spark event log: $sourceName", e)
- logError(s"Malformed line #$lineNumber: $currentLine\n")
+ val prefix = Option(currentLine).map(_.take(1024)).orNull
+ val length = Option(currentLine).fold(0)(_.length)
+ logError(s"Malformed line #$lineNumber: $prefix " +
+ s"(line length: $length characters; showing at most 1024)\n")
false
}
}
@@ -222,21 +241,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)
+ }
+ }
+ }
+
}
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 eba646fc9fae..7823afe8521e 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala
@@ -23,13 +23,17 @@ import java.util.concurrent.atomic.AtomicInteger
import scala.collection.mutable.ArrayBuffer
+import com.fasterxml.jackson.core.{JsonParseException, JsonProcessingException}
+import com.fasterxml.jackson.databind.exc.MismatchedInputException
import org.apache.hadoop.fs.Path
+import org.apache.logging.log4j.Level
import org.scalatest.BeforeAndAfter
import org.apache.spark._
import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.deploy.history.EventLogFileReader
import org.apache.spark.deploy.history.EventLogTestHelper._
+import org.apache.spark.internal.config.History
import org.apache.spark.io.{CompressionCodec, LZ4CompressionCodec}
import org.apache.spark.util.{JsonProtocol, JsonProtocolSuite, Utils}
@@ -109,6 +113,221 @@ class ReplayListenerSuite extends SparkFunSuite with
BeforeAndAfter with LocalSp
assert(eventMonster.loggedEvents(1) ===
JsonProtocol.sparkEventToJsonString(applicationEnd))
}
+ test("SPARK-59804: Replay line limit uses UTF-8 bytes") {
+ val start = SparkListenerApplicationStart("x" * 6000, None, 125L, "user",
None)
+ val json = JsonProtocol.sparkEventToJsonString(start)
+ assert(json.getBytes(StandardCharsets.UTF_8).length < 8 * 1024)
+ val listener = new EventBufferingListener
+ val conf = new SparkConf(false).set(History.EVENT_LOG_MAX_LINE_LENGTH.key,
"8k")
+ val bus = new ReplayListenerBus(ReplayListenerBus.maxLineLength(conf))
+ bus.addListener(listener)
+ val input = new ByteArrayInputStream(json.getBytes(StandardCharsets.UTF_8))
+ assert(bus.replay(input, "ascii"))
+ assert(listener.loggedEvents.toSeq == Seq(json))
+ }
+
+ test("SPARK-59804: Replay line limit handles raw UTF-8 boundaries and line
endings") {
+ // scalastyle:off nonascii
+ val suffixes = Seq("x", "\u00e9", "\u4e2d", "\ud83d\ude00")
+ // scalastyle:on nonascii
+ for {
+ suffix <- suffixes
+ padding <- Seq(8, 8190, 8191, 8192)
+ ending <- Seq("\n", "\r\n", "")
+ delta <- -3 to 2
+ } {
+ // Put the limit inside the final code point, and split UTF-8 sequences
across read buffers.
+ val line = "x" * padding + suffix
+ val bytes = line.getBytes(StandardCharsets.UTF_8)
+ val limit = bytes.length + delta
+ val seen = new ArrayBuffer[String]
+ val bus = new ReplayListenerBus(limit)
+ val input = new ByteArrayInputStream((line +
ending).getBytes(StandardCharsets.UTF_8))
+ withClue(s"codePoint=${suffix.codePointAt(0)} padding=$padding " +
+ s"ending=${ending.map(_.toInt).mkString(",")} delta=$delta: ") {
+ // Observe raw lines before JSON parsing, without letting Jackson
escape the test input.
+ assert(bus.replay(input, "utf8", eventsFilter = line => {
+ seen += line
+ false
+ }))
+ assert(seen.toSeq == (if (delta >= 0) Seq(line) else Seq.empty))
+ }
+ }
+ }
+
+ test("SPARK-59804: Replay line buffer stops retaining oversized content") {
+ for (limit <- Seq(0, 1, 8, 1024)) {
+ val buffer = new ReplayListenerBus.BoundedLineBuffer(limit)
+ for (_ <- 0 until 10000) {
+ buffer.append('x')
+ assert(buffer.length <= limit + 1)
+ }
+ assert(buffer.length == 0)
+ assert(buffer.capacity == 0)
+ assert(buffer.result().isEmpty)
+ }
+ }
+
+ test("SPARK-59804: Replay retains a public no-argument constructor") {
+ val bus = classOf[ReplayListenerBus].getConstructor().newInstance()
+ assert(bus.effectiveMaxLineLength ==
ReplayListenerBus.DEFAULT_MAX_LINE_LENGTH)
+ }
+
+ test("SPARK-59804: Replay reports the physical location of skipped lines") {
+ val end =
JsonProtocol.sparkEventToJsonString(SparkListenerApplicationEnd(1000L))
+ val input = new ByteArrayInputStream(
+ Seq(end, "x" * 2048, end, "x" * 2048).mkString("\n")
+ .getBytes(StandardCharsets.UTF_8))
+ val appender = new LogAppender
+ appender.setThreshold(Level.DEBUG)
+ withLogAppender(appender, level = Some(Level.DEBUG)) {
+ assert(new ReplayListenerBus(1024).replay(input, "skipped-lines"))
+ }
+ val warnings = appender.loggingEvents.filter(_.getLevel == Level.WARN)
+ assert(warnings.size == 1)
+ assert(warnings.head.getMessage.getFormattedMessage.contains("first
skipped line: 2"))
+ val debug = appender.loggingEvents.filter(_.getLevel == Level.DEBUG)
+ .map(_.getMessage.getFormattedMessage)
+ assert(debug.contains("Skipped event log line 2 in skipped-lines"))
+ assert(debug.contains("Skipped event log line 4 in skipped-lines"))
+ }
+
+ test("SPARK-59804: Replay bounds malformed line diagnostics") {
+ val line = """{"Event":"SparkListenerJobStart","Job ID":1,"Stage
IDs":[],""" +
+ """"Properties":{"bad":1},"padding":"""" + "x" * 10000 + "tail-marker\"}"
+ val appender = new LogAppender
+ withLogAppender(appender) {
+ assert(!new ReplayListenerBus().replay(
+ new ByteArrayInputStream(line.getBytes(StandardCharsets.UTF_8)),
"semantic-error"))
+ }
+ val diagnostic =
appender.loggingEvents.map(_.getMessage.getFormattedMessage)
+ .find(_.startsWith("Malformed line #1:")).get
+ assert(diagnostic.contains(s"line length: ${line.length} characters"))
+ assert(!diagnostic.contains("tail-marker"))
+ assert(diagnostic.length < 1200)
+ }
+
+ test("SPARK-59804: Replay line limit defaults and maximum supported
configuration") {
+ val conf = new SparkConf(false)
+ assert(conf.get(History.EVENT_LOG_MAX_LINE_LENGTH) == 256L * 1024 * 1024)
+ assert(ReplayListenerBus.maxLineLength(conf) ==
ReplayListenerBus.DEFAULT_MAX_LINE_LENGTH)
+ for (value <- Seq("0", "-1", "512m", "536870913", "576m", "600m", "1g",
+ "2147483647", "3g")) {
+ conf.set(History.EVENT_LOG_MAX_LINE_LENGTH.key, value)
+ assert(ReplayListenerBus.maxLineLength(conf) ==
ReplayListenerBus.MAX_LINE_LENGTH)
+ }
+ }
+
+ test("SPARK-59804: Replay constructor normalizes the effective line limit") {
+ val maximum = ReplayListenerBus.MAX_LINE_LENGTH
+ for (limit <- Seq(Int.MinValue, -1, 0, maximum, maximum + 1,
Int.MaxValue)) {
+ val bus = new ReplayListenerBus(limit)
+ assert(bus.effectiveMaxLineLength == maximum)
+ val listener = new EventBufferingListener
+ bus.addListener(listener)
+ val json =
JsonProtocol.sparkEventToJsonString(SparkListenerApplicationEnd(1000L))
+ assert(bus.replay(new
ByteArrayInputStream(json.getBytes(StandardCharsets.UTF_8)),
+ "normalized"))
+ assert(listener.loggedEvents.toSeq == Seq(json))
+ }
+ for (limit <- Seq(1, 8192, ReplayListenerBus.DEFAULT_MAX_LINE_LENGTH,
maximum - 1)) {
+ assert(new ReplayListenerBus(limit).effectiveMaxLineLength == limit)
+ val conf = new SparkConf(false).set(History.EVENT_LOG_MAX_LINE_LENGTH,
limit.toLong)
+ assert(ReplayListenerBus.maxLineLength(conf) == limit)
+ }
+ assert(new ReplayListenerBus().effectiveMaxLineLength ==
+ ReplayListenerBus.DEFAULT_MAX_LINE_LENGTH)
+ }
+
+ test("SPARK-59804: Replay preserves physical line numbers after skipping
long lines") {
+ val end =
JsonProtocol.sparkEventToJsonString(SparkListenerApplicationEnd(1000L))
+ val appender = new LogAppender
+ val bus = new ReplayListenerBus(1024)
+ val input = new ByteArrayInputStream(
+ (end + "\n" + "x" * 2048 + "\n" + "x" * 2048 + "\n{}\n")
+ .getBytes(StandardCharsets.UTF_8))
+ withLogAppender(appender) {
+ assert(!bus.replay(input, "line-numbers"))
+ }
+ assert(appender.loggingEvents.exists(_.getMessage.getFormattedMessage
+ .contains("Malformed line #4: {}")))
+ }
+
+ test("SPARK-59804: Replay logs physical line numbers before rethrowing JSON
errors") {
+ val end =
JsonProtocol.sparkEventToJsonString(SparkListenerApplicationEnd(1000L))
+ val mappingError =
+
"""{"Event":"org.apache.spark.util.TestListenerEvent","foo":"x","bar":[]}"""
+ for ((malformed, errorClass) <- Seq(
+ ("{bad", classOf[JsonParseException]),
+ (mappingError, classOf[MismatchedInputException]))) {
+ val appender = new LogAppender
+ val bus = new ReplayListenerBus(1024)
+ val input = new ByteArrayInputStream(
+ Seq(end, "x" * 2048, "x" * 2048, malformed, end).mkString("\n")
+ .getBytes(StandardCharsets.UTF_8))
+ withLogAppender(appender) {
+ val error = intercept[JsonProcessingException] {
+ bus.replay(input, "json-errors", maybeTruncated = true)
+ }
+ assert(errorClass.isInstance(error))
+ }
+ val diagnostics =
appender.loggingEvents.filter(_.getMessage.getFormattedMessage
+ .contains("Exception parsing Spark event log: json-errors at line 4"))
+ assert(diagnostics.size == 1)
+ assert(diagnostics.head.getThrown == null)
+ }
+ }
+
+ test("SPARK-59804: Replay propagates stream IO errors without reporting a
stale parse line") {
+ val error = new IOException("read failure")
+ val end =
JsonProtocol.sparkEventToJsonString(SparkListenerApplicationEnd(1000L))
+ val input = new ByteArrayInputStream((end +
"\n").getBytes(StandardCharsets.UTF_8)) {
+ override def read(bytes: Array[Byte], offset: Int, length: Int): Int = {
+ if (available() == 0) throw error
+ super.read(bytes, offset, length)
+ }
+ }
+ val bus = new ReplayListenerBus()
+ val listener = new EventBufferingListener
+ bus.addListener(listener)
+ val appender = new LogAppender
+ withLogAppender(appender) {
+ assert(intercept[IOException] {
+ bus.replay(input, "read-error")
+ } eq error)
+ }
+ assert(listener.loggedEvents.toSeq == Seq(end))
+ assert(!appender.loggingEvents.exists(_.getMessage.getFormattedMessage
+ .contains("Exception parsing Spark event log")))
+ }
+
+ test("SPARK-59804: Replay preserves line numbers for truncated logs and
filtered iterators") {
+ val end =
JsonProtocol.sparkEventToJsonString(SparkListenerApplicationEnd(1000L))
+ val truncated = "{\"Event\":"
+ val appender = new LogAppender
+ val bus = new ReplayListenerBus(1024)
+ val input = new ByteArrayInputStream(
+ ("x" * 2048 + "\n" + end + "\n" +
truncated).getBytes(StandardCharsets.UTF_8))
+ withLogAppender(appender) {
+ assert(bus.replay(input, "truncated", maybeTruncated = true,
+ eventsFilter = (line: String) => line != end))
+ assert(bus.replay(Iterator(end, end, truncated), "iterator",
maybeTruncated = true,
+ eventsFilter = (line: String) => line != end))
+ }
+ val warnings = appender.loggingEvents.map(_.getMessage.getFormattedMessage)
+ .filter(_.contains("Got JsonParseException"))
+ assert(warnings.size == 2)
+ assert(warnings.forall(_.contains("at line 3,")))
+ }
+
+ test("SPARK-59804: Replay still rejects malformed UTF-8 in skipped lines") {
+ val bytes = Array.fill[Byte](20 * 1024)('x'.toByte) ++ Array(0xff.toByte,
'\n'.toByte)
+ val bus = new ReplayListenerBus(1024)
+ intercept[java.nio.charset.MalformedInputException] {
+ bus.replay(new ByteArrayInputStream(bytes), "invalid-utf8")
+ }
+ }
+
/**
* 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 088fe34f148f..2969667247a2 100644
--- a/docs/monitoring.md
+++ b/docs/monitoring.md
@@ -385,12 +385,16 @@ Security options for the Spark History Server are covered
more detail in the
</tr>
<tr>
<td>spark.history.fs.eventLog.maxLineLength</td>
- <td>512m</td>
+ <td>256m</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/>
+ Maximum UTF-8 byte length of a single event log line during replay,
excluding the line
+ ending. Longer lines are skipped with a warning, which bounds 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, or above 536870912, use the maximum supported limit of 536870912
bytes (512 MiB).
+ This cap avoids JVM array-size limits but does not guarantee sufficient
heap space.<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>
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
