viirya commented on code in PR #59327:
URL: https://github.com/apache/spark/pull/59327#discussion_r4238767477
##########
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:
Changed in e0f71e55c10. `skipToLineStart` now returns `(in, 0L)` for a
non-positive `target`, and the caller calls it unconditionally (the `else`
branch is gone). The `skipToLineStart` test covers `0` and `-1`.
##########
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:
Changed in e0f71e55c10. The log now has a malformed
`SparkListenerApplicationEnd` line between the environment update and the end
event. The listing pass halts before it and the tail reparse starts after it,
but the whole-log fallback fails on it, so the app is listed as completed only
if the tail reparse finds the end event. The log-message check is removed; the
test fails if the skip goes to `target` instead of `target - 1`.
##########
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:
Added `Replay from an offset reports where line numbers start` to
`ReplayListenerSuite` in e0f71e55c10. It calls `replayFromOffset` with offset
42 and checks the label on the over-long line warning, `Malformed line #N` and
the truncated-file `JsonParseException` warning, and that offset 0 leaves the
diagnostics unlabeled. The `JsonProcessingException` error is covered by the
`FsHistoryProviderSuite` test.
##########
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:
Added `end event reparse stops at the end of a stream shorter than the skip
target` in e0f71e55c10. It writes a one-line lz4 log whose compressed file is
larger than its content (asserted in the test), and uses
`END_EVENT_REPARSE_CHUNK_SIZE = 1`, so `LZ4BlockInputStream.skip` returns 0
before `target - 1`. The scan runs in a separate thread with a one-minute
timeout, so a non-terminating skip loop fails the test instead of hanging it;
it does fail that way when the EOF check is removed.
--
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]
