viirya commented on code in PR #59074:
URL: https://github.com/apache/spark/pull/59074#discussion_r4117451556
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -104,35 +107,54 @@ private[spark] class ReplayListenerBus(
line
}
- @tailrec private def fetchLine(): String = {
+ @tailrec private def fetchLine(): (String, Int) = {
val sb = new java.lang.StringBuilder()
var overLong = false
+ var byteLength = 0L
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
+ if (!overLong) {
+ // The strict decoder guarantees valid surrogate pairs. Count
their four UTF-8
+ // bytes on the high surrogate and zero on the low surrogate.
+ byteLength += (if (c < 0x80) {
+ 1
+ } else if (c < 0x800) {
+ 2
+ } else if (Character.isHighSurrogate(c.toChar)) {
+ 4
+ } else if (Character.isLowSurrogate(c.toChar)) {
+ 0
+ } else {
+ 3
+ })
+ // Allow one extra byte until we know whether the line ends in
CRLF.
+ if (byteLength <= maxLineLength.toLong + 1) {
Review Comment:
Confirmed. I reproduced the OOM in a separate `-Xmx1g` JVM using a lazy
stream containing U+0101 followed by 600,000,000 ASCII characters. The 512 MiB
limit fails during builder growth; the revised implementation with a 256 MiB
limit successfully drains and skips the same input.
Changed both defaults to `256m` in ab7c85da8f6 to keep the maximum buffered
character count close to the previous effective bound. The configuration docs
now describe the additional heap cost of buffer growth, UTF-16 storage and JSON
parsing. This does not guarantee that every accepted 256 MiB event fits in a 1
GiB heap.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -104,35 +107,54 @@ private[spark] class ReplayListenerBus(
line
}
- @tailrec private def fetchLine(): String = {
+ @tailrec private def fetchLine(): (String, Int) = {
val sb = new java.lang.StringBuilder()
var overLong = false
+ var byteLength = 0L
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
+ if (!overLong) {
+ // The strict decoder guarantees valid surrogate pairs. Count
their four UTF-8
+ // bytes on the high surrogate and zero on the low surrogate.
+ byteLength += (if (c < 0x80) {
+ 1
+ } else if (c < 0x800) {
+ 2
+ } else if (Character.isHighSurrogate(c.toChar)) {
+ 4
+ } else if (Character.isLowSurrogate(c.toChar)) {
+ 0
+ } else {
+ 3
+ })
+ // Allow one extra byte until we know whether the line ends in
CRLF.
+ if (byteLength <= maxLineLength.toLong + 1) {
+ sb.append(c.toChar)
+ } else {
+ overLong = true
Review Comment:
Addressed in ab7c85da8f6. The reader now uses a `BoundedLineBuffer`, and the
new test checks its retained character count after every append while feeding
content well beyond the limit. It covers zero and small limits and allows only
the one extra character reserved for a possible trailing CR.
I also tested mutations that remove the append cap or double it. Both cause
this test to fail, even though the oversized line would still be rejected at
the end.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -104,35 +107,54 @@ private[spark] class ReplayListenerBus(
line
}
- @tailrec private def fetchLine(): String = {
+ @tailrec private def fetchLine(): (String, Int) = {
val sb = new java.lang.StringBuilder()
var overLong = false
+ var byteLength = 0L
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
+ if (!overLong) {
+ // The strict decoder guarantees valid surrogate pairs. Count
their four UTF-8
+ // bytes on the high surrogate and zero on the low surrogate.
+ byteLength += (if (c < 0x80) {
+ 1
+ } else if (c < 0x800) {
+ 2
+ } else if (Character.isHighSurrogate(c.toChar)) {
+ 4
+ } else if (Character.isLowSurrogate(c.toChar)) {
+ 0
+ } else {
+ 3
+ })
+ // Allow one extra byte until we know whether the line ends in
CRLF.
+ if (byteLength <= maxLineLength.toLong + 1) {
+ sb.append(c.toChar)
+ } else {
+ overLong = true
+ }
}
c = reader.read()
Review Comment:
Agreed that this deserves a separate follow-up. We will address
block-oriented reading and validate it with an SHS replay benchmark there. This
PR keeps the existing read strategy and focuses on the byte-limit and
diagnostic fixes; I have not independently rerun the throughput measurements
quoted here.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -104,35 +107,54 @@ private[spark] class ReplayListenerBus(
line
}
- @tailrec private def fetchLine(): String = {
+ @tailrec private def fetchLine(): (String, Int) = {
val sb = new java.lang.StringBuilder()
var overLong = false
+ var byteLength = 0L
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
+ if (!overLong) {
+ // The strict decoder guarantees valid surrogate pairs. Count
their four UTF-8
+ // bytes on the high surrogate and zero on the low surrogate.
+ byteLength += (if (c < 0x80) {
+ 1
+ } else if (c < 0x800) {
+ 2
+ } else if (Character.isHighSurrogate(c.toChar)) {
+ 4
+ } else if (Character.isLowSurrogate(c.toChar)) {
+ 0
+ } else {
+ 3
+ })
+ // Allow one extra byte until we know whether the line ends in
CRLF.
+ if (byteLength <= maxLineLength.toLong + 1) {
+ sb.append(c.toChar)
+ } else {
+ overLong = true
+ }
}
c = reader.read()
}
- if (overLong) {
+ if (sb.length() > 0 && sb.charAt(sb.length() - 1) == '\r') {
+ sb.setLength(sb.length() - 1)
+ byteLength -= 1
+ }
+ if (overLong || byteLength > maxLineLength) {
Review Comment:
Removed `overLong` in ab7c85da8f6. The bounded accumulator stops updating
once the byte count exceeds the buffer allowance, and the final content-byte
check determines whether to skip the line. This also avoids continuing to count
the rest of an arbitrarily long discarded line.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -104,35 +107,54 @@ private[spark] class ReplayListenerBus(
line
}
- @tailrec private def fetchLine(): String = {
+ @tailrec private def fetchLine(): (String, Int) = {
val sb = new java.lang.StringBuilder()
var overLong = false
+ var byteLength = 0L
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
+ if (!overLong) {
+ // The strict decoder guarantees valid surrogate pairs. Count
their four UTF-8
+ // bytes on the high surrogate and zero on the low surrogate.
+ byteLength += (if (c < 0x80) {
Review Comment:
Applied in ab7c85da8f6: each surrogate half contributes two bytes, with a
comment explaining the strict decoder's valid-pair guarantee. The new raw UTF-8
tests exercise supplementary code points and limits inside those code points. A
mutation counting three bytes per half fails those tests.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -148,15 +170,21 @@ private[spark] class ReplayListenerBus(
sourceName: String,
maybeTruncated: Boolean,
eventsFilter: ReplayEventsFilter): Boolean = {
+ replayEntries(lines.zipWithIndex, sourceName, maybeTruncated, eventsFilter)
Review Comment:
We will handle this in a separate follow-up. The tail-reparse reader and its
offset/partial-line semantics are independent of the two issues addressed here.
This PR leaves that path unchanged; the follow-up will address the bypass of
the bounded reader and clarify line-number semantics after skipping into a file.
##########
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(
Review Comment:
We will address compaction reader and retention consistency in a separate
follow-up, including reproducing the possible loss of live-entity events. That
needs its own retention tests and a decision about handling oversized start
events across both passes. This PR does not change the compaction rewrite path.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -148,15 +170,21 @@ private[spark] class ReplayListenerBus(
sourceName: String,
maybeTruncated: Boolean,
eventsFilter: ReplayEventsFilter): Boolean = {
+ replayEntries(lines.zipWithIndex, sourceName, maybeTruncated, eventsFilter)
+ }
+
+ private def replayEntries(
Review Comment:
Confirmed and fixed in ab7c85da8f6. `JsonProcessingException` now logs the
source and physical line number before the same exception is rethrown. Tests
cover both malformed JSON and a mapping error on physical line 4 after two
skipped lines.
Truncated-last-line handling and ordinary stream IOException propagation are
preserved. A separate test checks that a stream read failure does not receive a
stale JSON parse-line diagnostic. Removing the new JSON diagnostic makes its
regression test fail.
##########
core/src/main/scala/org/apache/spark/internal/config/History.scala:
##########
@@ -167,8 +167,8 @@ 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 " +
+ .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. Setting " +
"this to 0 or a negative value disables the limit. " +
Review Comment:
Corrected the configuration docs and helper comment in ab7c85da8f6. They now
state that non-positive values and values above `Int.MaxValue` use the maximum
supported limit of 2,147,483,647 bytes. The existing configuration
normalization is unchanged. Added tests for zero, negative, exact-maximum and
larger values.
##########
core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala:
##########
@@ -109,6 +109,85 @@ 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 bus = new ReplayListenerBus(8 * 1024)
+ 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 UTF-8 boundaries and line endings") {
+ val end =
JsonProtocol.sparkEventToJsonString(SparkListenerApplicationEnd(1000L))
+ // scalastyle:off nonascii
+ val names = Seq("x", "\u00e9", "\u4e2d", "\ud83d\ude00")
Review Comment:
Confirmed: Jackson escapes U+1F600 in the original test. Replaced that grid
in ab7c85da8f6 with raw UTF-8 input observed through the replay filter before
JSON parsing.
The final code point is 1–4 bytes wide and starts at offsets 8, 8190, 8191
or 8192. Limits range from three bytes below to two bytes above the content
length, covering positions inside multibyte code points and sequences split
across the reader's byte buffer. LF, CRLF and unterminated final lines are
included. A mutant counting supplementary characters as 3+3 bytes fails the new
grid.
--
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]
