dongjoon-hyun commented on code in PR #59074:
URL: https://github.com/apache/spark/pull/59074#discussion_r4125737022
##########
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:
Binary compatibility (pre-existing from SPARK-59407, but this PR is a good
place to fix it before the maintenance releases): every released version
(3.5.9, 4.0.4, 4.1.3, 4.2.0) declares `class ReplayListenerBus extends
SparkListenerBus with Logging`, which has a public no-arg constructor. With
`(maxLineLength: Int = DEFAULT_MAX_LINE_LENGTH)`, scalac only emits
`<init>(I)V`; default arguments don't generate a no-arg constructor.
The class is `private[spark]` and excluded from MiMa, but event-log tools
already use it and would break on 3.5.10 / 4.0.5 / 4.1.4 / 4.2.1:
- Sparklens: `Class.forName("org.apache.spark.scheduler.ReplayListenerBus")`
+ `newInstance()`
([EventHistoryReporter.scala#L20-L21](https://github.com/qubole/sparklens/blob/7f2ea4bac91eff4435923881c263e6d42caa0949/src/main/scala/com/qubole/sparklens/app/EventHistoryReporter.scala#L20-L21))
-> `InstantiationException`.
- aws-emr-advisor: `getDeclaredConstructor().newInstance()`
([EmrSparkLogParser.scala#L75-L76](https://github.com/aws-samples/aws-emr-advisor/blob/3caea29121322c7e76b61443694355f80d66bf6c/src/main/scala/com/amazonaws/emr/spark/EmrSparkLogParser.scala#L75-L76))
-> `NoSuchMethodException`, and `class CustomReplayListenerBus extends
ReplayListenerBus`
([CustomReplayListenerBus.scala#L25](https://github.com/aws-samples/aws-emr-advisor/blob/3caea29121322c7e76b61443694355f80d66bf6c/src/main/scala/org/apache/spark/utils/CustomReplayListenerBus.scala#L25))
-> `NoSuchMethodError: 'void <init>()'` when compiled against a released
version.
The reflective cases fail even after recompiling. Could we add an auxiliary
constructor?
```scala
def this() = this(ReplayListenerBus.DEFAULT_MAX_LINE_LENGTH)
```
It compiles on Scala 2.12 and 2.13 with or without the default argument,
`new ReplayListenerBus()` is not ambiguous, and all in-repo call shapes still
compile, including `new ReplayListenerBus(maxLineLength = ...)` and the
anonymous subclass in `StreamingQueryListenerSuite`. If 4.2.1 ships from RC1,
keeping the default argument as well also preserves
`$lessinit$greater$default$1` for code compiled against 4.2.1.
##########
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:
Test coverage: this only reads `effectiveMaxLineLength`, and no test replays
through a bus built with an out-of-range limit. So reverting
`ReplayListenerBus` L114 to the raw parameter (`new
BoundedLineBuffer(maxLineLength)`, as in the base code) passes the whole suite,
although `new ReplayListenerBus(0)` would then skip every non-empty line, and
`-1` / `Int.MinValue` every line. All production callers pass the
already-normalized `maxLineLength(conf)`, so there's no user impact today, but
the hand-ported branch-3.5 backport could regress without a failing test. Could
we replay one event through the `0`, `-1` and `Int.MinValue` buses in the first
loop and assert that it arrives?
##########
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:
Once `byteLength` exceeds `bufferLimit`, the line is certain to be skipped,
but the retained prefix stays reachable until the `\n` is read. With the
default `256m`, that is ~302 MB (Latin-1) or ~604 MB (UTF-16, i.e. any char >=
U+0100 early in the line) of dead heap for the whole drain (~13 s per GB at the
current read rate), overlapping with concurrent UI loads and compaction. In a
`-Xmx1g` model, bus A draining U+0101 + ~868 MB of ASCII made bus B fail with
`OutOfMemoryError` on a 200 MB line, and releasing the buffer on overflow made
it pass.
This is pre-existing from SPARK-59407, but it's a small fix:
```scala
if (byteLength <= bufferLimit) {
buffer.append(c)
} else {
// The line will be skipped; release the retained prefix while
draining.
buffer.setLength(0)
buffer.trimToSize()
}
```
After overflow `byteLength >= maxLineLength + 2`, so `result()` still
returns `None` regardless of a trailing CR. It doesn't lower the peak, only how
long the memory is held. The assertion at `ReplayListenerSuite` L164
(`buffer.length == limit + 1`) would become `== 0`.
##########
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:
FYI, because this `StringBuilder` starts at 16 and grows as `2c + 2`, the
retained buffer is a step function of the limit rather than proportional to it.
For ASCII-heavy lines, every limit from 150,994,942 to 301,989,885 bytes (~144m
to ~288m) ends with the same 301,989,886-char buffer as the `256m` default, and
~288m to 512m all end with 603,979,774 chars.
So if an operator hits an OOM while a long line is drained and lowers the
limit to `200m` or `150m`, nothing changes. With `-Xmx800m`, U+0101 followed by
300,000,000 ASCII chars failed at the same point (length 150,994,942, when the
302 MB UTF-16 array is copied into a 604 MB one) for `256m`, `200m` and `150m`,
while `143m` and `128m` skipped it. Could the docs mention that only limits
below ~144m (e.g. `128m`) reduce the buffer, or could the growth be capped at
`bufferLimit`?
##########
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:
Pre-existing and optional: the new JSON branch above logs only the source
and line, but this generic branch still logs the entire line (`Malformed line
#...: ${MDC(LINE, currentLine)}` at L219), which can be up to 256 MiB by
default (512 MiB max). A large line that is valid JSON but semantically wrong
ends up here, e.g. a `JobStart` whose `Properties` contain a numeric value
fails `extractString`'s `require` with `IllegalArgumentException`. With the
same code in Spark 3.5.0, a 33,554,544-char line produced a single
33,554,606-char ERROR record, and building the message temporarily needs about
twice the line size (the `log"..."` builder doubles when the trailing `\n` is
appended). Since this PR is about bounding replay memory, could we log a
bounded prefix plus the line length instead? The `Malformed line #4: {}`
assertion would still pass.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -67,22 +69,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))
Review Comment:
A thought for the block-read follow-up, not a blocker here: the limit is a
UTF-8 byte count, but the reader decodes to UTF-16 first and re-derives the
byte width per char. Since `0x0A` never appears inside a multi-byte UTF-8
sequence, splitting lines on raw bytes into a buffer capped at `limit + 1`
bytes, and strictly decoding only the lines that are kept, makes the unit exact
by construction. Skipped bytes can still be streamed through the strict decoder
to keep the "malformed UTF-8 in skipped lines" contract. It would also:
- bound a skipped non-Latin-1 line to `limit + 1` bytes instead of a UTF-16
builder (~604 MB plus growth at `256m`),
- let a `MalformedInputException` be attributed to a physical line (today
the decoder's 8 KiB read-ahead can throw before the preceding lines are
delivered),
- provide the raw-byte handling that the tail-reparse (partial first line)
and truncated-UTF-8-at-EOF follow-ups need.
A quick Java port matched this reader on all the grids in this PR plus
random and malformed inputs. The reader alone ran at ~2,450 / ~310 MB/s (ASCII
/ CJK), vs ~77 / ~91 MB/s here and ~1,140 / ~215 MB/s for a bulk `char[]`
reader. It still needs the CR reservation, a cap (around the 1 GiB UTF-16
`String` limit), and a heap-oriented default.
##########
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)
Review Comment:
For the block-read follow-up: a new `BoundedLineBuffer` (and a 16-char
`StringBuilder` regrown by doubling) per line, including skipped lines,
allocates ~4.3 bytes per input byte on the event logs under
`core/src/test/resources/spark-events`, vs ~1.7 with one buffer per iterator
that is reset for each line (Jackson `readTree` alone is ~3.9). Today the
per-char `BufferedReader.read()` lock hides the CPU cost, but with a bulk
reader it becomes ~10% of the read loop. Most of this is inherited from
SPARK-59407 (this PR adds ~40 B per line), so it's fine to handle it there. A
reused buffer would need a capacity cap so that it doesn't pin a huge array for
the rest of the file.
##########
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 " +
Review Comment:
nit: This hard-codes `536870912` twice, duplicating
`ReplayListenerBus.MAX_LINE_LENGTH`, similar to my earlier comment about
`DEFAULT_MAX_LINE_LENGTH`. We could define the cap in `History` above
`EVENT_LOG_MAX_LINE_LENGTH` (like `DEFAULT_LOG_DIR`), interpolate it with
`.doc(s"...")`, and derive `ReplayListenerBus.MAX_LINE_LENGTH` from it. The
other direction (referencing `ReplayListenerBus.MAX_LINE_LENGTH` from this doc)
would create an initialization cycle, because the `ReplayListenerBus` object
already reads `History.EVENT_LOG_MAX_LINE_LENGTH.defaultValue` during its
initialization. `monitoring.md` would still need a manual update.
##########
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:
Following up on my earlier backport comment: `v4.2.1-rc1` (cut 2026-09-28
06:11 UTC, about an hour before the latest commit here) already contains
SPARK-59407 with the old semantics: `.version("4.2.1")`, the `512m` default,
the `sb.length() * 2` check, and "Setting this to 0 or a negative value
disables the limit". `branch-4.2` has already moved to `4.2.2-SNAPSHOT` without
this fix. My earlier comment only covered `branch-4.3`, assuming nothing had
shipped yet.
If 4.2.1 is finalized from RC1 and this PR is backported afterwards, a patch
release (4.2.2) would change the unit, the default and the `<= 0` meaning of a
released config:
- Default users: the same for ASCII, but stricter for non-ASCII-heavy lines
(e.g. 268M -> 134M chars for 2-byte text, 268M -> ~89.5M chars for 3-byte text).
- `8k`: an `ApplicationStart` line with 3,000 x U+4E2D (3,085 chars / 9,085
bytes) is replayed by RC1 but skipped here, so the app is missing from the
listing.
- `0` / negative / `3g`: RC1 never skips (`sb.length() * 2 < Int.MaxValue`
is always true), while this PR clamps to 512 MiB, so there is no way to replay
events above 512 MiB anymore.
- An explicit `512m` (RC1's documented default): RC1 keeps at most
268,435,456 chars and skips U+0101 followed by 600,000,000 ASCII chars with
`-Xmx1g`, while this PR grows the builder to 603,979,774 chars (~1.2 GB as
UTF-16) and fails with `OutOfMemoryError` on UI load or compaction.
Could we get this into `branch-4.2` before 4.2.1 is finalized, e.g. by
raising it on the RC1 vote so that it lands in RC2? Otherwise, we need a
`core-migration-guide.md` entry for 4.2.1 -> 4.2.2 and should qualify "also
available in ... 4.2.1" here and in `monitoring.md`.
##########
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:
This stream throws on the very first `read()`, so no line has been parsed
and `currentLine` / `lineNumber` are still at their initial values; the
stale-line situation in the test name never occurs. For example, a later change
like `case ioe: IOException => if (currentLine != null) logError(log"... at
line ${MDC(LINE_NUM, lineNumber)}")` would still pass, while it would report
`at line 1` for a real log whose first event parses and whose compressed stream
then fails. Could the stream serve one valid event line (e.g. `end + "\n"`)
before throwing, keeping the `eq error` and no-diagnostic assertions?
##########
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
Review Comment:
nit: This hand-rolls what `zipWithIndex` already does at L151. If the reader
emits one `Option` per physical line, both overloads number lines the same way,
and the manual counter, the tuple-typed state and the `@tailrec` recursion go
away:
```scala
var warned = false
Iterator.continually(reader.read()).takeWhile(_ != -1).map { first =>
val buffer = new BoundedLineBuffer(effectiveMaxLineLength)
var c = first
while (c != -1 && c != '\n') {
buffer.append(c.toChar)
c = reader.read()
}
val maybeLine = buffer.result()
if (maybeLine.isEmpty && !warned) {
logWarning(...) // unchanged
warned = true
}
maybeLine
}.zipWithIndex.collect { case (Some(line), index) => (line, index) }
```
A `NextIterator[Option[String]]` also works if relying on `takeWhile`'s
one-element look-ahead feels too implicit. I checked that this behaves the same
as the current iterator on Scala 2.12 and 2.13, including the look-ahead from
`lineEntries.hasNext` in the `JsonParseException` handler and exceptions thrown
mid-line. Feel free to skip it if you'd rather keep the diff against
SPARK-59407 small.
##########
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:
Since the physical `index` is available here now, could the warning include
it (e.g. `index + 1` of the first skipped line)? Today a replay can return
`true` after dropping several lines, and the only trace is one "Skipped event
log lines longer than 268435456 bytes in <path>" per file, without a location
or a count; later skips are not logged at any level. `replayEntries` already
uses one WARN plus a per-occurrence `logDebug` for unrecognized events
(L176-L183), which would fit here too. A count at EOF alone wouldn't be enough,
because listing replays often stop early with `HaltReplayException`.
##########
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") {
Review Comment:
nit: Since this PR fixes a specific issue and adds tests to an existing
suite, could we prefix the new test names with the JIRA ID, e.g.
`test("SPARK-59804: Replay line limit uses UTF-8 bytes")`, as the contributing
guide suggests? The same applies to the other new tests (L128, L157, L169,
L180, L194, L208, L233, L248, L267). It helps to trace failures across the
backport branches.
##########
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
Review Comment:
nit: Could the `@param` mention the normalization, e.g. "Values <= 0 or
above `ReplayListenerBus.MAX_LINE_LENGTH` (512 MiB) use `MAX_LINE_LENGTH`"?
Otherwise a caller passing `0` (RC1's "disables the limit" semantics) or `1 <<
30` silently gets a 512 MiB limit.
--
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]
