This is an automated email from the ASF dual-hosted git repository.
viirya pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new d99a2d9766cb [SPARK-59804][UI] Fix UTF-8 byte limits and physical line
numbers during event log replay
d99a2d9766cb is described below
commit d99a2d9766cbc91518f2fb6b9d6bcf8e231850e9
Author: Liang-Chi Hsieh <[email protected]>
AuthorDate: Wed Sep 30 11:21:03 2026 -0700
[SPARK-59804][UI] Fix UTF-8 byte limits and physical line numbers during
event log replay
Fix the UTF-8 byte limit and physical-line diagnostics introduced by
SPARK-59407 (#58700):
- Count event-log content in UTF-8 bytes, excluding line endings, while
retaining only a bounded prefix of oversized lines.
- Carry physical line indices across skipped lines, and log the source and
line without a duplicate stack trace before rethrowing JSON parsing or mapping
exceptions.
- Reduce the default from `512m` to `256m`, keeping the worst-case buffered
character count close to the previous implementation's effective default bound.
- Clamp the effective UTF-8 content limit to `512m` in both configuration
resolution and the reader constructor, and document the heap overhead. Derive
the reader default from the History configuration.
- Restore the public JVM no-argument constructor while retaining the Scala
default argument.
- Release retained prefixes once a line is known to be oversized, identify
skipped-line locations, and limit malformed-line excerpts to 1,024 characters.
- Document stepped buffer growth and centralize the maximum limit in
History.
The iterator-based replay API, strict UTF-8 decoding, existing
truncated-last-line handling and ordinary input-stream IOException behavior are
preserved. Independent follow-ups will address the unbounded compaction-filter
and tail-reparse paths, tail-relative diagnostics and UTF-8 alignment,
incomplete UTF-8 at EOF, bulk-read performance, and the pre-existing end-to-end
test assertions.
JIRA: https://issues.apache.org/jira/browse/SPARK-59804
Release coordination: 4.2.1 RC1 contains the previous implementation. This
has been raised on the Spark dev mailing list to discuss an RC2 containing the
reviewed fix; no migration entry or final release-version decision is made in
this PR pending that discussion.
The original reader compares `StringBuilder.length() * 2` with a
byte-valued setting. With an 8 KiB limit, it incorrectly skips a valid ASCII
application-start event with a 6,000-character application name. UTF-8
multibyte characters and CRLF endings also produce inconsistent boundaries.
Dropping oversized lines before `zipWithIndex` renumbers subsequent
diagnostics. For example, a valid event on line 1, oversized lines 2 and 3, and
an invalid event on line 4 report line 2. JSON processing exceptions also need
an explicit diagnostic before they are rethrown to retain that physical
location.
Correcting the unit without adjusting the default permits roughly twice as
many buffered characters. With a lazy line containing U+0101 followed by
600,000,000 ASCII characters, the UTF-8 reader with a 512 MiB limit runs out of
memory in a separate 1 GiB heap JVM; the revised 256 MiB default successfully
drains and skips that input. Buffer growth, UTF-16 storage and JSON parsing
still require heap beyond the configured content limit.
A separate `-Xmx3g` JVM also reproduces a JVM array-size failure with
`600m` and 620,000,000 ASCII characters followed by U+0100, despite the line
fitting within that configured limit. StringBuilder capacity growth followed by
UTF-16 inflation fails independently of available heap. Clamping the effective
content limit to `512m` successfully skips that input; it does not guarantee
adequate heap for every workload.
Yes. Events at or below the effective UTF-8 content limit are replayed,
LF/CRLF endings do not consume the limit, and parsing diagnostics identify
physical lines after skipped content. The default is reduced from `512m` to
`256m` to keep its maximum buffered character count close to the previous
effective bound.
Non-positive values and values above `512m` now resolve to an effective
maximum of 536,870,912 bytes (512 MiB), including direct constructor arguments.
This clamps the content limit itself, so oversized lines are skipped instead of
replayed as truncated prefixes. Smaller positive limits are unchanged.
```
build/sbt 'core/testOnly *ReplayListenerSuite'
build/sbt 'core/Compile/scalastyle' 'core/Test/scalastyle'
```
All 19 suite tests and both scalastyle checks pass. Coverage includes the 8
KiB ASCII configuration case, raw 1- through 4-byte UTF-8 input with length and
read-buffer boundary splits, line endings, directly observed buffer bounds,
configuration normalization, event delivery through normalized constructors,
reflective no-argument construction, buffer release, skipped-line locations,
bounded malformed-line diagnostics, physical-line diagnostics without an
attached throwable, JSON parsi [...]
Targeted mutation checks confirmed that removing or doubling the buffer
cap, miscounting surrogate pairs, or removing JSON line diagnostics each makes
its corresponding test fail. The oversized lazy-stream scenarios were also
exercised in separate `-Xmx1g` and `-Xmx3g` JVMs as described above.
Generated-by: OpenAI Codex (GPT-6)
Closes #59074 from viirya/codex/replay-line-limits.
Authored-by: Liang-Chi Hsieh <[email protected]>
Signed-off-by: Liang-Chi Hsieh <[email protected]>
(cherry picked from commit 719947a5652f83a80492db50a6f1e5fe14d9e6b8)
Signed-off-by: Liang-Chi Hsieh <[email protected]>
---
.../org/apache/spark/internal/config/History.scala | 17 +-
.../apache/spark/scheduler/ReplayListenerBus.scala | 143 ++++++++++----
.../spark/scheduler/ReplayListenerSuite.scala | 219 +++++++++++++++++++++
docs/monitoring.md | 14 +-
4 files changed, 341 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 b24889065407..b240c924aa7e 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
@@ -138,17 +138,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("4.0.5")
.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 75cb82a6f131..7c40502b5f4f 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
@@ -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 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.
@@ -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)
+ }
+ }
+ }
+
}
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..882d3e711a02 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 = _ != end))
+ assert(bus.replay(Iterator(end, end, truncated), "iterator",
maybeTruncated = true,
+ eventsFilter = _ != 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 d32cc99f91b3..ee3da0c68813 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]
