[SPARK-60158][UI] Avoid per-character synchronized reads in the bounded event log line reader by viirya · Pull Request #59360 · apache/spark

viirya · 2026-10-11T06:36:45Z

What changes were proposed in this pull request?

The bounded line reader used by ReplayListenerBus.replay(InputStream, ...) (added in SPARK-59407, byte accounting fixed in SPARK-59804) called BufferedReader.read() once per character. This PR makes it read decoded characters in bulk instead:

boundedLines reads from the InputStreamReader into an 8192-char array, instead of wrapping the reader in a BufferedReader and calling read() per character. BufferedReader.read() takes a lock on every call.
BoundedLineBuffer.append(c) is replaced by appendUntilNewline(chars, start, end), which finds the next \n and adds up the UTF-8 length in a single pass, appends the range in one call if it fits, and returns the index of the \n. Once the line exceeds the allowance, it releases the retained prefix and only scans for \n from then on.

The reader's semantics are unchanged: the UTF-8 content byte limit with the trailing-CR allowance, physical line indices including skipped lines, strict decoding, and releasing the buffer of an over-long line while it is drained. The Iterator[String] overload of replay is not affected.

Why are the changes needed?

Since SPARK-59407, line splitting in this reader is several times slower than the previous Source.getLines()/readLine() (measured at about 14x in the review of #59074). Every replay through the InputStream overload pays this: loading application UIs in the History Server, event log compaction, and listing with fast in-progress parsing.

It also matters for #59327 (SPARK-60109), which routes the History Server end-event reparse through this reader. For compressed logs, that reparse reads most of the decompressed file, because the skip target is computed from the compressed length, so without this change the listing of completed applications would become several times slower.

A local measurement of a full replay through replay(InputStream, ...) with the History Server listing filter (210 MB of uncompressed event lines built from spark-events/local-1642039451826 with randomized digits, median of 3 runs, JDK 21):

codec	file size	decompression only	`Source.getLines()` + `replay(Iterator)`	bounded reader before	bounded reader after
lz4	56 MB	0.08 s	0.21 s	1.13 s	0.22 s
zstd	16 MB	0.06 s	0.17-0.18 s	1.10 s	0.20 s

Does this PR introduce any user-facing change?

No behavior change. Replaying event logs through the bounded reader (History Server UI loading, listing and event log compaction) is faster.

How was this patch tested?

New tests in ReplayListenerSuite:

SPARK-60158: Line buffer gives the same result for any split of a line: random lines (ASCII, CR, 2-, 3- and 4-byte characters) split at random points; the returned \n index and the result match a reference that drops one trailing CR and keeps the line if its UTF-8 length fits, and the buffer is released once the line exceeds the limit plus the CR allowance.
SPARK-60158: Replay splits lines across read buffers like a reference splitter: randomized inputs with lines up to 20000 characters, so that lines and multi-byte characters cross the 8192-char read buffer, compared with a reference that splits on \n, drops one trailing CR and keeps lines within the byte limit. Each input is replayed both from a ByteArrayInputStream and from a stream that returns 1 to N bytes per read with available() == 0, like codec streams.
SPARK-60158: Replay reports physical line numbers for lines spanning read buffers: a skipped 20000-char line followed by a malformed line, with full and short reads; the skip warning and Malformed line #N report the physical line numbers.

The existing SPARK-59804 tests (byte limit grid across the read buffer boundary, physical line numbers, skipped line diagnostics, malformed UTF-8 and stream IO errors) pass unchanged. The SPARK-59804 line buffer test now uses appendUntilNewline. ReplayListenerSuite, FsHistoryProviderSuite (RocksDB backend), EventLogFileCompactorSuite, core/scalastyle and core/Test/scalastyle pass.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Claude Opus 5.5)

…ed event log line reader Co-authored-by: Claude Code

dongjoon-hyun

+1, LGTM. Thank you, @viirya. The bulk read keeps the semantics of the bounded reader: the range append matches single-character appends, including when it stops counting and when it releases the buffer, and the EOF, empty-line, trailing-CR and buffer-boundary cases behave as before. Malformed input also fails as before, because the previous BufferedReader read the same 8192-char chunks.

I left 6 minor, non-blocking inline comments:

The new splitter test only uses ByteArrayInputStream, so every read returns a full 8192-char chunk. The short reads of codec streams are not exercised.
append(c: Char) no longer has a production caller and duplicates the limit logic.
Each chunk is scanned twice, once for \n and once for the UTF-8 width.
A line that fits in one chunk still goes through a growing StringBuilder.
Design question: splitting on raw bytes could avoid decoding over-long lines and recounting their width.
No test checks the physical line indices when lines span several read buffers.

dongjoon-hyun · 2026-10-12T11:10:27Z

