dongjoon-hyun commented on code in PR #59074:
URL: https://github.com/apache/spark/pull/59074#discussion_r4119010915
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -242,4 +248,37 @@ private[spark] object ReplayListenerBus {
// 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
Review Comment:
This retains up to `maxLineLength + 1` chars, but nothing caps that against
the `StringBuilder` implementation limits, so larger configured limits now fail
with `OutOfMemoryError`, regardless of `-Xmx`, where the current code skips the
line:
- **Latin-1 to UTF-16 inflation (limits >= 603,979,775 bytes, i.e. from
576m).** Once more than 603,979,774 Latin-1 chars are retained, the builder's
capacity is 1,207,959,550, and the first char >= U+0100 makes `inflate()` fail
with `UTF16 String size is 1207959550, should be less than 1073741823`. This
also hits lines within the limit: with `600m`, 620,000,000 `x` followed by
U+0100 (620,000,002 bytes) throws, while the current code skips it. The current
code needs 1152m or more for this. It is the [576m, 1152m) case from the first
round, which the new default avoids only for the default.
- **UTF-16 maximum length (limits >= 1,073,741,823 bytes, about 1g).** A
UTF-16 builder cannot exceed 1,073,741,822 chars, so e.g. `€` followed by about
1.18 billion `x` at `1g` throws `Requested array size exceeds VM limit`, while
the current code skips it.
- `<= 0` and values above `Int.MaxValue` fail the same way as the current
code, but the new doc now calls 2147483647 bytes the "maximum supported limit".
On the listing path, the `Error` passes `mergeApplicationListing`'s `case e:
Exception` and is swallowed by the `Future` of `pool.submit` in
`submitLogProcessTask`, so the app is silently missing from the listing. `256m`
and an explicit `512m` are safe.
Could we clamp the effective limit in `maxLineLength(conf)` (and in the
constructor if needed), e.g. to `512m` (any value up to 603,979,773 keeps the
capacity at or below 603,979,774), and document that as the maximum? The clamp
needs to apply to the limit itself, not only to `bufferLimit`, so that lines
within the limit are not silently truncated. I reproduced these with a verbatim
copy of the reader on JDK 21 (up to `-Xmx8g`).
##########
core/src/main/scala/org/apache/spark/internal/config/History.scala:
##########
@@ -167,16 +167,18 @@ private[spark] object History {
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. Values at or below 0, or above 2147483647, use the maximum
supported limit of " +
+ "2147483647 bytes. " +
"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")
+ .createWithDefaultString("256m")
Review Comment:
A reminder about the backports from the first round: this changes the unit
and the default of a key that branch-4.x, branch-4.3, branch-4.2, branch-4.1,
branch-4.0 and branch-3.5 already have (all still `512m` with the `chars * 2`
check, and not in any release yet).
- [v4.3.0-rc1](https://github.com/apache/spark/tree/v4.3.0-rc1) (2026-08-31)
does not contain SPARK-59407, but `branch-4.3` does, so this needs to land on
`branch-4.3` before the next 4.3.0 RC. Otherwise 4.3.0 ships the old semantics
(e.g. `8k` means about 4096 chars and `<= 0` means no limit), and changing it
afterwards becomes a behavior change of a released config.
- The diff cherry-picks cleanly to branch-4.x and branch-4.3, and
branch-4.2/4.1/4.0 conflict only in the scaladoc (`byes`/`bytes`). On
branch-3.5, the new `case jpe: JsonProcessingException` block with
`log"..."`/`MDC` merges without a conflict but does not compile, so it needs a
separate PR like #59058.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -67,22 +66,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)
Review Comment:
Pre-existing and preserved by this PR (`Source.getLines()` with `Codec.UTF8`
also fails on malformed input), but related to the truncation handling: if the
last, unterminated line of an in-progress or crashed app's log ends inside a
multi-byte UTF-8 sequence, this decoder throws `MalformedInputException` from
`reader.read()` at EOF. That happens in `lineEntries.hasNext`, outside the
inner `try`, so it never reaches the `maybeTruncated` tolerance for
`JsonParseException`; it is rethrown as an `IOException`, and the UI rebuild
fails after its retry (permanently for a crashed app). The same log cut at an
ASCII boundary replays fine.
Compressed in-progress logs expose lengths at compression block boundaries
(128 KiB blocks for zstd, 32 KiB for lz4) rather than at code point boundaries,
so with CJK-heavy content this happened in 3/80 (zstd) and 5/80 (lz4) simulated
cuts; uncompressed local files were always cut at code point boundaries in my
runs. A follow-up could treat a `MalformedInputException` raised after the
underlying stream hit EOF as a truncated last line when `maybeTruncated` is set.
##########
core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala:
##########
@@ -109,6 +112,149 @@ class ReplayListenerSuite extends SparkFunSuite with
BeforeAndAfter with LocalSp
assert(eventMonster.loggedEvents(1) ===
JsonProtocol.sparkEventToJsonString(applicationEnd))
}
+ test("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("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("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 == limit + 1)
+ assert(buffer.result().isEmpty)
+ }
+ }
+
+ test("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", "2147483647", "3g")) {
+ conf.set(History.EVENT_LOG_MAX_LINE_LENGTH.key, value)
+ assert(ReplayListenerBus.maxLineLength(conf) == Int.MaxValue)
+ }
+ }
+
+ test("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("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))
+ }
+ assert(appender.loggingEvents.exists(_.getMessage.getFormattedMessage
+ .contains("Exception parsing Spark event log: json-errors at line 4")))
+ }
+ }
+
+ test("Replay propagates stream IO errors without reporting a stale parse
line") {
+ val error = new IOException("read failure")
+ val input = new InputStream {
+ override def read(): Int = throw error
+ }
+ val appender = new LogAppender
+ withLogAppender(appender) {
+ assert(intercept[IOException] {
+ new ReplayListenerBus().replay(input, "read-error")
+ } eq error)
+ }
+ assert(!appender.loggingEvents.exists(_.getMessage.getFormattedMessage
+ .contains("Exception parsing Spark event log")))
+ }
+
+ test("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("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
Review Comment:
About the "existing compressed/end-to-end replay tests" in the PR
description: they don't really check the replayed events (pre-existing, not
from this PR).
- `testApplicationReplay`, used by "End-to-end replay" and "End-to-end
replay with compression", compares each event with itself,
`JsonProtocolSuite.assertEquals(e1, e1)` (L408). SPARK-28770 changed it from
`(e1, e2)`, which looks like a typo. It also `zip`s without a size check and
ignores the result of `replay` (L396), so it passes even if nothing is replayed.
- This test only checks the event count (L300) and also ignores the result
of `replay` (L298).
So nothing checks the content of a multi-buffer LZ4/Snappy/ZSTD log read
through the new reader. Could we assert the result, compare
`originalEvents.size` with `replayedEvents.size` and use `assertEquals(e1, e2)`
(and pick the current app's log instead of the latest file in the shared
directory), here or in a small test-only follow-up? Otherwise, please adjust
the claim in the description.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -202,6 +205,10 @@ 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)}", jpe)
Review Comment:
Thanks for adding the location. Passing `jpe` here duplicates the stack
trace, though, since every caller already logs or retries the same
`IOException`:
- UI loads: `createHybridStore`, `createDiskStore` and `createInMemoryStore`
(FsHistoryProvider L1598, L1672, L1701) retry once on `IOException`, so one
malformed line now adds 2 ERROR stack traces per request (4 with the hybrid
store), where the current code logs none at ERROR (only the INFO "trying
again"). Failed loads are not cached, so this repeats on every refresh.
- Listing (L1032) and compaction (L1224) log the same exception again at
ERROR.
Could we drop the throwable and keep only the source and line (optionally
with `jpe.getOriginalMessage`)? The new test only checks the message, so it
would still pass.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -148,15 +145,21 @@ private[spark] class ReplayListenerBus(
sourceName: String,
maybeTruncated: Boolean,
eventsFilter: ReplayEventsFilter): Boolean = {
+ replayEntries(lines.zipWithIndex, sourceName, maybeTruncated, eventsFilter)
Review Comment:
Two notes for the tail-reparse follow-up (FsHistoryProvider L1104-L1127),
which calls this overload:
1. There, `lines` starts after the byte skip and the dropped partial line,
so the new ERROR at L208-L211 prints the full path with a skip-relative number.
For example, a truncated `SparkListenerApplicationEnd` line at physical line
12,004 of a completed app is reported as `at line 3959`. Before this PR, JSON
errors on this path had no bus-level line number. Please cover it in the
follow-up, together with the existing `Malformed line #N` and the truncation
WARN, which are also skip-relative there, or label the source on that path,
e.g. `s"$path (lines counted after skipping $target bytes)"`.
2. Pre-existing: if `target` lands inside a multi-byte UTF-8 sequence (e.g.
CJK SQL text or job descriptions, which Jackson writes as raw UTF-8),
`source.next()` (L1124) throws `MalformedInputException` before `replay` is
called. `mergeApplicationListing` only logs it, and since the placeholder
`LogInfo` was already written with the same file size, `shouldReloadLog` never
retries it, so the completed app stays missing from the listing (and `target`
is deterministic, so a restart fails the same way). Replaying through
`boundedLines` after the skip, as planned, would throw the same way because of
its strict decoder, so the follow-up should drop the bytes up to the first `\n`
before decoding. I reproduced it at the unit level, not yet on a real SHS.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -226,13 +233,12 @@ 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.
+ * Default UTF-8 line-length cap, matching
spark.history.fs.eventLog.maxLineLength.
+ * Keep the maximum buffered character count close to the original 512 MiB /
2 cap.
*/
- val DEFAULT_MAX_LINE_LENGTH: Int = 512 * 1024 * 1024
+ val DEFAULT_MAX_LINE_LENGTH: Int = 256 * 1024 * 1024
Review Comment:
nit: This duplicates the `256m` default of
`History.EVENT_LOG_MAX_LINE_LENGTH`, and this PR had to change both. Deriving
it, e.g. `History.EVENT_LOG_MAX_LINE_LENGTH.defaultValue.get.toInt` (like
`ChunkedByteBuffer` does with `BUFFER_WRITE_CHUNK_SIZE`; the API is the same
down to branch-3.5), keeps them in sync. Also, "the original 512 MiB / 2 cap"
refers to the unreleased code that this PR replaces, so it will be hard to
interpret after the merge. Could the comment describe the current bound instead
(a skipped line retains at most this many chars plus a possible trailing CR
while it is drained) and leave the history to the PR description?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
