dongjoon-hyun commented on code in PR #59327:
URL: https://github.com/apache/spark/pull/59327#discussion_r4235589030
##########
core/src/test/scala/org/apache/spark/deploy/history/FsHistoryProviderSuite.scala:
##########
@@ -1248,6 +1248,124 @@ 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 keeps a line that starts exactly at the skip
offset") {
+ val log = newLogFile("end-event-boundary", None, inProgress = false)
+ val lines = endEventTestLines("end-event-boundary")
+ val bytes = writeRawLog(log, lines)
+ // The application end event is the last line: skip exactly to its first
byte.
+ assert(bytes.length > lineBytes(lines.last))
+ val conf = createTestConf()
+ .set(END_EVENT_REPARSE_CHUNK_SIZE, lineBytes(lines.last).toLong)
+ val appender = new LogAppender("end event reparse")
+ withLogAppender(appender) {
+ updateAndCheck(new FsHistoryProvider(conf)) { list =>
+ assert(list.size === 1)
+ assert(list(0).attempts.head.completed)
+ }
+ }
+ // The end event is found by the reparse, not by the fallback that parses
the whole log.
+ val messages = appender.loggingEvents.map(_.getMessage.getFormattedMessage)
+ assert(!messages.exists(_.contains("since end event was not found")),
messages)
Review Comment:
**[2] Minor: this relies on the absence of a log message.**
The list assertions above pass either way, because the whole-log fallback
also lists the app as completed. So this negative match on the INFO text `since
end event was not found` is the only thing that detects a regression. If that
message is reworded or its level changes, the assertion passes vacuously. A
positive signal that does not depend on unrelated wording would make the test
more robust.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -215,20 +234,30 @@ private[spark] class ReplayListenerBus(
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)}")
+ log"at line ${MDC(LINE_NUM, lineNumber)}" + lineOrigin(startOffset))
throw jpe
case ioe: IOException =>
throw ioe
case e: Exception =>
logError(log"Exception parsing Spark event log: ${MDC(PATH,
sourceName)}", e)
val prefix = Option(currentLine).map(_.take(1024)).orNull
val length = Option(currentLine).fold(0)(_.length)
- logError(log"Malformed line #${MDC(LINE_NUM, lineNumber)}: ${MDC(LINE,
prefix)} " +
+ logError(log"Malformed line #${MDC(LINE_NUM, lineNumber)}" +
lineOrigin(startOffset) +
+ log": ${MDC(LINE, prefix)} " +
log"(line length: ${MDC(SIZE, length)} characters; showing at most
1024)\n")
false
}
}
+ /** Describes where line numbers start when the replay does not start at the
source start. */
+ private def lineOrigin(startOffset: Long): MessageWithContext = {
Review Comment:
**[3] Minor: two of the four `lineOrigin` call sites are untested.**
The over-long line warning and the `JsonProcessingException` error are
covered in `FsHistoryProviderSuite`. The `Malformed line #N` error and the
truncated-file `JsonParseException` warning are not, and `ReplayListenerSuite`
has no direct test for `replayFromOffset`. A small `ReplayListenerSuite` case
that calls `replayFromOffset` with a positive offset on malformed and truncated
input would cover both.
##########
core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala:
##########
@@ -1772,6 +1763,55 @@ private[history] class FsHistoryProvider(conf:
SparkConf, clock: Clock)
private[spark] object FsHistoryProvider {
+ /**
+ * Positions `in` at the start of the first line that begins at or after
byte `target`
+ * (`target` > 0), returning a stream that starts there and the byte offset
of that line.
+ * If the stream ends first, the returned stream is at EOF and the offset is
the stream length.
+ *
+ * Since the skip may land anywhere, including inside a multi-byte UTF-8
character, the line
+ * boundary is found on raw bytes before any decoding: '\n' never occurs
inside a multi-byte
+ * UTF-8 sequence. Bytes read past the boundary are put back in front of the
stream.
+ */
+ private[history] def skipToLineStart(in: InputStream, target: Long):
(InputStream, Long) = {
Review Comment:
**[1] Minor: the `target > 0` precondition is only documented.**
With `target <= 0`, the skip loop does nothing, but the scan still discards
everything up to the first `\n`. The helper would then silently drop line 0 and
return the length of that line as the offset. The only caller guards this
today, but returning `(in, 0L)` for `target <= 0` (or adding `require(target >
0)`) would make the contract explicit. The first option would also let the
caller drop its `else (in, 0L)` branch.
##########
core/src/test/scala/org/apache/spark/deploy/history/FsHistoryProviderSuite.scala:
##########
@@ -1248,6 +1248,124 @@ 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 keeps a line that starts exactly at the skip
offset") {
+ val log = newLogFile("end-event-boundary", None, inProgress = false)
+ val lines = endEventTestLines("end-event-boundary")
+ val bytes = writeRawLog(log, lines)
+ // The application end event is the last line: skip exactly to its first
byte.
+ assert(bytes.length > lineBytes(lines.last))
+ val conf = createTestConf()
+ .set(END_EVENT_REPARSE_CHUNK_SIZE, lineBytes(lines.last).toLong)
+ val appender = new LogAppender("end event reparse")
+ withLogAppender(appender) {
+ updateAndCheck(new FsHistoryProvider(conf)) { list =>
+ assert(list.size === 1)
+ assert(list(0).attempts.head.completed)
+ }
+ }
+ // The end event is found by the reparse, not by the fallback that parses
the whole log.
+ val messages = appender.loggingEvents.map(_.getMessage.getFormattedMessage)
+ assert(!messages.exists(_.contains("since end event was not found")),
messages)
+ }
+
+ Seq(None, Some(CompressionCodec.LZ4)).foreach { codecName =>
+ test("end event reparse applies the line length limit and reports relative
line numbers " +
+ s"(codec = ${codecName.getOrElse("none")})") {
+ val log = newLogFile("end-event-limit", None, inProgress = false,
codecName)
+ val tail = Seq(jobStartLine(2, "x" * 2048), jobStartLine(3, "y"))
+ val lines = endEventTestLines("end-event-limit", tail: _*)
+ val codec = codecName.map(CompressionCodec.createCodec(new SparkConf(),
_))
+ writeRawLog(log, lines, codec)
+ // The skip runs on the uncompressed stream while the chunk size is
subtracted from the
+ // file length. Skip into the middle of the first line: replay then
starts at the second
+ // line, and the over-long line is at relative line
`lines.indexOf(tail.head)`.
+ val target = 5L
+ val conf = createTestConf()
+ .set(END_EVENT_REPARSE_CHUNK_SIZE, log.length() - target)
+ .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 offset = lineBytes(lines.head)
+ val messages =
appender.loggingEvents.map(_.getMessage.getFormattedMessage)
+ assert(messages.exists(m => m.contains("Skipped event log lines longer
than 1024 bytes") &&
+ m.contains(s"${log.getName}; first skipped line:
${lines.indexOf(tail.head)} " +
+ s"(lines counted from uncompressed byte offset $offset)")), messages)
+ }
+ }
+
+ test("end event reparse reports 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(lineBytes).sum
+ val conf = createTestConf().set(END_EVENT_REPARSE_CHUNK_SIZE, bytes.length
- offset + 10L)
+ val appender = new LogAppender("end event reparse")
+ withLogAppender(appender) {
+ // A completed log fails on a malformed line that is not the last one,
so the application
+ // is not listed.
+ updateAndCheck(new FsHistoryProvider(conf)) { list =>
+ assert(list.isEmpty)
+ }
+ }
+ val messages = appender.loggingEvents.map(_.getMessage.getFormattedMessage)
+ assert(messages.exists(_.contains(s"${log.getName} at line 2 " +
+ s"(lines counted from uncompressed byte offset $offset)")), messages)
+ assert(!messages.exists(_.contains(s"${log.getName} (lines")), messages)
+ }
+
+ test("skipToLineStart aligns to the first line starting at or after the
target") {
+ def alignStream(in: InputStream, target: Long): (String, Long) = {
+ val (rest, offset) = FsHistoryProvider.skipToLineStart(in, target)
+ (new String(rest.readAllBytes(), StandardCharsets.UTF_8), offset)
+ }
+ def align(data: String, target: Long): (String, Long) = {
+ alignStream(new
ByteArrayInputStream(data.getBytes(StandardCharsets.UTF_8)), target)
+ }
+ // Inside a line, at its '\n', and exactly at the start of the next line.
+ assert(align("abc\ndef\n", 1) === (("def\n", 4L)))
+ assert(align("abc\ndef\n", 3) === (("def\n", 4L)))
+ assert(align("abc\ndef\n", 4) === (("def\n", 4L)))
+ assert(align("abc\ndef\n", 5) === (("", 8L)))
+ // CRLF line endings, and a target at the '\r'.
+ assert(align("ab\r\ncd", 2) === (("cd", 4L)))
+ assert(align("ab\r\ncd", 4) === (("cd", 4L)))
+ // No line starts at or after the target, or the target is beyond the end
of the stream.
+ assert(align("abc", 1) === (("", 3L)))
+ assert(align("abc\n", 10) === (("", 4L)))
+ // A partial line longer than the read buffer.
+ val long = "x" * 20000 + "\nnext"
+ assert(align(long, 1) === (("next", 20001L)))
+ // A stream whose skip() never advances, e.g. at the end of a decompressed
stream.
+ def noSkip(data: String): InputStream = {
+ new FilterInputStream(new
ByteArrayInputStream(data.getBytes(StandardCharsets.UTF_8))) {
+ override def skip(n: Long): Long = 0L
+ }
+ }
+ assert(alignStream(noSkip("abc\ndef"), 2) === (("def", 4L)))
+ assert(alignStream(noSkip("abc\n"), 100) === (("", 4L)))
Review Comment:
**[4] Minor: the EOF-before-target path is covered only by a fake stream.**
This `FilterInputStream` whose `skip` always returns 0 checks the loop logic
well. The real trigger, though, is a codec stream (lz4/zstd) whose decompressed
length is shorter than `target`. An integration case with a tiny lz4 log and
`END_EVENT_REPARSE_CHUNK_SIZE = 1` (so that the compressed length exceeds the
decompressed one) would pin down the end-to-end behavior that motivated the fix.
--
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]