+        .filter(_.getBytes(StandardCharsets.UTF_8).length <= limit)
+      val seen = new ArrayBuffer[String]
+      assert(new ReplayListenerBus(limit).replay(
+        new ByteArrayInputStream(data.getBytes(StandardCharsets.UTF_8)), "reference",


[1] Minor: the read pattern of codec streams is not exercised.

ByteArrayInputStream.available() is always positive, so StreamDecoder keeps decoding until the 8192-char array is full, and every reader.read(chars) here returns a full chunk. With LZ4/ZStd streams, available() is often 0, so reads come back short and uneven. A chunk can then end right after a \r, right before a \n, or be a single \n. Wrapping the input in a stream that returns a random 1..N bytes per read and reports available() == 0 would cover the fill()/pos/len handling under those reads.

Added in 714c458. The reference splitter test now also runs each input through a stream that returns 1 to N bytes per read (N is 1, 3, 64 or 8192) and reports available() == 0, so chunks end at arbitrary points, including right after a \r and right before or at a \n. A mutation that does not treat a \n at the start of a new chunk as a line end passes all the existing tests but fails this one.

dongjoon-hyun · 2026-10-12T11:10:27Z

@@ -275,19 +294,44 @@ private[spark] object ReplayListenerBus {

    def append(c: Char): Unit = {


[2] Minor: append(c: Char) is now used only by tests.

fetchLine only calls the range overload. That leaves two copies of the overflow and release logic, and only one of them runs in production. It also means the SPARK-59804 test Replay line buffer stops retaining oversized content now checks the path that replay no longer uses. Removing append(c) and moving that test to the range overload, or making append(c) delegate to append(Array(c), 0, 1), would keep a single implementation. The equivalence test would then only be needed if both methods stay.

Done in 714c458, together with [3]. append(c) and the range append are replaced by a single appendUntilNewline(chars, start, end), which is what fetchLine calls, so there is one copy of the limit and release logic. The SPARK-59804 buffer test now uses it, and the equivalence test is replaced by one that splits random lines at random points and compares the result with a reference (drop one trailing CR, keep the line if its UTF-8 length fits), and checks that the buffer is released once the line exceeds the limit plus the CR allowance.

dongjoon-hyun · 2026-10-12T11:10:28Z

+          var endOfLine = false
+          while (!endOfLine && fill()) {
+            var i = pos
+            while (i < len && chars(i) != '\n') {


[3] Minor: each chunk is scanned twice.

This loop looks for \n, and then BoundedLineBuffer.append(chars, start, end) walks the same range again to add up utf8Width. Counting the width in this loop (or passing it to the buffer) would make it a single pass. After the line goes over the limit, only the \n search is needed.

Changed in 714c458: appendUntilNewline finds the \n and adds up the UTF-8 width in the same loop, and once the line goes over the limit it only searches for \n. The full-replay numbers stay the same (lz4 0.22 s, zstd 0.19 s, vs 0.20 s / 0.17 s for Source.getLines()).

dongjoon-hyun · 2026-10-12T11:10:28Z

@@ -116,15 +130,20 @@ private[spark] class ReplayListenerBus(

      @tailrec private def fetchLine(): (String, Int) = {
        val buffer = new BoundedLineBuffer(effectiveMaxLineLength)


[4] Minor: optional fast path for lines that fit in one chunk.

Each line still gets a new StringBuilder (initial capacity 16) that grows by doubling, about 10 copies for a 10 KB line, and is copied once more by toString/substring. Typical event lines fit in one 8192-char chunk, so when the \n is found in the same chunk and the line is within the limit, new String(chars, pos, i - pos) would skip the builder. This is optional, since the PR's numbers already match Source.getLines().

I'd prefer to leave this out. The numbers are already on par with Source.getLines(), and the fast path would need its own handling of the trailing CR and the limit, which would be a second place to keep in sync with BoundedLineBuffer.

dongjoon-hyun · 2026-10-12T11:10:28Z

-    val reader = new BufferedReader(new InputStreamReader(logData, decoder))
+    // Read decoded characters in bulk and scan them for '\n', instead of calling read() per
+    // character on a BufferedReader, which takes a lock on every call.
+    val reader = new InputStreamReader(logData, decoder)


[5] Design question (non-blocking): splitting on bytes.

In UTF-8 the byte 0x0A never appears inside a multi-byte sequence, so lines could be split and limited on raw bytes, and only the accepted lines decoded strictly. That gives the exact byte count without the per-char utf8Width rule, and an over-long line is skipped without being decoded. The trade-off is that malformed UTF-8 inside a skipped line would no longer fail the replay. I'm fine with the current approach. I'm only mentioning it in case the remaining overhead matters later.

Thanks. I'd keep decoding every line for now: SPARK-59804 kept failing the replay on malformed UTF-8 inside a skipped line (Replay still rejects malformed UTF-8 in skipped lines), and splitting on raw bytes would drop that. Raw-byte splitting also comes up for an incomplete UTF-8 sequence at EOF (SPARK-60114), so I'll look at it there if the remaining overhead matters.

dongjoon-hyun · 2026-10-12T11:10:28Z

+    }
+  }
+
+  test("SPARK-60158: Replay splits lines across read buffers like a reference splitter") {


[6] Minor: line indices across read buffers are not covered.

This test sees only line contents through eventsFilter, not the physical index that boundedLines returns. The existing Replay reports the physical location of skipped lines input (about 4 KB) fits in one read buffer. So no test checks that lineIndex stays right when a skipped line spans several fill() calls. An example is a 20000-char skipped line followed by a malformed line, where the Malformed line #N and first skipped line numbers should still be the physical ones.

Added SPARK-60158: Replay reports physical line numbers for lines spanning read buffers in 714c458: a skipped 20000-char line followed by a malformed line, with full reads and with 1-byte and up-to-100-byte reads. It checks first skipped line: 2 and Malformed line #4, and fails if the line index is advanced per chunk instead of per line.

…nd line index tests Co-authored-by: Claude Code

…ed event log line reader ### What changes were proposed in this pull request? The bounded line reader used by `ReplayListenerBus.replay(InputStream, ...)` (added in SPARK-59407, byte accounting fixed in SPARK-59804) called `BufferedReader.read()` once per character. This PR makes it read decoded characters in bulk instead: - `boundedLines` reads from the `InputStreamReader` into an 8192-char array, instead of wrapping the reader in a `BufferedReader` and calling `read()` per character. `BufferedReader.read()` takes a lock on every call. - `BoundedLineBuffer.append(c)` is replaced by `appendUntilNewline(chars, start, end)`, which finds the next `\n` and adds up the UTF-8 length in a single pass, appends the range in one call if it fits, and returns the index of the `\n`. Once the line exceeds the allowance, it releases the retained prefix and only scans for `\n` from then on. The reader's semantics are unchanged: the UTF-8 content byte limit with the trailing-CR allowance, physical line indices including skipped lines, strict decoding, and releasing the buffer of an over-long line while it is drained. The `Iterator[String]` overload of `replay` is not affected. ### Why are the changes needed? Since SPARK-59407, line splitting in this reader is several times slower than the previous `Source.getLines()`/`readLine()` (measured at about 14x in the review of #59074). Every replay through the `InputStream` overload pays this: loading application UIs in the History Server, event log compaction, and listing with fast in-progress parsing. It also matters for #59327 (SPARK-60109), which routes the History Server end-event reparse through this reader. For compressed logs, that reparse reads most of the decompressed file, because the skip target is computed from the compressed length, so without this change the listing of completed applications would become several times slower. A local measurement of a full replay through `replay(InputStream, ...)` with the History Server listing filter (210 MB of uncompressed event lines built from `spark-events/local-1642039451826` with randomized digits, median of 3 runs, JDK 21): | codec | file size | decompression only | `Source.getLines()` + `replay(Iterator)` | bounded reader before | bounded reader after | |---|---|---|---|---|---| | lz4 | 56 MB | 0.08 s | 0.21 s | 1.13 s | 0.22 s | | zstd | 16 MB | 0.06 s | 0.17-0.18 s | 1.10 s | 0.20 s | ### Does this PR introduce _any_ user-facing change? No behavior change. Replaying event logs through the bounded reader (History Server UI loading, listing and event log compaction) is faster. ### How was this patch tested? New tests in `ReplayListenerSuite`: - `SPARK-60158: Line buffer gives the same result for any split of a line`: random lines (ASCII, CR, 2-, 3- and 4-byte characters) split at random points; the returned `\n` index and the result match a reference that drops one trailing CR and keeps the line if its UTF-8 length fits, and the buffer is released once the line exceeds the limit plus the CR allowance. - `SPARK-60158: Replay splits lines across read buffers like a reference splitter`: randomized inputs with lines up to 20000 characters, so that lines and multi-byte characters cross the 8192-char read buffer, compared with a reference that splits on `\n`, drops one trailing CR and keeps lines within the byte limit. Each input is replayed both from a `ByteArrayInputStream` and from a stream that returns 1 to N bytes per read with `available() == 0`, like codec streams. - `SPARK-60158: Replay reports physical line numbers for lines spanning read buffers`: a skipped 20000-char line followed by a malformed line, with full and short reads; the skip warning and `Malformed line #N` report the physical line numbers. The existing SPARK-59804 tests (byte limit grid across the read buffer boundary, physical line numbers, skipped line diagnostics, malformed UTF-8 and stream IO errors) pass unchanged. The SPARK-59804 line buffer test now uses `appendUntilNewline`. `ReplayListenerSuite`, `FsHistoryProviderSuite` (RocksDB backend), `EventLogFileCompactorSuite`, `core/scalastyle` and `core/Test/scalastyle` pass. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5.5) Closes #59360 from viirya/SPARK-60158. Authored-by: Liang-Chi Hsieh <viirya@gmail.com> Signed-off-by: Liang-Chi Hsieh <viirya@gmail.com> (cherry picked from commit a2659a7) Signed-off-by: Liang-Chi Hsieh <viirya@gmail.com>

viirya · 2026-10-12T21:35:40Z

Merge Summary:

merged into master a2659a7
merged into branch-4.x 679157d
merged into branch-4.3 a4c63cb
merged into branch-4.2 5e0c846
merged into branch-4.1 876d98f
merged into branch-4.0 826617e
merged into branch-3.5 aca614f

Posted by merge_spark_pr.py

viirya · 2026-10-12T21:37:09Z

Thanks @dongjoon-hyun !

uros-b

Hi folks @viirya and @dongjoon-hyun, please note that this has broken:

branch-4.0
branch-4.1

viirya · 2026-10-13T16:03:55Z

Hi folks @viirya and @dongjoon-hyun, please note that this has broken:
* branch-4.0

* branch-4.1

To clarify that, the broken branches (4.1/4.0/3.5) are not caused by this PR. It is caused by new PyArrow release. I fixed it in #59423.

…ine alignment to the History Server end-event reparse ### What changes were proposed in this pull request? When building the application listing, `FsHistoryProvider` looks for the application end event by reopening the last event log file, skipping to `len - spark.history.fs.endEventReparseChunkSize` bytes and re-parsing the rest. This PR changes how that tail is read: - The new `FsHistoryProvider.skipToLineStart` positions the stream at the first line that starts at or after the skip target, working on raw bytes before any UTF-8 decoding. `0x0A` never occurs inside a multi-byte UTF-8 sequence, so this is safe at any byte offset. - It skips to `target - 1` and scans from there, so a line that starts exactly at `target` (possibly the end event itself) is kept instead of being discarded. - When `skip` returns 0, it reads one byte to tell EOF apart from a short skip, so it terminates when the decompressed stream is shorter than `target` (e.g. when the compressed length exceeds the decompressed one). - The partial line is scanned in bulk; bytes read past the line boundary are put back in front of the stream with a `SequenceInputStream`. - The tail is then replayed through the bounded `InputStream` reader instead of the `Iterator[String]` overload fed by `Source.fromInputStream(in)(Codec.UTF8).getLines()`, so `spark.history.fs.eventLog.maxLineLength` and the strict decoder apply on this path as well. - The new `ReplayListenerBus.replayFromOffset` takes the byte offset of the stream's first line. When it is positive, every replay diagnostic that prints a line number (JSON errors, `Malformed line #N`, the truncation warning and the over-long line warning) adds `(lines counted from uncompressed byte offset N)`, with the offset logged under its own `OFFSET` key. The source name stays the plain file path, so the `PATH` / `FILE_NAME` fields are unchanged. The existing `replay` signatures are unchanged. `maybeTruncated`, `eventsFilter` and `HaltReplayException` handling are unchanged, since all `replay` variants share `replayEntries`. Without a skip (`target <= 0`), replay starts at offset 0 and the diagnostics are as before. ### Why are the changes needed? This is the tail-reparse follow-up discussed in #59074. With the defaults (completed apps, `endEventReparseChunkSize=1m`), every completed application takes this path, and because the target is computed from the compressed file length while the skip runs on the decompressed stream, most of a compressed log can be read there. 1. The `Iterator[String]` overload bypasses the bounded line reader (SPARK-59407 / SPARK-59804), so `maxLineLength` did not apply and an over-long line was fully materialized. 2. If the skip target landed inside a multi-byte UTF-8 character (e.g. CJK SQL text or job descriptions, which Jackson writes as raw UTF-8), `source.next()` threw `MalformedInputException` before replay started. `mergeApplicationListing` only logs it, and since the placeholder listing entry was already written with the same file size, the log is not retried, so the completed application stayed missing from the listing. The target is deterministic, so restarting the History Server failed the same way. 3. Line numbers in the replay diagnostics were relative to the skip point but printed next to the full file path. This builds on #59360 (SPARK-60158, merged), which makes the bounded reader read in bulk; without it, routing the tail through the bounded reader would make this step several times slower. The two should be backported together. ### Does this PR introduce _any_ user-facing change? Yes, in the History Server: - A completed application whose end-event reparse offset lands inside a multi-byte UTF-8 character now appears in the application listing instead of being dropped. - `spark.history.fs.eventLog.maxLineLength` now also applies to the end-event reparse, so over-long lines there are skipped with a warning instead of being materialized. - Replay log messages on this path name the byte offset that their line numbers are relative to. - A line that starts exactly at the reparse offset is no longer discarded, so an end event there no longer forces a reparse of the whole log. - If the decompressed stream ends before the reparse offset, the scan no longer spins forever in the skip loop; the replay sees an empty input (previously, reaching EOF exactly at the offset failed with `NoSuchElementException` from `source.next()`). ### How was this patch tested? New tests in `FsHistoryProviderSuite`: - `end event reparse when the skip offset lands inside a multi-byte character`: the reparse offset is set to the third byte of a 3-byte character, so the line boundary scan starts on a continuation byte. Before this PR, the application is missing from the listing. - `end event reparse keeps a line that starts exactly at the skip offset`: the offset is the first byte of the end event, and a malformed line before it makes the whole-log fallback reparse fail, so the app is listed as completed only if the tail reparse finds the end event. - `end event reparse stops at the end of a stream shorter than the skip target`: a tiny lz4 log whose compressed file is larger than its content, with a 1-byte chunk; the scan must finish (it runs with a timeout). - `end event reparse applies the line length limit and reports relative line numbers` (uncompressed and lz4): an over-long line in the tail is skipped under a 1k limit, and the warning names the relative line number and the offset in the uncompressed stream. - `end event reparse reports JSON errors with the skip offset`: a malformed end event in the tail is reported with the relative line number and the offset, the path field stays the plain file name, and the application is not listed. - `skipToLineStart aligns to the first line starting at or after the target`: targets inside a line, at its `\n`, exactly at a line start, past EOF and non-positive; on each byte of a 3-byte character; CRLF; no trailing newline; a partial line longer than the read buffer; and a stream whose `skip` never advances. New test in `ReplayListenerSuite`: - `Replay from an offset reports where line numbers start`: `replayFromOffset` with a positive offset labels the over-long line warning, `Malformed line #N` and the truncated-file warning, and offset 0 leaves the diagnostics unchanged. `FsHistoryProviderSuite` (RocksDB backend), `ReplayListenerSuite`, `core/scalastyle` and `core/Test/scalastyle` pass on master. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5.5) Closes #59327 from viirya/SPARK-60109. Authored-by: Liang-Chi Hsieh <viirya@gmail.com> Signed-off-by: Liang-Chi Hsieh <viirya@gmail.com>

…ine alignment to the History Server end-event reparse ### What changes were proposed in this pull request? When building the application listing, `FsHistoryProvider` looks for the application end event by reopening the last event log file, skipping to `len - spark.history.fs.endEventReparseChunkSize` bytes and re-parsing the rest. This PR changes how that tail is read: - The new `FsHistoryProvider.skipToLineStart` positions the stream at the first line that starts at or after the skip target, working on raw bytes before any UTF-8 decoding. `0x0A` never occurs inside a multi-byte UTF-8 sequence, so this is safe at any byte offset. - It skips to `target - 1` and scans from there, so a line that starts exactly at `target` (possibly the end event itself) is kept instead of being discarded. - When `skip` returns 0, it reads one byte to tell EOF apart from a short skip, so it terminates when the decompressed stream is shorter than `target` (e.g. when the compressed length exceeds the decompressed one). - The partial line is scanned in bulk; bytes read past the line boundary are put back in front of the stream with a `SequenceInputStream`. - The tail is then replayed through the bounded `InputStream` reader instead of the `Iterator[String]` overload fed by `Source.fromInputStream(in)(Codec.UTF8).getLines()`, so `spark.history.fs.eventLog.maxLineLength` and the strict decoder apply on this path as well. - The new `ReplayListenerBus.replayFromOffset` takes the byte offset of the stream's first line. When it is positive, every replay diagnostic that prints a line number (JSON errors, `Malformed line #N`, the truncation warning and the over-long line warning) adds `(lines counted from uncompressed byte offset N)`, with the offset logged under its own `OFFSET` key. The source name stays the plain file path, so the `PATH` / `FILE_NAME` fields are unchanged. The existing `replay` signatures are unchanged. `maybeTruncated`, `eventsFilter` and `HaltReplayException` handling are unchanged, since all `replay` variants share `replayEntries`. Without a skip (`target <= 0`), replay starts at offset 0 and the diagnostics are as before. ### Why are the changes needed? This is the tail-reparse follow-up discussed in #59074. With the defaults (completed apps, `endEventReparseChunkSize=1m`), every completed application takes this path, and because the target is computed from the compressed file length while the skip runs on the decompressed stream, most of a compressed log can be read there. 1. The `Iterator[String]` overload bypasses the bounded line reader (SPARK-59407 / SPARK-59804), so `maxLineLength` did not apply and an over-long line was fully materialized. 2. If the skip target landed inside a multi-byte UTF-8 character (e.g. CJK SQL text or job descriptions, which Jackson writes as raw UTF-8), `source.next()` threw `MalformedInputException` before replay started. `mergeApplicationListing` only logs it, and since the placeholder listing entry was already written with the same file size, the log is not retried, so the completed application stayed missing from the listing. The target is deterministic, so restarting the History Server failed the same way. 3. Line numbers in the replay diagnostics were relative to the skip point but printed next to the full file path. This builds on #59360 (SPARK-60158, merged), which makes the bounded reader read in bulk; without it, routing the tail through the bounded reader would make this step several times slower. The two should be backported together. ### Does this PR introduce _any_ user-facing change? Yes, in the History Server: - A completed application whose end-event reparse offset lands inside a multi-byte UTF-8 character now appears in the application listing instead of being dropped. - `spark.history.fs.eventLog.maxLineLength` now also applies to the end-event reparse, so over-long lines there are skipped with a warning instead of being materialized. - Replay log messages on this path name the byte offset that their line numbers are relative to. - A line that starts exactly at the reparse offset is no longer discarded, so an end event there no longer forces a reparse of the whole log. - If the decompressed stream ends before the reparse offset, the scan no longer spins forever in the skip loop; the replay sees an empty input (previously, reaching EOF exactly at the offset failed with `NoSuchElementException` from `source.next()`). ### How was this patch tested? New tests in `FsHistoryProviderSuite`: - `end event reparse when the skip offset lands inside a multi-byte character`: the reparse offset is set to the third byte of a 3-byte character, so the line boundary scan starts on a continuation byte. Before this PR, the application is missing from the listing. - `end event reparse keeps a line that starts exactly at the skip offset`: the offset is the first byte of the end event, and a malformed line before it makes the whole-log fallback reparse fail, so the app is listed as completed only if the tail reparse finds the end event. - `end event reparse stops at the end of a stream shorter than the skip target`: a tiny lz4 log whose compressed file is larger than its content, with a 1-byte chunk; the scan must finish (it runs with a timeout). - `end event reparse applies the line length limit and reports relative line numbers` (uncompressed and lz4): an over-long line in the tail is skipped under a 1k limit, and the warning names the relative line number and the offset in the uncompressed stream. - `end event reparse reports JSON errors with the skip offset`: a malformed end event in the tail is reported with the relative line number and the offset, the path field stays the plain file name, and the application is not listed. - `skipToLineStart aligns to the first line starting at or after the target`: targets inside a line, at its `\n`, exactly at a line start, past EOF and non-positive; on each byte of a 3-byte character; CRLF; no trailing newline; a partial line longer than the read buffer; and a stream whose `skip` never advances. New test in `ReplayListenerSuite`: - `Replay from an offset reports where line numbers start`: `replayFromOffset` with a positive offset labels the over-long line warning, `Malformed line #N` and the truncated-file warning, and offset 0 leaves the diagnostics unchanged. `FsHistoryProviderSuite` (RocksDB backend), `ReplayListenerSuite`, `core/scalastyle` and `core/Test/scalastyle` pass on master. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5.5) Closes #59327 from viirya/SPARK-60109. Authored-by: Liang-Chi Hsieh <viirya@gmail.com> Signed-off-by: Liang-Chi Hsieh <viirya@gmail.com> (cherry picked from commit 003652f) Signed-off-by: Liang-Chi Hsieh <viirya@gmail.com>

…ed event log line reader ### What changes were proposed in this pull request? The bounded line reader used by `ReplayListenerBus.replay(InputStream, ...)` (added in SPARK-59407, byte accounting fixed in SPARK-59804) called `BufferedReader.read()` once per character. This PR makes it read decoded characters in bulk instead: - `boundedLines` reads from the `InputStreamReader` into an 8192-char array, instead of wrapping the reader in a `BufferedReader` and calling `read()` per character. `BufferedReader.read()` takes a lock on every call. - `BoundedLineBuffer.append(c)` is replaced by `appendUntilNewline(chars, start, end)`, which finds the next `\n` and adds up the UTF-8 length in a single pass, appends the range in one call if it fits, and returns the index of the `\n`. Once the line exceeds the allowance, it releases the retained prefix and only scans for `\n` from then on. The reader's semantics are unchanged: the UTF-8 content byte limit with the trailing-CR allowance, physical line indices including skipped lines, strict decoding, and releasing the buffer of an over-long line while it is drained. The `Iterator[String]` overload of `replay` is not affected. ### Why are the changes needed? Since SPARK-59407, line splitting in this reader is several times slower than the previous `Source.getLines()`/`readLine()` (measured at about 14x in the review of apache#59074). Every replay through the `InputStream` overload pays this: loading application UIs in the History Server, event log compaction, and listing with fast in-progress parsing. It also matters for apache#59327 (SPARK-60109), which routes the History Server end-event reparse through this reader. For compressed logs, that reparse reads most of the decompressed file, because the skip target is computed from the compressed length, so without this change the listing of completed applications would become several times slower. A local measurement of a full replay through `replay(InputStream, ...)` with the History Server listing filter (210 MB of uncompressed event lines built from `spark-events/local-1642039451826` with randomized digits, median of 3 runs, JDK 21): | codec | file size | decompression only | `Source.getLines()` + `replay(Iterator)` | bounded reader before | bounded reader after | |---|---|---|---|---|---| | lz4 | 56 MB | 0.08 s | 0.21 s | 1.13 s | 0.22 s | | zstd | 16 MB | 0.06 s | 0.17-0.18 s | 1.10 s | 0.20 s | ### Does this PR introduce _any_ user-facing change? No behavior change. Replaying event logs through the bounded reader (History Server UI loading, listing and event log compaction) is faster. ### How was this patch tested? New tests in `ReplayListenerSuite`: - `SPARK-60158: Line buffer gives the same result for any split of a line`: random lines (ASCII, CR, 2-, 3- and 4-byte characters) split at random points; the returned `\n` index and the result match a reference that drops one trailing CR and keeps the line if its UTF-8 length fits, and the buffer is released once the line exceeds the limit plus the CR allowance. - `SPARK-60158: Replay splits lines across read buffers like a reference splitter`: randomized inputs with lines up to 20000 characters, so that lines and multi-byte characters cross the 8192-char read buffer, compared with a reference that splits on `\n`, drops one trailing CR and keeps lines within the byte limit. Each input is replayed both from a `ByteArrayInputStream` and from a stream that returns 1 to N bytes per read with `available() == 0`, like codec streams. - `SPARK-60158: Replay reports physical line numbers for lines spanning read buffers`: a skipped 20000-char line followed by a malformed line, with full and short reads; the skip warning and `Malformed line #N` report the physical line numbers. The existing SPARK-59804 tests (byte limit grid across the read buffer boundary, physical line numbers, skipped line diagnostics, malformed UTF-8 and stream IO errors) pass unchanged. The SPARK-59804 line buffer test now uses `appendUntilNewline`. `ReplayListenerSuite`, `FsHistoryProviderSuite` (RocksDB backend), `EventLogFileCompactorSuite`, `core/scalastyle` and `core/Test/scalastyle` pass. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5.5) Closes apache#59360 from viirya/SPARK-60158. Authored-by: Liang-Chi Hsieh <viirya@gmail.com> Signed-off-by: Liang-Chi Hsieh <viirya@gmail.com>

…ine alignment to the History Server end-event reparse ### What changes were proposed in this pull request? When building the application listing, `FsHistoryProvider` looks for the application end event by reopening the last event log file, skipping to `len - spark.history.fs.endEventReparseChunkSize` bytes and re-parsing the rest. This PR changes how that tail is read: - The new `FsHistoryProvider.skipToLineStart` positions the stream at the first line that starts at or after the skip target, working on raw bytes before any UTF-8 decoding. `0x0A` never occurs inside a multi-byte UTF-8 sequence, so this is safe at any byte offset. - It skips to `target - 1` and scans from there, so a line that starts exactly at `target` (possibly the end event itself) is kept instead of being discarded. - When `skip` returns 0, it reads one byte to tell EOF apart from a short skip, so it terminates when the decompressed stream is shorter than `target` (e.g. when the compressed length exceeds the decompressed one). - The partial line is scanned in bulk; bytes read past the line boundary are put back in front of the stream with a `SequenceInputStream`. - The tail is then replayed through the bounded `InputStream` reader instead of the `Iterator[String]` overload fed by `Source.fromInputStream(in)(Codec.UTF8).getLines()`, so `spark.history.fs.eventLog.maxLineLength` and the strict decoder apply on this path as well. - The new `ReplayListenerBus.replayFromOffset` takes the byte offset of the stream's first line. When it is positive, every replay diagnostic that prints a line number (JSON errors, `Malformed line #N`, the truncation warning and the over-long line warning) adds `(lines counted from uncompressed byte offset N)`, with the offset logged under its own `OFFSET` key. The source name stays the plain file path, so the `PATH` / `FILE_NAME` fields are unchanged. The existing `replay` signatures are unchanged. `maybeTruncated`, `eventsFilter` and `HaltReplayException` handling are unchanged, since all `replay` variants share `replayEntries`. Without a skip (`target <= 0`), replay starts at offset 0 and the diagnostics are as before. ### Why are the changes needed? This is the tail-reparse follow-up discussed in apache#59074. With the defaults (completed apps, `endEventReparseChunkSize=1m`), every completed application takes this path, and because the target is computed from the compressed file length while the skip runs on the decompressed stream, most of a compressed log can be read there. 1. The `Iterator[String]` overload bypasses the bounded line reader (SPARK-59407 / SPARK-59804), so `maxLineLength` did not apply and an over-long line was fully materialized. 2. If the skip target landed inside a multi-byte UTF-8 character (e.g. CJK SQL text or job descriptions, which Jackson writes as raw UTF-8), `source.next()` threw `MalformedInputException` before replay started. `mergeApplicationListing` only logs it, and since the placeholder listing entry was already written with the same file size, the log is not retried, so the completed application stayed missing from the listing. The target is deterministic, so restarting the History Server failed the same way. 3. Line numbers in the replay diagnostics were relative to the skip point but printed next to the full file path. This builds on apache#59360 (SPARK-60158, merged), which makes the bounded reader read in bulk; without it, routing the tail through the bounded reader would make this step several times slower. The two should be backported together. ### Does this PR introduce _any_ user-facing change? Yes, in the History Server: - A completed application whose end-event reparse offset lands inside a multi-byte UTF-8 character now appears in the application listing instead of being dropped. - `spark.history.fs.eventLog.maxLineLength` now also applies to the end-event reparse, so over-long lines there are skipped with a warning instead of being materialized. - Replay log messages on this path name the byte offset that their line numbers are relative to. - A line that starts exactly at the reparse offset is no longer discarded, so an end event there no longer forces a reparse of the whole log. - If the decompressed stream ends before the reparse offset, the scan no longer spins forever in the skip loop; the replay sees an empty input (previously, reaching EOF exactly at the offset failed with `NoSuchElementException` from `source.next()`). ### How was this patch tested? New tests in `FsHistoryProviderSuite`: - `end event reparse when the skip offset lands inside a multi-byte character`: the reparse offset is set to the third byte of a 3-byte character, so the line boundary scan starts on a continuation byte. Before this PR, the application is missing from the listing. - `end event reparse keeps a line that starts exactly at the skip offset`: the offset is the first byte of the end event, and a malformed line before it makes the whole-log fallback reparse fail, so the app is listed as completed only if the tail reparse finds the end event. - `end event reparse stops at the end of a stream shorter than the skip target`: a tiny lz4 log whose compressed file is larger than its content, with a 1-byte chunk; the scan must finish (it runs with a timeout). - `end event reparse applies the line length limit and reports relative line numbers` (uncompressed and lz4): an over-long line in the tail is skipped under a 1k limit, and the warning names the relative line number and the offset in the uncompressed stream. - `end event reparse reports JSON errors with the skip offset`: a malformed end event in the tail is reported with the relative line number and the offset, the path field stays the plain file name, and the application is not listed. - `skipToLineStart aligns to the first line starting at or after the target`: targets inside a line, at its `\n`, exactly at a line start, past EOF and non-positive; on each byte of a 3-byte character; CRLF; no trailing newline; a partial line longer than the read buffer; and a stream whose `skip` never advances. New test in `ReplayListenerSuite`: - `Replay from an offset reports where line numbers start`: `replayFromOffset` with a positive offset labels the over-long line warning, `Malformed line #N` and the truncated-file warning, and offset 0 leaves the diagnostics unchanged. `FsHistoryProviderSuite` (RocksDB backend), `ReplayListenerSuite`, `core/scalastyle` and `core/Test/scalastyle` pass on master. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5.5) Closes apache#59327 from viirya/SPARK-60109. Authored-by: Liang-Chi Hsieh <viirya@gmail.com> Signed-off-by: Liang-Chi Hsieh <viirya@gmail.com>

[SPARK-60158][UI] Avoid per-character synchronized reads in the bound…

49c02a0

…ed event log line reader Co-authored-by: Claude Code

viirya mentioned this pull request Oct 11, 2026

[SPARK-60109][UI] Apply the replay line-length limit and UTF-8-safe line alignment to the History Server end-event reparse #59327

Closed

viirya requested review from HyukjinKwon and dongjoon-hyun October 11, 2026 06:43

dongjoon-hyun approved these changes Oct 12, 2026

View reviewed changes

Address review comments: single-pass appendUntilNewline, short-read a…

714c458

…nd line index tests Co-authored-by: Claude Code

dongjoon-hyun approved these changes Oct 12, 2026

View reviewed changes

viirya closed this in a2659a7 Oct 12, 2026

viirya deleted the SPARK-60158 branch October 12, 2026 21:36

uros-b reviewed Oct 13, 2026

View reviewed changes

viirya mentioned this pull request Oct 13, 2026

[SPARK-60109][UI][3.5] Apply the replay line-length limit and UTF-8-safe line alignment to the History Server end-event reparse #59439

Closed

		@@ -275,19 +294,44 @@ private[spark] object ReplayListenerBus {

		def append(c: Char): Unit = {

		@@ -116,15 +130,20 @@ private[spark] class ReplayListenerBus(

		@tailrec private def fetchLine(): (String, Int) = {
		val buffer = new BoundedLineBuffer(effectiveMaxLineLength)

Conversation

viirya commented Oct 11, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Was this patch authored or co-authored using generative AI tooling?

Uh oh!

dongjoon-hyun left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

viirya commented Oct 12, 2026

Uh oh!

viirya commented Oct 12, 2026

Uh oh!

uros-b left a comment

Choose a reason for hiding this comment

Uh oh!

viirya commented Oct 13, 2026

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

viirya commented Oct 11, 2026 •

edited

Loading