viirya commented on code in PR #59074:
URL: https://github.com/apache/spark/pull/59074#discussion_r4126667357
##########
core/src/main/scala/org/apache/spark/internal/config/History.scala:
##########
@@ -167,16 +167,19 @@ 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 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. " +
"Introduced in 4.3.0; also available in 3.5.10, 4.0.5, 4.1.4 and
4.2.1; and in " +
Review Comment:
Confirmed that v4.2.1-rc1 contains the old implementation and branch-4.2 is
now 4.2.2-SNAPSHOT. I raised this on the Spark dev mailing list so we can
decide whether to produce an RC2 with the reviewed fix before finalizing 4.2.1.
I am holding off on migration changes pending that discussion.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -35,15 +35,17 @@ import org.apache.spark.util.JsonProtocol
/**
* A SparkListenerBus that can be used to replay events from serialized event
data.
*
- * @param maxLineLength Maximum number of byes (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.
*/
private[spark] class ReplayListenerBus(
Review Comment:
Confirmed with the compiled class: the no-argument constructor was missing.
I restored it while retaining the default argument, and added a reflection
regression test. The compiled class now exposes both constructors and the Scala
default-argument method.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -225,21 +235,56 @@ 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 = 512 * 1024 * 1024
- /** 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 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) {
Review Comment:
The retained prefix is now cleared and trimmed once overflow makes the line
unconditionally skippable. The buffer test checks both zero length and zero
capacity after overflow; this reduces retention during draining, not the
allocation peak.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -225,21 +235,56 @@ 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 = 512 * 1024 * 1024
- /** 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()
Review Comment:
Confirmed the capacity steps from the JDK growth rule. I documented that
lowering 256m to 200m or 150m may leave the same buffer capacity, with 128m as
an example of a smaller step. Changing buffer growth remains part of the reader
follow-up.
##########
core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala:
##########
@@ -109,6 +112,166 @@ 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", "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("Replay constructor normalizes the effective line limit") {
Review Comment:
The constructor test now replays an event and checks delivery for every
out-of-range argument in the loop, including 0, -1 and Int.MinValue.
##########
core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala:
##########
@@ -109,6 +112,166 @@ 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", "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("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)) {
+ assert(new ReplayListenerBus(limit).effectiveMaxLineLength == maximum)
+ }
+ 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("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))
+ }
+ 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("Replay propagates stream IO errors without reporting a stale parse
line") {
Review Comment:
The stream now serves one complete event before throwing. The test verifies
that the event was delivered, the original IOException is rethrown, and no
stale parsing diagnostic is emitted.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -104,35 +110,29 @@ 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 " +
Review Comment:
Added the first skipped physical line number to the warning and a DEBUG
message for every skipped line. A regression test checks one warning and the
locations of both skipped lines.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -202,6 +208,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)}")
+ throw jpe
case ioe: IOException =>
throw ioe
case e: Exception =>
Review Comment:
The malformed-line diagnostic now includes at most the first 1,024
characters and reports the full character count. A regression test uses valid
JSON with a numeric property value to exercise the generic exception path.
--
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]
