Skip to content

[SPARK-60109][UI] Apply the replay line-length limit and UTF-8-safe line alignment to the History Server end-event reparse #59327

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
viirya wants to merge 6 commits into
base: master
Choose a base branch
from
+314 −26
Open

[SPARK-60109][UI] Apply the replay line-length limit and UTF-8-safe line alignment to the History Server end-event reparse #59327

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
bfc59f2
[SPARK-60109][UI] Apply the replay line-length limit and UTF-8-safe l…
viirya Oct 9, 2026
3b54157
Address review comments: EOF-safe skip, keep a line at the target, bu…
viirya Oct 10, 2026
e0f71e5
Address review comments: non-positive target, positive boundary check…
viirya Oct 10, 2026
cf13cc0
Address review comments: clarify skipToLineStart contract, start the …
viirya Oct 11, 2026
e17a1c7
Address review comment: cover the stream after the put-back bytes in …
viirya Oct 11, 2026
803eb11
Merge remote-tracking branch 'upstream/master' into SPARK-60109
viirya Oct 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
73 changes: 58 additions & 15 deletions core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,14 +17,13 @@

package org.apache.spark.deploy.history

import java.io.{File, FileNotFoundException, IOException}
import java.io.{ByteArrayInputStream, File, FileNotFoundException, InputStream, IOException, SequenceInputStream}
import java.lang.{Long => JLong}
import java.util.{Date, NoSuchElementException, ServiceLoader}
import java.util.concurrent.{ConcurrentHashMap, ExecutorService, TimeUnit}
import java.util.zip.ZipOutputStream

import scala.collection.mutable
import scala.io.{Codec, Source}
import scala.jdk.CollectionConverters._
import scala.util.control.NonFatal
import scala.xml.Node
Expand Down Expand Up @@ -1110,21 +1109,11 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
if (target > 0) {
logInfo(log"Looking for end event; skipping ${MDC(NUM_BYTES, target)} bytes" +
log" from ${MDC(PATH, logPath)}...")
var skipped = 0L
while (skipped < target) {
skipped += in.skip(target - skipped)
}
}

val source = Source.fromInputStream(in)(Codec.UTF8).getLines()

// Because skipping may leave the stream in the middle of a line, read the next line
// before replaying.
if (target > 0) {
source.next()
}
val (tail, startOffset) = skipToLineStart(in, target)

bus.replay(source, lastFile.getPath.toString, !appCompleted, eventsFilter)
bus.replayFromOffset(tail, lastFile.getPath.toString, startOffset, !appCompleted,
eventsFilter)
}
}

Expand Down Expand Up @@ -1772,6 +1761,60 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)

