viirya commented on code in PR #59074:
URL: https://github.com/apache/spark/pull/59074#discussion_r4119567878
##########
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:
I reproduced the 600m / 620,000,000 ASCII characters + U+0100 case with the
actual reader on JDK 21 and -Xmx3g. The effective content limit is now capped
at 512m in both configuration resolution and the constructor, and the same
input is successfully skipped. Tests cover normalization and the docs
distinguish this JVM limit from heap requirements.
##########
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:
Confirmed: v4.3.0-rc1 does not contain SPARK-59407, while branch-4.3 does.
This fix needs to reach branch-4.3 before the next RC, along with the other
affected branches. Branch-3.5 needs a separate backport adapting the logging
API.
##########
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:
Removed the throwable from the location diagnostic and kept the same
exception rethrow. The tests now assert a single location diagnostic with
getThrown == null for both parsing and mapping errors.
##########
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:
Confirmed the pre-existing assertion gaps. I have narrowed the testing
description; we will fix the replay result, event count/content assertions and
log-file selection in a separate test follow-up.
##########
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:
Confirmed the skip-relative numbering and reproduced source.next() failing
when the offset lands inside a UTF-8 sequence. We will cover both in the
tail-reparse follow-up, including discarding the partial line as raw bytes
before decoding and correcting the diagnostic location. The SHS listing/retry
behavior still needs an integration regression test.
##########
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:
Reproduced: an ASCII-truncated final JSON line is tolerated with
maybeTruncated, while a partial UTF-8 code point at EOF throws
MalformedInputException. We will handle this separately, including
compressed-log regression coverage.
##########
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:
The reader default now comes from
History.EVENT_LOG_MAX_LINE_LENGTH.defaultValue. I also replaced the historical
comment with a description of the shared default.
--
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]
