dongjoon-hyun commented on code in PR #59327:
URL: https://github.com/apache/spark/pull/59327#discussion_r4235445164
##########
core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala:
##########
@@ -1107,24 +1106,23 @@ private[history] class FsHistoryProvider(conf:
SparkConf, clock: Clock)
Utils.tryWithResource(EventLogFileReader.openEventLog(lastFile.getPath,
fsForPath(lastFile.getPath))) { in =>
val target = lastFile.getLen - reparseChunkSize
- if (target > 0) {
+ val sourceName = 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)
Review Comment:
**[1] This loop never terminates when the stream ends before `target`.**
This is pre-existing, but it is in the block this PR rewrites, and it
contradicts the PR description ("If the skip reaches the end of the stream, the
replay now sees an empty input"). `target` comes from the compressed `getLen`,
while the skip runs on the decompressed stream. If the decompressed stream is
shorter than `target`, `skip` returns 0 at EOF (e.g. `LZ4BlockInputStream.skip`
returns 0 once `finished`, and `BufferedInputStream` / `InputStream.skip`
behave the same way), so `skipped` stops growing and the scan thread spins
forever.
Examples: a small `endEventReparseChunkSize` with a tiny lz4/zstd log whose
framing overhead makes the compressed file larger than its content, or a log
replaced by a shorter file after listing. Could we stop when `skip` returns 0
and confirm EOF with `in.read() == -1`? Then the "empty input" behavior holds
in every case.
##########
core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala:
##########
@@ -1107,24 +1106,23 @@ private[history] class FsHistoryProvider(conf:
SparkConf, clock: Clock)
Utils.tryWithResource(EventLogFileReader.openEventLog(lastFile.getPath,
fsForPath(lastFile.getPath))) { in =>
val target = lastFile.getLen - reparseChunkSize
- if (target > 0) {
+ val sourceName = 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)
}
+ // Because skipping may leave the stream in the middle of a line, or
even in the
+ // middle of a multi-byte UTF-8 character, discard the rest of that
line before any
+ // decoding. Line numbers reported by the replay are then relative
to this offset.
+ val offset = target + skipPartialLine(in)
Review Comment:
**[2] A line starting exactly at `target` is always discarded.**
`skipPartialLine` always throws away bytes up to the next `\n`. When
`target` lands exactly on a line start (byte `target - 1` is `\n`), it discards
a complete line, which can be the `SparkListenerApplicationEnd` line itself.
The app then misses its end event and falls back to the slower path. This is
pre-existing, but now that the code works on raw bytes it is cheap to avoid:
skip `target - 1` bytes, read one byte, and call `skipPartialLine` only if that
byte is not `\n`.
##########
core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala:
##########
@@ -1107,24 +1106,23 @@ private[history] class FsHistoryProvider(conf:
SparkConf, clock: Clock)
Utils.tryWithResource(EventLogFileReader.openEventLog(lastFile.getPath,
fsForPath(lastFile.getPath))) { in =>
val target = lastFile.getLen - reparseChunkSize
- if (target > 0) {
+ val sourceName = 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)
}
+ // Because skipping may leave the stream in the middle of a line, or
even in the
+ // middle of a multi-byte UTF-8 character, discard the rest of that
line before any
+ // decoding. Line numbers reported by the replay are then relative
to this offset.
+ val offset = target + skipPartialLine(in)
+ s"${lastFile.getPath} (lines counted from uncompressed byte offset
$offset)"
Review Comment:
**[4] The offset label overloads `sourceName`.**
`sourceName` is logged under `MDC(PATH, ...)` and `MDC(FILE_NAME, ...)` in
`ReplayListenerBus`. With this change, structured logs on this path get values
like `hdfs://.../app-1 (lines counted from uncompressed byte offset 123456)` in
the path field, which no longer matches the file. Tooling that groups or links
logs by that key sees a different "path" for each offset. Could we pass the
offset to `replay` separately (or log it under its own key) and keep the path
field clean?
##########
core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala:
##########
@@ -1107,24 +1106,23 @@ private[history] class FsHistoryProvider(conf:
SparkConf, clock: Clock)
Utils.tryWithResource(EventLogFileReader.openEventLog(lastFile.getPath,
fsForPath(lastFile.getPath))) { in =>
val target = lastFile.getLen - reparseChunkSize
- if (target > 0) {
+ val sourceName = 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)
}
+ // Because skipping may leave the stream in the middle of a line, or
even in the
+ // middle of a multi-byte UTF-8 character, discard the rest of that
line before any
+ // decoding. Line numbers reported by the replay are then relative
to this offset.
+ val offset = target + skipPartialLine(in)
+ s"${lastFile.getPath} (lines counted from uncompressed byte offset
$offset)"
+ } else {
+ lastFile.getPath.toString
}
- 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()
- }
-
- bus.replay(source, lastFile.getPath.toString, !appCompleted,
eventsFilter)
+ bus.replay(in, sourceName, !appCompleted, eventsFilter)
Review Comment:
**[3] UTF-8 safety is fixed only at the start of the tail.**
`replay(InputStream, ...)` uses the strict decoder, and a
`MalformedInputException` there is an `IOException`, which `replayEntries`
rethrows regardless of `maybeTruncated`. With
`spark.history.fs.inProgressOptimization.enabled=false`, an in-progress log
takes this path while the writer may have flushed only the first bytes of a
multi-byte character at EOF. The listing for that app then fails on this scan
even though `maybeTruncated = true` is meant to tolerate a partial last line.
Should a malformed sequence at EOF be handled like a truncated last line when
`maybeTruncated` is set? A separate JIRA is also fine, since the same applies
to the main replay path.
##########
core/src/test/scala/org/apache/spark/deploy/history/FsHistoryProviderSuite.scala:
##########
@@ -1248,6 +1248,61 @@ abstract class FsHistoryProviderSuite extends
SparkFunSuite with Matchers with P
}
}
+ test("end event reparse when the skip offset lands inside a multi-byte
character") {
+ val log = newLogFile("end-event-utf8", None, inProgress = false)
+ val bytes = writeRawLog(log, endEventTestLines("end-event-utf8",
+ jobStartLine(1, 0x4e2d.toChar.toString * 200)))
+ // The job description is written as raw UTF-8: skip to the second byte of
a character.
+ val leadByte = bytes.indexWhere(b => (b & 0xF0) == 0xE0)
+ assert(leadByte > 0)
+ val conf = createTestConf().set(END_EVENT_REPARSE_CHUNK_SIZE,
(bytes.length - leadByte - 1L))
+ updateAndCheck(new FsHistoryProvider(conf)) { list =>
+ assert(list.size === 1)
+ assert(list(0).attempts.head.completed)
+ }
+ }
+
+ test("end event reparse applies the line length limit and reports relative
line numbers") {
+ val log = newLogFile("end-event-limit", None, inProgress = false)
+ val tail = Seq(jobStartLine(2, "x" * 2048), jobStartLine(3, "y"))
+ val lines = endEventTestLines("end-event-limit", tail: _*)
+ val bytes = writeRawLog(log, lines)
+ // Skip into the middle of the line before the tail. The next line starts
at `offset` and
+ // the over-long line is the first line after it.
+ val offset = lines.takeWhile(_ != tail.head).map(_.length + 1).sum
+ val conf = createTestConf()
+ .set(END_EVENT_REPARSE_CHUNK_SIZE, bytes.length - offset + 10L)
+ .set(EVENT_LOG_MAX_LINE_LENGTH.key, "1k")
+ val appender = new LogAppender("end event reparse")
+ withLogAppender(appender) {
+ updateAndCheck(new FsHistoryProvider(conf)) { list =>
+ assert(list.size === 1)
+ assert(list(0).attempts.head.completed)
+ }
+ }
+ val sourceName = s"${log.getName} (lines counted from uncompressed byte
offset $offset)"
+ val messages = appender.loggingEvents.map(_.getMessage.getFormattedMessage)
+ assert(messages.exists(m => m.contains("Skipped event log lines longer
than 1024 bytes") &&
+ m.contains(s"$sourceName; first skipped line: 1")), messages)
+ }
+
+ test("end event reparse labels JSON errors with the skip offset") {
+ val log = newLogFile("end-event-malformed", None, inProgress = false)
+ val malformed = "{\"Event\":\"SparkListenerApplicationEnd\",\"Timestamp\":"
+ val tail = Seq(jobStartLine(2, "x"), malformed, jobStartLine(3, "y"))
+ val lines = endEventTestLines("end-event-malformed", tail: _*)
+ val bytes = writeRawLog(log, lines)
+ val offset = lines.takeWhile(_ != tail.head).map(_.length + 1).sum
+ val conf = createTestConf().set(END_EVENT_REPARSE_CHUNK_SIZE, bytes.length
- offset + 10L)
+ val appender = new LogAppender("end event reparse")
+ withLogAppender(appender) {
+ new FsHistoryProvider(conf).checkForLogs()
Review Comment:
**[8] Only the log message is asserted here.**
This test checks the log text, but not what happens to the application
listing after the `JsonParseException` propagates. A change that silently
swallowed the malformed end event (listing the app as incomplete, or dropping
it) would still pass. Could we also assert the expected listing state, e.g. via
`provider.getListing()`?
##########
core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala:
##########
@@ -1772,6 +1770,22 @@ private[history] class FsHistoryProvider(conf:
SparkConf, clock: Clock)
private[spark] object FsHistoryProvider {
+ /**
+ * Discards bytes up to and including the next '\n', returning the number of
bytes discarded.
+ * This works on raw bytes so that it is safe to call at an arbitrary byte
offset: '\n' never
+ * occurs inside a multi-byte UTF-8 sequence. The bytes are read one at a
time so that the
+ * stream is left exactly at the start of the next line.
+ */
+ private[history] def skipPartialLine(in: InputStream): Long = {
+ var discarded = 0L
+ var b = in.read()
+ while (b != -1 && b != '\n') {
Review Comment:
**[5] One `read()` call per byte, with no bound.**
`in` is a `BufferedInputStream` (or a codec stream), so each `read()` is a
synchronized method call. If the skip lands at the start of a very long line (a
large SQL plan, or a corrupted line far above `maxLineLength`), this makes one
call per byte, and it repeats on every rescan of that log. A bulk read into a
byte array that scans for `\n` and passes the leftover bytes back in front of
the stream (e.g. `new SequenceInputStream(new ByteArrayInputStream(rest), in)`)
would avoid that while keeping the alignment exact.
##########
core/src/test/scala/org/apache/spark/deploy/history/FsHistoryProviderSuite.scala:
##########
@@ -1248,6 +1248,61 @@ abstract class FsHistoryProviderSuite extends
SparkFunSuite with Matchers with P
}
}
+ test("end event reparse when the skip offset lands inside a multi-byte
character") {
+ val log = newLogFile("end-event-utf8", None, inProgress = false)
+ val bytes = writeRawLog(log, endEventTestLines("end-event-utf8",
+ jobStartLine(1, 0x4e2d.toChar.toString * 200)))
+ // The job description is written as raw UTF-8: skip to the second byte of
a character.
+ val leadByte = bytes.indexWhere(b => (b & 0xF0) == 0xE0)
+ assert(leadByte > 0)
+ val conf = createTestConf().set(END_EVENT_REPARSE_CHUNK_SIZE,
(bytes.length - leadByte - 1L))
+ updateAndCheck(new FsHistoryProvider(conf)) { list =>
+ assert(list.size === 1)
+ assert(list(0).attempts.head.completed)
+ }
+ }
+
+ test("end event reparse applies the line length limit and reports relative
line numbers") {
+ val log = newLogFile("end-event-limit", None, inProgress = false)
+ val tail = Seq(jobStartLine(2, "x" * 2048), jobStartLine(3, "y"))
+ val lines = endEventTestLines("end-event-limit", tail: _*)
+ val bytes = writeRawLog(log, lines)
+ // Skip into the middle of the line before the tail. The next line starts
at `offset` and
+ // the over-long line is the first line after it.
+ val offset = lines.takeWhile(_ != tail.head).map(_.length + 1).sum
Review Comment:
**[6] The expected offset is computed in UTF-16 chars, not UTF-8 bytes.**
`_.length + 1` counts chars, while the production offset counts UTF-8 bytes.
This matches today only because every line before the tail is ASCII. If a line
in `endEventTestLines` ever contains non-ASCII text (for example, reusing the
CJK description from the first test), the expected offset would diverge. Using
`_.getBytes(StandardCharsets.UTF_8).length + 1` here and at L1295 would match
what is being verified.
##########
core/src/test/scala/org/apache/spark/deploy/history/FsHistoryProviderSuite.scala:
##########
@@ -1248,6 +1248,61 @@ abstract class FsHistoryProviderSuite extends
SparkFunSuite with Matchers with P
}
}
+ test("end event reparse when the skip offset lands inside a multi-byte
character") {
Review Comment:
**[7] Coverage gaps.**
All three new tests use uncompressed logs. According to the PR description,
most completed apps take the compressed path (the target comes from the
compressed length and the skip runs on the decompressed stream), but none of
these tests exercise it, so the "uncompressed byte offset" label is not
verified there. The claimed empty-input behavior when the skip reaches EOF is
also untested (see [1]), and so are the `skipPartialLine` edge cases: no
trailing `\n`, `\r\n`, and `target` exactly at a line start. A small direct
test of `skipPartialLine` plus one compressed-codec variant would cover most of
this.
--
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]