private[spark] object FsHistoryProvider {

/**
* Positions `in` at the start of the first line that begins at or after byte `target`,

@HyukjinKwon HyukjinKwon Oct 11, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit (P3): in itself isn't left at the line start. When a \n is found, in has already been read up to 8 KiB past the boundary (in.read(buffer)), and only the returned SequenceInputStream(rest, in) starts at the line. "Bytes read past the boundary are put back in front of the stream" is also ambiguous, because they are prepended only in the returned stream, not pushed back into in.

The only caller uses the returned stream, but a future caller who trusts this sentence and keeps reading in (for example the tryWithResource handle) would silently lose up to 8 KiB of events. Something like this would be clearer: "Returns a stream that starts at the first line beginning at or after byte target of in, together with that line's byte offset. in itself is consumed and must not be read afterwards; bytes read past the boundary are prepended to the returned stream."

Verification:

  • Inspection: The scaladoc no longer claims that in itself is positioned at the line start, and its description of the returned stream, the consumption of in and the non-positive-target case matches the implementation.

@viirya viirya Oct 11, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Updated in cf13cc0 along the lines you suggested. The scaladoc now says that the returned stream starts at the first line beginning at or after target, and that in is consumed and must not be read afterwards, since the bytes read past the boundary are prepended to the returned stream rather than pushed back into in.

* returning a stream that starts there and the byte offset of that line. If the stream ends
* first, the returned stream is at EOF and the offset is the stream length. A non-positive
* `target` returns `in` unchanged with offset 0.
*
* Since the skip may land anywhere, including inside a multi-byte UTF-8 character, the line
* boundary is found on raw bytes before any decoding: '\n' never occurs inside a multi-byte
* UTF-8 sequence. Bytes read past the boundary are put back in front of the stream.
*/
private[history] def skipToLineStart(in: InputStream, target: Long): (InputStream, Long) = {

@dongjoon-hyun dongjoon-hyun Oct 10, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[1] Minor: the target > 0 precondition is only documented.

With target <= 0, the skip loop does nothing, but the scan still discards everything up to the first \n. The helper would then silently drop line 0 and return the length of that line as the offset. The only caller guards this today, but returning (in, 0L) for target <= 0 (or adding require(target > 0)) would make the contract explicit. The first option would also let the caller drop its else (in, 0L) branch.

@viirya viirya Oct 10, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Changed in e0f71e5. skipToLineStart now returns (in, 0L) for a non-positive target, and the caller calls it unconditionally (the else branch is gone). The skipToLineStart test covers 0 and -1.

if (target <= 0) {
(in, 0L)
} else {
// Stop one byte early, so that a line starting exactly at `target` is kept.
var position = 0L
var eof = false
while (!eof && position < target - 1) {
val skipped = in.skip(target - 1 - position)
if (skipped > 0) {
position += skipped
} else if (in.read() == -1) {
// skip() may return 0 at EOF, e.g. when `target` comes from a compressed length that
// exceeds the decompressed one.
eof = true
} else {
position += 1
}
}

var result: InputStream = in
val buffer = new Array[Byte](8192)
var found = false
while (!eof && !found) {
val n = in.read(buffer)
if (n == -1) {
eof = true
} else {
var i = 0
while (i < n && buffer(i) != '\n') {
i += 1
}
position += math.min(i + 1, n)
if (i < n) {
found = true
val rest = new ByteArrayInputStream(buffer, i + 1, n - i - 1)
result = new SequenceInputStream(rest, in)
}
}
}
(result, position)
}
}

private val APPL_START_EVENT_PREFIX = "{\"Event\":\"SparkListenerApplicationStart\""

private val APPL_END_EVENT_PREFIX = "{\"Event\":\"SparkListenerApplicationEnd\""
Expand Down
51 changes: 40 additions & 11 deletions core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ import com.fasterxml.jackson.core.{JsonParseException, JsonProcessingException}
import com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException

import org.apache.spark.SparkConf
import org.apache.spark.internal.Logging
import org.apache.spark.internal.{Logging, MessageWithContext}
import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config.History
import org.apache.spark.scheduler.ReplayListenerBus._
Expand Down Expand Up @@ -72,8 +72,23 @@ private[spark] class ReplayListenerBus(
sourceName: String,
maybeTruncated: Boolean = false,
eventsFilter: ReplayEventsFilter = SELECT_ALL_FILTER): Boolean = {
val lines = boundedLines(logData, sourceName)
replayEntries(lines, sourceName, maybeTruncated, eventsFilter)
replayFromOffset(logData, sourceName, 0L, maybeTruncated, eventsFilter)
}

/**
* Variant of [[replay()]] for a stream positioned at the start of a line that begins at
* `startOffset` bytes into the (uncompressed) source. A positive `startOffset` is added to
* the diagnostics so that the reported line numbers, which count from the start of the
* stream, can be located in the source.
*/
private[spark] def replayFromOffset(
logData: InputStream,
sourceName: String,
startOffset: Long,
maybeTruncated: Boolean,
eventsFilter: ReplayEventsFilter): Boolean = {
val lines = boundedLines(logData, sourceName, startOffset)
replayEntries(lines, sourceName, startOffset, maybeTruncated, eventsFilter)
}

/**
Expand All @@ -84,7 +99,8 @@ private[spark] class ReplayListenerBus(
*/
private def boundedLines(
logData: InputStream,
sourceName: String): Iterator[(String, Int)] = {
sourceName: String,
startOffset: Long): Iterator[(String, Int)] = {
// Fail on malformed input like Source.getLines() does instead of replacing it.
val decoder = StandardCharsets.UTF_8.newDecoder()
.onMalformedInput(CodingErrorAction.REPORT)
Expand Down Expand Up @@ -131,10 +147,12 @@ private[spark] class ReplayListenerBus(
if (!warned) {
logWarning(log"Skipped event log lines longer than " +
log"${MDC(MAX_SIZE, effectiveMaxLineLength)} bytes in " +
log"${MDC(FILE_NAME, sourceName)}; first skipped line: ${MDC(LINE_NUM, index + 1)}")
log"${MDC(FILE_NAME, sourceName)}; first skipped line: " +
log"${MDC(LINE_NUM, index + 1)}" + lineOrigin(startOffset))
warned = true
}
logDebug(s"Skipped event log line ${index + 1} in $sourceName")
logDebug(s"Skipped event log line ${index + 1} in $sourceName" +
lineOrigin(startOffset).message)
fetchLine()
} else {
(maybeLine.get, index)
Expand All @@ -153,12 +171,13 @@ private[spark] class ReplayListenerBus(
sourceName: String,
maybeTruncated: Boolean,
eventsFilter: ReplayEventsFilter): Boolean = {
replayEntries(lines.zipWithIndex, sourceName, maybeTruncated, eventsFilter)
replayEntries(lines.zipWithIndex, sourceName, 0L, maybeTruncated, eventsFilter)
}

private def replayEntries(
lines: Iterator[(String, Int)],
sourceName: String,
startOffset: Long,
maybeTruncated: Boolean,
eventsFilter: ReplayEventsFilter): Boolean = {
var currentLine: String = null
Expand Down Expand Up @@ -202,8 +221,8 @@ private[spark] class ReplayListenerBus(
throw jpe
} else {
logWarning(log"Got JsonParseException from log file ${MDC(FILE_NAME, sourceName)}" +
log" at line ${MDC(LINE_NUM, lineNumber)}, " +
log"the file might not have finished writing cleanly.")
log" at line ${MDC(LINE_NUM, lineNumber)}" + lineOrigin(startOffset) +
log", the file might not have finished writing cleanly.")
}
}
}
Expand All @@ -215,20 +234,30 @@ private[spark] class ReplayListenerBus(
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)}")
log"at line ${MDC(LINE_NUM, lineNumber)}" + lineOrigin(startOffset))
throw jpe
case ioe: IOException =>
throw ioe
case e: Exception =>
logError(log"Exception parsing Spark event log: ${MDC(PATH, sourceName)}", e)
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)} " +
logError(log"Malformed line #${MDC(LINE_NUM, lineNumber)}" + lineOrigin(startOffset) +
log": ${MDC(LINE, prefix)} " +
log"(line length: ${MDC(SIZE, length)} characters; showing at most 1024)\n")
false
}
}

/** Describes where line numbers start when the replay does not start at the source start. */
private def lineOrigin(startOffset: Long): MessageWithContext = {

@dongjoon-hyun dongjoon-hyun Oct 10, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[3] Minor: two of the four lineOrigin call sites are untested.

The over-long line warning and the JsonProcessingException error are covered in FsHistoryProviderSuite. The Malformed line #N error and the truncated-file JsonParseException warning are not, and ReplayListenerSuite has no direct test for replayFromOffset. A small ReplayListenerSuite case that calls replayFromOffset with a positive offset on malformed and truncated input would cover both.

@viirya viirya Oct 10, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added Replay from an offset reports where line numbers start to ReplayListenerSuite in e0f71e5. It calls replayFromOffset with offset 42 and checks the label on the over-long line warning, Malformed line #N and the truncated-file JsonParseException warning, and that offset 0 leaves the diagnostics unlabeled. The JsonProcessingException error is covered by the FsHistoryProviderSuite test.

if (startOffset > 0) {
log" (lines counted from uncompressed byte offset ${MDC(OFFSET, startOffset)})"
} else {
log""
}
}

override protected def isIgnorableException(e: Throwable): Boolean = {
e.isInstanceOf[HaltReplayException]
}
Expand Down
Loading