Repository navigation
Conversation
…ine alignment to the History Server end-event reparse 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. 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. 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()`). 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. 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> (cherry picked from commit 003652f)
dongjoon-hyun
approved these changes
Oct 13, 2026
dongjoon-hyun
left a comment
dongjoon-hyun
left a comment
Member
There was a problem hiding this comment.
+1, LGTM (Pending CIs). Thank you, @viirya .
Member
Author
|
Closing this for the reason #59450 (comment). |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
This is a backport of #59327 (SPARK-60109) to branch-3.5. It builds on SPARK-60158 (#59360), which is already on branch-3.5.
The change is the same as on master: the History Server end-event reparse aligns to the next line on raw bytes before decoding (
FsHistoryProvider.skipToLineStart), and replays the tail through the bounded reader withReplayListenerBus.replayFromOffset, sospark.history.fs.eventLog.maxLineLengthand strict decoding apply, and the diagnostics name the byte offset that their line numbers are relative to.Differences from the master commit, because branch-3.5 has no structured logging and builds with JDK 8:
ReplayListenerBusandFsHistoryProviderkeep the branch-3.5 string interpolation style, andlineOriginreturns aString. The message text is the same as on master.skipToLineStarttest reads the returned stream with GuavaByteStreams.toByteArrayinstead ofInputStream.readAllBytes."lz4"instead ofCompressionCodec.LZ4.Why are the changes needed?
The same bugs exist on branch-3.5: the end-event reparse bypasses the bounded line reader, a skip target inside a multi-byte UTF-8 character drops a completed application from the listing, the skip loop does not terminate when the decompressed stream is shorter than the target, a line starting exactly at the target is discarded, and the diagnostics on this path print skip-relative line numbers next to the full path. See #59327 for details.
Does this PR introduce any user-facing change?
Yes, in the History Server, the same as #59327: a completed application whose reparse offset lands inside a multi-byte character is listed again,
spark.history.fs.eventLog.maxLineLengthalso applies to the end-event reparse, and the replay diagnostics on this path name the byte offset that their line numbers are relative to.How was this patch tested?
The tests from #59327, adapted as described above.
FsHistoryProviderSuite(RocksDB backend),ReplayListenerSuite,core/scalastyleandcore/Test/scalastylepass on branch-3.5 with JDK 8.Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5.5)