viirya commented on code in PR #59327:
URL: https://github.com/apache/spark/pull/59327#discussion_r4235565947
##########
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:
Fixed in 3b54157e183. `skipToLineStart` now reads one byte when `skip`
returns 0: `-1` ends the skip at EOF, otherwise the byte counts as skipped. The
replay then sees an empty input, as the description says. Covered by the
`skipToLineStart` test with a stream whose `skip` always returns 0, both before
EOF and with a target past EOF.
##########
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:
Fixed in 3b54157e183. It now skips to `target - 1` and scans from there, so
a line that starts exactly at `target` is kept. The new test `end event reparse
keeps a line that starts exactly at the skip offset` puts the target on the
first byte of the end event and asserts that the app is listed as completed
without the "end event was not found" whole-log reparse; it fails if the skip
goes to `target` instead.
##########
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:
Changed in 3b54157e183. The source name is now the plain path. The new
`replayFromOffset` takes the start offset separately, and the diagnostics that
print a line number append `(lines counted from uncompressed byte offset N)`
with the offset under its own `OFFSET` key. The public `replay` signatures are
unchanged. The JSON-error test also asserts that the path is no longer followed
by the label.
##########
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:
Agreed, and as you note this also affects the main replay path and was the
same with `Source.getLines()`. I filed SPARK-60114 for it, so this PR keeps the
decoding behavior unchanged.
##########
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:
Changed in 3b54157e183. The partial line is now scanned in 8 KiB reads, and
the bytes after the `\n` are put back in front of the stream with
`SequenceInputStream`, as suggested. The `skipToLineStart` test includes a
partial line longer than the buffer.
##########
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:
Fixed in 3b54157e183; the tests now compute line lengths with
`getBytes(UTF_8).length + 1`.
##########
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:
Added in 3b54157e183:
- The over-long line test now also runs with lz4, using a target derived
from the compressed length. The expected offset is computed from the
uncompressed bytes.
- A direct `skipToLineStart` test covers targets inside a line, at its `\n`,
exactly at a line start and past EOF; CRLF; no trailing newline; a partial line
longer than the read buffer; and a stream whose `skip` never advances.
--
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]
