viirya commented on code in PR #59327:
URL: https://github.com/apache/spark/pull/59327#discussion_r4243687734
##########
core/src/test/scala/org/apache/spark/deploy/history/FsHistoryProviderSuite.scala:
##########
@@ -1248,6 +1248,149 @@ 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. Target the third byte of a
3-byte character,
+ // so that the line boundary scan, which starts one byte before the
target, starts on a
+ // continuation byte.
+ val leadByte = bytes.indexWhere(b => (b & 0xF0) == 0xE0)
+ assert(leadByte > 0)
+ val conf = createTestConf().set(END_EVENT_REPARSE_CHUNK_SIZE,
(bytes.length - leadByte - 2L))
+ 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)
+ // The whole-log fallback reparse fails on this malformed line, so the
application can only
+ // be listed as completed if the tail reparse finds the end event.
+ val malformed = "{\"Event\":\"SparkListenerApplicationEnd\",\"Timestamp\":"
+ val lines = endEventTestLines("end-event-boundary", malformed)
+ writeRawLog(log, lines)
+ // The application end event is the last line: skip exactly to its first
byte.
+ val conf = createTestConf()
+ .set(END_EVENT_REPARSE_CHUNK_SIZE, lineBytes(lines.last).toLong)
+ updateAndCheck(new FsHistoryProvider(conf)) { list =>
+ assert(list.size === 1)
+ assert(list(0).attempts.head.completed)
+ }
+ }
+
+ test("end event reparse stops at the end of a stream shorter than the skip
target") {
+ val log = newLogFile("end-event-short", None, inProgress = false,
Some(CompressionCodec.LZ4))
+ val codec = CompressionCodec.createCodec(new SparkConf(),
CompressionCodec.LZ4)
+ val bytes = writeRawLog(log, Seq(JsonProtocol.sparkEventToJsonString(
+ SparkListenerApplicationStart("a", Some("b"), 1L, "c", None))),
Some(codec))
+ // The lz4 framing makes this tiny log larger than its content, so with a
1-byte chunk the
+ // skip target is past the end of the decompressed stream: skip() returns
0 there.
+ assert(log.length() - 2 > bytes.length)
+ val provider = new FsHistoryProvider(
+ createTestConf().set(END_EVENT_REPARSE_CHUNK_SIZE, 1L))
+ // Run the scan in a separate thread so that a skip loop that never ends
fails the test.
+ val scan = new Thread(() => provider.checkForLogs())
+ scan.setDaemon(true)
+ scan.start()
+ scan.join(TimeUnit.MINUTES.toMillis(1))
+ assert(!scan.isAlive, "the scan did not finish")
+ assert(provider.getListing().toSeq.map(_.id) === Seq("b"))
+ }
+
+ 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)))
+ // Targets on each byte of a 3-byte character, and on the '\n' after it.
+ val cjk = "a" + 0x4e2d.toChar + "\nb"
+ (1 to 5).foreach { target =>
+ assert(align(cjk, target) === (("b", 5L)), s"target = $target")
+ }
+ // A non-positive target leaves the stream unchanged.
+ assert(align("abc\ndef", 0) === (("abc\ndef", 0L)))
+ assert(align("abc\ndef", -1) === (("abc\ndef", 0L)))
+ // A partial line longer than the read buffer.
+ val long = "x" * 20000 + "\nnext"
+ assert(align(long, 1) === (("next", 20001L)))
Review Comment:
Thanks, added your case in e17a1c798c3: more than 8 KiB of varied content
after the boundary, so the rest of `in` has to follow the put-back bytes in
order. It fails when the put-back is reduced to the buffered remainder or when
the two streams are concatenated in the opposite order, and passes with the
current implementation.
--
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]
