HyukjinKwon commented on code in PR #59327:
URL: https://github.com/apache/spark/pull/59327#discussion_r4241974345
##########
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:
**Nit (P3):** `skipToLineStart` returns `new SequenceInputStream(rest, in)`,
and in production the part that comes from `in` carries nearly the whole
reparsed tail: about `endEventReparseChunkSize` for uncompressed logs and most
of the file for compressed ones. None of the inputs here exercises that part,
because each one ends inside the 8 KiB read that contains the `\n`. In this
case, for example, the third read returns `\n` + `"next"` and `in` is already
at EOF. The provider-level logs are only a few KB, so they also arrive in one
read. Replacing the put-back with `rest` alone, or with `new
SequenceInputStream(in, rest)`, still passes every new test, and in the
provider tests without a malformed line the whole-log fallback would also mask
a truncated tail.
A case with more than one buffer of varied content after the boundary would
catch both:
```scala
val longTail = (1 to 3000).mkString(",")
assert(align("abc\n" + longTail, 1) === ((longTail, 4L)))
```
**Verification:**
- **Behavior:** When the content after the first line boundary is longer
than the read buffer, skipToLineStart returns that complete content in order,
together with the byte offset of the line start.
- **Regression:** The new case fails when the put-back is reduced to the
buffered remainder alone or when the two streams are concatenated in the
opposite order, and it 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]
