[SPARK-59804][UI] Fix UTF-8 byte limits and physical line numbers during event log replay by viirya · Pull Request #59074 · apache/spark

viirya · 2026-09-27T00:14:11Z

What changes were proposed in this pull request?

Fix the UTF-8 byte limit and physical-line diagnostics introduced by SPARK-59407 (#58700):

Count event-log content in UTF-8 bytes, excluding line endings, while retaining only a bounded prefix of oversized lines.
Carry physical line indices across skipped lines, and log the source and line without a duplicate stack trace before rethrowing JSON parsing or mapping exceptions.
Reduce the default from 512m to 256m, keeping the worst-case buffered character count close to the previous implementation's effective default bound.
Clamp the effective UTF-8 content limit to 512m in both configuration resolution and the reader constructor, and document the heap overhead. Derive the reader default from the History configuration.
Restore the public JVM no-argument constructor while retaining the Scala default argument.
Release retained prefixes once a line is known to be oversized, identify skipped-line locations, and limit malformed-line excerpts to 1,024 characters.
Document stepped buffer growth and centralize the maximum limit in History.

The iterator-based replay API, strict UTF-8 decoding, existing truncated-last-line handling and ordinary input-stream IOException behavior are preserved. Independent follow-ups will address the unbounded compaction-filter and tail-reparse paths, tail-relative diagnostics and UTF-8 alignment, incomplete UTF-8 at EOF, bulk-read performance, and the pre-existing end-to-end test assertions.

JIRA: https://issues.apache.org/jira/browse/SPARK-59804

Release coordination: 4.2.1 RC1 contains the previous implementation. This has been raised on the Spark dev mailing list to discuss an RC2 containing the reviewed fix; no migration entry or final release-version decision is made in this PR pending that discussion.

Why are the changes needed?

The original reader compares StringBuilder.length() * 2 with a byte-valued setting. With an 8 KiB limit, it incorrectly skips a valid ASCII application-start event with a 6,000-character application name. UTF-8 multibyte characters and CRLF endings also produce inconsistent boundaries.

Dropping oversized lines before zipWithIndex renumbers subsequent diagnostics. For example, a valid event on line 1, oversized lines 2 and 3, and an invalid event on line 4 report line 2. JSON processing exceptions also need an explicit diagnostic before they are rethrown to retain that physical location.

Correcting the unit without adjusting the default permits roughly twice as many buffered characters. With a lazy line containing U+0101 followed by 600,000,000 ASCII characters, the UTF-8 reader with a 512 MiB limit runs out of memory in a separate 1 GiB heap JVM; the revised 256 MiB default successfully drains and skips that input. Buffer growth, UTF-16 storage and JSON parsing still require heap beyond the configured content limit.

A separate -Xmx3g JVM also reproduces a JVM array-size failure with 600m and 620,000,000 ASCII characters followed by U+0100, despite the line fitting within that configured limit. StringBuilder capacity growth followed by UTF-16 inflation fails independently of available heap. Clamping the effective content limit to 512m successfully skips that input; it does not guarantee adequate heap for every workload.

Does this PR introduce any user-facing change?

Yes. Events at or below the effective UTF-8 content limit are replayed, LF/CRLF endings do not consume the limit, and parsing diagnostics identify physical lines after skipped content. The default is reduced from 512m to 256m to keep its maximum buffered character count close to the previous effective bound.

Non-positive values and values above 512m now resolve to an effective maximum of 536,870,912 bytes (512 MiB), including direct constructor arguments. This clamps the content limit itself, so oversized lines are skipped instead of replayed as truncated prefixes. Smaller positive limits are unchanged.

How was this patch tested?

build/sbt 'core/testOnly *ReplayListenerSuite'
build/sbt 'core/Compile/scalastyle' 'core/Test/scalastyle'

All 19 suite tests and both scalastyle checks pass. Coverage includes the 8 KiB ASCII configuration case, raw 1- through 4-byte UTF-8 input with length and read-buffer boundary splits, line endings, directly observed buffer bounds, configuration normalization, event delivery through normalized constructors, reflective no-argument construction, buffer release, skipped-line locations, bounded malformed-line diagnostics, physical-line diagnostics without an attached throwable, JSON parsing and mapping exceptions, filtered iterators, truncation, and malformed UTF-8. The suite also runs its existing compressed/end-to-end tests, but their pre-existing assertion gaps mean they do not establish full replayed-event content equality; those test fixes remain a separate follow-up.

Targeted mutation checks confirmed that removing or doubling the buffer cap, miscounting surrogate pairs, or removing JSON line diagnostics each makes its corresponding test fail. The oversized lazy-stream scenarios were also exercised in separate -Xmx1g and -Xmx3g JVMs as described above.

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

Generated-by: OpenAI Codex (GPT-6)

…ing event log replay

dongjoon-hyun

Thank you for fixing these, @viirya. I reviewed 1df219a.

The new byte accounting and the physical line indices look correct: a faithful port of the new fetchLine was cross-checked against a reference implementation (split on \n, drop one trailing \r, keep a line iff its UTF-8 content is <= the limit, keep the physical index) on more than a million randomized inputs (1-4 byte code points, LF/CRLF/unterminated endings, limits around the boundary, Int.MaxValue) with zero mismatches.

Summary of the inline comments, most important first:

Memory bound at the default (ReplayListenerBus.scala L136): the in-loop buffer bound changed from ~maxLineLength / 2 chars to maxLineLength + 1 chars while the default stays 512m. With the default SHS heap (-Xmx1g), an over-limit line that the current code skips now fails with OutOfMemoryError before it is rejected. Please consider 256m as the default, a separate heap-oriented cap, or documenting the heap multiple.
Test coverage: nothing observes the materialization cap anymore (L139); the U+1F600 case is serialized by Jackson as the ASCII escape \ud83d\ude00, so the surrogate branches are never exercised and no multi-byte character straddles the limit (ReplayListenerSuite.scala L127); plus minor clue/redundancy nits (L146).
Diagnostics (L176): JsonParseException is an IOException, so the most common mid-file corruption is rethrown without the new physical Malformed line #N.
Pre-existing gaps outside this diff that share the semantics this PR defines (fine as follow-ups): the end-event reparse in FsHistoryProvider (see L173) and the compaction rewrite in EventFilter (see L78) still read lines with an unbounded Source.getLines(); the latter can permanently drop events of live entities after compaction.
Performance (L142, pre-existing since SPARK-59407 and kept here): the per-char locked BufferedReader.read() makes line splitting ~14x slower than the previous Source.getLines().
Nits: the "disables the limit" wording (History.scala L173), redundant overLong || (L148), a simpler width expression (L124), and the PR description says three new tests fail on the original implementation while four do (L112).

Since this changes the unit of a config that is already on master, branch-4.x, branch-4.3, branch-4.2, branch-4.1, branch-4.0 and branch-3.5 (all unreleased), it should land on all of those branches so that the same key has the same meaning everywhere.

dongjoon-hyun · 2026-09-27T21:42:59Z

+                3
+              })
+              // Allow one extra byte until we know whether the line ends in CRLF.
+              if (byteLength <= maxLineLength.toLong + 1) {


This changes the per-line memory bound, not only the unit. Previously the builder stopped growing at maxLineLength / 2 chars; now it can hold maxLineLength + 1 chars, and a single char >= U+0100 inflates it to UTF-16 (2 bytes per char). With the unchanged default of 512m:

Default SHS heap (-Xmx1g), one line of U+0101 followed by 600,000,000 ASCII chars: the current code stops at capacity 301,989,886 chars and skips the line. With this PR the builder grows to capacity 603,979,774 chars (a ~1.2 GB array) and fails with OutOfMemoryError: Java heap space before the line is rejected. case e: Exception does not catch it.

For configured limits in [576m, 1152m), a Latin-1 builder that has grown past 603,979,774 chars has capacity 1,207,959,550. AbstractStringBuilder.inflate() sizes the UTF-16 array from the capacity, so the first char >= U+0100 fails with UTF16 String size is 1207959550, should be less than 1073741823 regardless of the heap size.

On JDK 8 (branch-3.5) there are no compact strings, so even pure-ASCII lines need about twice the heap of the current code.

The docs still say the setting bounds "the memory replay can use". Could we default to 256m (keeps the current bound for ASCII lines), keep a separate heap-oriented cap for the buffer, or document that heap usage can reach about 2-4x the limit?

Confirmed. I reproduced the OOM in a separate -Xmx1g JVM using a lazy stream containing U+0101 followed by 600,000,000 ASCII characters. The 512 MiB limit fails during builder growth; the revised implementation with a 256 MiB limit successfully drains and skips the same input.

Changed both defaults to 256m in ab7c85d to keep the maximum buffered character count close to the previous effective bound. The configuration docs now describe the additional heap cost of buffer growth, UTF-16 storage and JSON parsing. This does not guarantee that every accepted 256 MiB event fits in a 1 GiB heap.

dongjoon-hyun · 2026-09-27T21:42:59Z

+              if (byteLength <= maxLineLength.toLong + 1) {
+                sb.append(c.toChar)
+              } else {
+                overLong = true


Test coverage: because L148 now re-decides from the full byteLength, this cap is no longer observable by any test. A mutant that always appends (never sets overLong), or one that uses 2 * maxLineLength + 1, produces the same lines, indices, warnings and exceptions for every input in the suite; only the buffered size of skipped lines changes (e.g. 8193 -> 9216 chars in "Over-long event log lines are skipped instead of materialized"). Before this PR, removing the cap made that test fail. A test that observes the buffer bound would guard SPARK-59407's main guarantee, e.g. by extracting the reader into a helper and asserting the max buffered length, or by streaming a huge line from a lazy InputStream with a small limit.

Addressed in ab7c85d. The reader now uses a BoundedLineBuffer, and the new test checks its retained character count after every append while feeding content well beyond the limit. It covers zero and small limits and allows only the one extra character reserved for a possible trailing CR.

I also tested mutations that remove the append cap or double it. Both cause this test to fail, even though the oversized line would still be rejected at the end.

dongjoon-hyun · 2026-09-27T21:42:59Z

+                overLong = true
+              }
            }
            c = reader.read()


Performance (pre-existing since SPARK-59407, kept by this rewrite): BufferedReader.read() takes a lock on every call (on JDK 21 an InternalLock/ReentrantLock, whose CAS survives C2 inlining). On ~100 MB of TaskEnd-like ASCII lines this loop splits at ~73 MB/s (13.6 s/GB) vs ~1,039 MB/s (0.96 s/GB) for the previous Source.getLines(); together with Jackson readTree it is 60.9 MB/s vs 293 MB/s. That is roughly +12.6 s of CPU per GB for UI rebuilds and compaction, and ~8 s to drain a 600 MiB line. The UTF-8 width branching added here is not measurable. Keeping the same per-char logic but reading into a local char[] via InputStreamReader.read(char[], ...) is ~5x faster, and bulk-scanning the buffer for \n with slice appends is ~15x faster. Maybe a follow-up?

Agreed that this deserves a separate follow-up. We will address block-oriented reading and validate it with an SHS replay benchmark there. This PR keeps the existing read strategy and focuses on the byte-limit and diagnostic fixes; I have not independently rerun the throughput measurements quoted here.

dongjoon-hyun · 2026-09-27T21:43:00Z

+            sb.setLength(sb.length() - 1)
+            byteLength -= 1
+          }
+          if (overLong || byteLength > maxLineLength) {


nit: overLong || is redundant. overLong is only set when byteLength >= maxLineLength + 2, byteLength no longer changes afterwards, and the CR strip subtracts at most 1, so byteLength > maxLineLength already holds (also for Int.MaxValue, 0 and negative limits; checked exhaustively for small limits plus random inputs). Either use if (byteLength > maxLineLength), or drop the flag and keep only if (byteLength <= maxLineLength + 1L) sb.append(c.toChar) in the loop, which removes one nesting level.

Removed overLong in ab7c85d. The bounded accumulator stops updating once the byte count exceeds the buffer allowance, and the final content-byte check determines whether to skip the line. This also avoids continuing to count the rest of an arbitrarily long discarded line.

dongjoon-hyun · 2026-09-27T21:43:00Z

+            if (!overLong) {
+              // The strict decoder guarantees valid surrogate pairs. Count their four UTF-8
+              // bytes on the high surrogate and zero on the low surrogate.
+              byteLength += (if (c < 0x80) {


nit: since the strict decoder only emits complete surrogate pairs, counting 2 bytes per surrogate half gives identical results (the running total matches at every code point boundary), so this could be if (c < 0x80) 1 else if (c < 0x800 || Character.isSurrogate(c.toChar)) 2 else 3, and the comment above becomes simpler.

Applied in ab7c85d: each surrogate half contributes two bytes, with a comment explaining the strict decoder's valid-pair guarantee. The new raw UTF-8 tests exercise supplementary code points and limits inside those code points. A mutation counting three bytes per half fails those tests.

dongjoon-hyun · 2026-09-27T21:43:00Z

+    replayEntries(lines.zipWithIndex, sourceName, maybeTruncated, eventsFilter)
+  }
+
+  private def replayEntries(


The physical line number only surfaces for non-IOException failures. JsonParseException (and JsonMappingExceptions such as MismatchedInputException) extend IOException, so for a malformed JSON line in the middle of a file, throw jpe (L219) is caught by case ioe: IOException => throw ioe (L233) before the Malformed line #N log (L237). For example, [end, 2048 x's, 2048 x's, {bad, end] with a 1024 limit throws without any line number, and FsHistoryProvider only logs Jackson's line: 1, column: N, which is relative to the single-line string. {} in the new test goes through the NPE path. The case ordering is old (SPARK-2261), but logging sourceName/lineNumber before rethrowing would make this fix effective for the most common corruption.

Confirmed and fixed in ab7c85d. JsonProcessingException now logs the source and physical line number before the same exception is rethrown. Tests cover both malformed JSON and a mapping error on physical line 4 after two skipped lines.

Truncated-last-line handling and ordinary stream IOException propagation are preserved. A separate test checks that a stream read failure does not receive a stale JSON parse-line diagnostic. Removing the new JSON diagnostic makes its regression test fail.

dongjoon-hyun · 2026-09-27T21:43:00Z

+      .doc("Maximum UTF-8 byte length of a single event log line during replay, excluding " +
+        "the line ending. Longer lines are skipped with a warning, bounding the " +
        "memory replay can use when an event log is corrupt or unexpectedly large. Setting " +
        "this to 0 or a negative value disables the limit. " +


With the new Long arithmetic, "disables the limit" is no longer strictly true. ReplayListenerBus.maxLineLength maps <= 0 (and values above Int.MaxValue) to Int.MaxValue, and lines with more than 2,147,483,647 UTF-8 bytes are now skipped with a warning (e.g. 2^30 x U+00E9 is 2,147,483,648 bytes but only a 1 GiB Latin-1 builder), whereas the previous sb.length() * 2 < Int.MaxValue overflowed and never skipped. Spark 3.4+ writers cannot produce such a line (one ByteArrayOutputStream per event), so this is mostly theoretical, but skipping the check when the limit is disabled, or documenting the 2^31-1 byte cap, would keep the doc accurate.

Corrected the configuration docs and helper comment in ab7c85d. They now state that non-positive values and values above Int.MaxValue use the maximum supported limit of 2,147,483,647 bytes. The existing configuration normalization is unchanged. Added tests for zero, negative, exact-maximum and larger values.

dongjoon-hyun · 2026-09-27T21:43:00Z

+  test("Replay line limit handles UTF-8 boundaries and line endings") {
+    val end = JsonProtocol.sparkEventToJsonString(SparkListenerApplicationEnd(1000L))
+    // scalastyle:off nonascii
+    val names = Seq("x", "\u00e9", "\u4e2d", "\ud83d\ude00")


The U+1F600 case does not exercise the 4-byte path. JsonProtocol serializes through Jackson's UTF8JsonGenerator, which writes supplementary characters as the ASCII escape \ud83d\ude00 (COMBINE_UNICODE_SURROGATES_IN_UTF8 is off by default in 2.22.2). So that JSON line is pure ASCII, the surrogate branches (ReplayListenerBus.scala L128-L131) are never taken, and mutants that count 3+3, 3+0 or 0+0 bytes for a pair all pass. Also, the boundary always falls on the trailing ASCII ","Timestamp":125,"User":"user"}, so no multi-byte character straddles the limit, and since the name starts at byte offset 110, the 8192-byte decoder boundary never splits a U+00E9/U+4E2D sequence (8192 - 110 = 8082 is divisible by 2 and 3). Could we build raw UTF-8 input directly (e.g. lines whose last code point is 2, 3 or 4 bytes) and put the limit at -1/0/+1/+2 bytes around it?

Confirmed: Jackson escapes U+1F600 in the original test. Replaced that grid in ab7c85d with raw UTF-8 input observed through the replay filter before JSON parsing.

The final code point is 1–4 bytes wide and starts at offsets 8, 8190, 8191 or 8192. Limits range from three bytes below to two bytes above the content length, covering positions inside multibyte code points and sequences split across the reader's byte buffer. LF, CRLF and unterminated final lines are included. A mutant counting supplementary characters as 3+3 bytes fails the new grid.

dongjoon-hyun · 2026-09-27T21:43:00Z

+        .getBytes(StandardCharsets.UTF_8))
+      assert(bus.replay(input, "utf8"))
+      val expected = if (excess == 0) Seq(end, json) else Seq(end)
+      assert(listener.loggedEvents.toSeq == expected, s"name=$name ending=$ending excess=$excess")


nit: (1) The clue omits repetitions and prints ending raw, so \n and \r\n failures look identical, e.g. s"name=$name repetitions=$repetitions ending=${ending.map(_.toInt).mkString(",")} excess=$excess"; the assert(bus.replay(input, "utf8")) above has no clue at all. (2) repetitions 100 vs 4097 gives the same outcome for every (name, ending, excess) combination, so it only doubles the grid (see the decoder-boundary note on L127). (3) "Replay line limit uses UTF-8 bytes" is subsumed by the name = "x", excess = 0 iterations here.

Updated in ab7c85d. Both replay success and observed-line assertions are wrapped in a clue containing the code point, padding, numeric line-ending bytes and limit delta. The old repetitions grid is replaced by deliberate boundary offsets.

I retained the short ASCII application-event test because it now also exercises the 8k configuration and successful JSON event delivery. The raw UTF-8 grid deliberately bypasses JSON parsing so serialization cannot hide the byte sequences being tested.

dongjoon-hyun · 2026-09-27T21:43:00Z

    assert(eventMonster.loggedEvents(1) === JsonProtocol.sparkEventToJsonString(applicationEnd))
  }

+  test("Replay line limit uses UTF-8 bytes") {


minor, about the PR description: it says "Three new regression tests fail against the original implementation", but four of the five new tests fail on the current code (this one, the boundary grid, the physical-line-number test and the truncated/filtered test); only "Replay still rejects malformed UTF-8 in skipped lines" passes there. Since the description becomes the commit message, it may be worth correcting.

Confirmed by rerunning all five tests from the initial PR revision against the original implementation: four fail, and the malformed-UTF-8 test passes. The earlier count of three came from a run before the truncated/filtered test was added. I removed the stale count from the PR description.

The revised suite has 15 passing tests, and both production and test scalastyle checks pass.

HyukjinKwon

Reviewed ab7c85d. The change looks correct to me, and the points from the earlier review that were marked addressed hold up on this revision:

BoundedLineBuffer byte accounting is right for UTF-8 (1/2/3 bytes, and 2 per surrogate half = 4 per valid pair from the strict decoder). Retention is capped at maxLineLength + 1 bytes' worth of chars, and a trailing CR is only discounted when it is the last retained char. So an over-long line with a retained interior CR is still rejected, and a code point straddling the bound is dropped and rejected.
Physical line numbers: the index is assigned before draining, so skipped lines still consume a number. The Iterator[String] overload keeps zipWithIndex, and the filter is applied after indexing on both paths, so the truncated-last-line detection is unchanged.
The new JsonProcessingException case only adds context. Previously these fell into the IOException rethrow with no source/line. lineNumber is assigned before parsing, and reader IO errors (including MalformedInputException) do not reach this case, which the IO test covers.
Default 256m UTF-8 bytes keeps the ASCII character capacity equal to the old 512m / 2 cap. All three production callers (FsHistoryProvider x2, EventLogFileCompactor) derive the limit from ReplayListenerBus.maxLineLength(conf). Since the config was only added in SPARK-59407 and has not shipped in a release yet, redefining its unit and default here is not a compatibility concern.
The tests discriminate the fix: the 8k ASCII test fails under the old chars * 2 rule, and the raw-UTF-8 grid crosses the decoder's 8192 buffer boundary for each encoded width and line ending.

The deferred items (tail reparse via the unbounded reader, compaction reader consistency, per-char BufferedReader.read() locking) make sense as the planned follow-ups.

One small readability nit inline.

HyukjinKwon · 2026-09-28T02:59:33Z

          }
-          if (overLong) {
+          val line = buffer.result()
+          if (line.isEmpty) {


nit: line is an Option[String] here, so line.isEmpty reads like an empty-string check at first glance (and an empty line is in fact accepted as Some("")). Something like buffer.result() match { case Some(line) => (line, index); case None => ... }, or naming it maybeLine, would make the skip branch clearer.

Renamed it to maybeLine in aaee04e to make the Option[String] semantics clear. Thanks!

dongjoon-hyun

Thank you for the update, @viirya. I reviewed aaee04e.

The points from the previous round that were marked as addressed hold up. I re-checked the new BoundedLineBuffer and the physical line indices with a verbatim copy of the reader: it matches a reference implementation on 900K randomized inputs (1-4 byte code points, CR/CRLF/unterminated endings, malformed UTF-8, streams split at arbitrary byte positions, limits around the boundaries and 0/negative/Int.MaxValue) with zero mismatches, and mutants of the surrogate width, the CR discount and the buffer allowance are all caught.

Summary of the inline comments, most important first:

OOM instead of skip for large limits (ReplayListenerBus.scala L256): retaining limit + 1 chars can exceed StringBuilder's implementation limits (Latin-1 to UTF-16 inflation from 576m, the UTF-16 maximum length from about 1g), so OutOfMemoryError is thrown regardless of the heap size where the current code skips the line, and even for some lines within the limit. On the listing path the error is swallowed silently. Clamping the effective limit (e.g. 512m) and documenting it as the maximum would fix it.
Backports (History.scala L181): this needs to land on branch-4.3 before the next 4.3.0 RC (v4.3.0-rc1 (2026-08-31) does not contain SPARK-59407, but branch-4.3 does), and branch-3.5 needs a separate PR because the new structured-logging block merges cleanly there but does not compile.
Duplicate stack traces (L210): the new log passes jpe although every caller already logs or retries the same exception, so UI loads now add 2-4 ERROR stack traces per request. Logging only the location would be enough.
Test coverage (ReplayListenerSuite.scala L259, pre-existing): the end-to-end replay tests compare e1 with e1 and ignore the result, so the "existing compressed/end-to-end replay tests" in the description don't check the replayed events.
Tail-reparse follow-up (L148): the new ERROR prints skip-relative line numbers on that path, and (pre-existing) a skip that lands inside a multi-byte sequence drops a completed app from the listing, which reusing boundedLines would not fix by itself.
Truncation inside a code point (L83, pre-existing): an in-progress log cut inside a multi-byte sequence fails with MalformedInputException instead of being tolerated as truncated.
Nit (L239): derive DEFAULT_MAX_LINE_LENGTH from the config default and describe the current bound in its comment.

Items 4-6 are pre-existing and fine as follow-ups.

dongjoon-hyun · 2026-09-28T05:55:38Z

+  private[scheduler] class BoundedLineBuffer(maxLineLength: Int) {
+    private val buffer = new java.lang.StringBuilder()
+    private var byteLength = 0L
+    private val bufferLimit = maxLineLength.toLong + 1


This retains up to maxLineLength + 1 chars, but nothing caps that against the StringBuilder implementation limits, so larger configured limits now fail with OutOfMemoryError, regardless of -Xmx, where the current code skips the line:

Latin-1 to UTF-16 inflation (limits >= 603,979,775 bytes, i.e. from 576m). Once more than 603,979,774 Latin-1 chars are retained, the builder's capacity is 1,207,959,550, and the first char >= U+0100 makes inflate() fail with UTF16 String size is 1207959550, should be less than 1073741823. This also hits lines within the limit: with 600m, 620,000,000 x followed by U+0100 (620,000,002 bytes) throws, while the current code skips it. The current code needs 1152m or more for this. It is the [576m, 1152m) case from the first round, which the new default avoids only for the default.

UTF-16 maximum length (limits >= 1,073,741,823 bytes, about 1g). A UTF-16 builder cannot exceed 1,073,741,822 chars, so e.g. € followed by about 1.18 billion x at 1g throws Requested array size exceeds VM limit, while the current code skips it.

<= 0 and values above Int.MaxValue fail the same way as the current code, but the new doc now calls 2147483647 bytes the "maximum supported limit".

On the listing path, the Error passes mergeApplicationListing's case e: Exception and is swallowed by the Future of pool.submit in submitLogProcessTask, so the app is silently missing from the listing. 256m and an explicit 512m are safe.

Could we clamp the effective limit in maxLineLength(conf) (and in the constructor if needed), e.g. to 512m (any value up to 603,979,773 keeps the capacity at or below 603,979,774), and document that as the maximum? The clamp needs to apply to the limit itself, not only to bufferLimit, so that lines within the limit are not silently truncated. I reproduced these with a verbatim copy of the reader on JDK 21 (up to -Xmx8g).

I reproduced the 600m / 620,000,000 ASCII characters + U+0100 case with the actual reader on JDK 21 and -Xmx3g. The effective content limit is now capped at 512m in both configuration resolution and the constructor, and the same input is successfully skipped. Tests cover normalization and the docs distinguish this JVM limit from heap requirements.

dongjoon-hyun · 2026-09-28T05:55:38Z

      .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
      .bytesConf(ByteUnit.BYTE)
-      .createWithDefaultString("512m")
+      .createWithDefaultString("256m")


A reminder about the backports from the first round: this changes the unit and the default of a key that branch-4.x, branch-4.3, branch-4.2, branch-4.1, branch-4.0 and branch-3.5 already have (all still 512m with the chars * 2 check, and not in any release yet).

v4.3.0-rc1 (2026-08-31) does not contain SPARK-59407, but branch-4.3 does, so this needs to land on branch-4.3 before the next 4.3.0 RC. Otherwise 4.3.0 ships the old semantics (e.g. 8k means about 4096 chars and <= 0 means no limit), and changing it afterwards becomes a behavior change of a released config.

The diff cherry-picks cleanly to branch-4.x and branch-4.3, and branch-4.2/4.1/4.0 conflict only in the scaladoc (byes/bytes). On branch-3.5, the new case jpe: JsonProcessingException block with log"..."/MDC merges without a conflict but does not compile, so it needs a separate PR like [SPARK-59407][UI][3.5] Improve compressed line handling in Spark History server #59058.

Confirmed: v4.3.0-rc1 does not contain SPARK-59407, while branch-4.3 does. This fix needs to reach branch-4.3 before the next RC, along with the other affected branches. Branch-3.5 needs a separate backport adapting the logging API.

dongjoon-hyun · 2026-09-28T05:55:38Z

      case _: EOFException if maybeTruncated => false
+      case jpe: JsonProcessingException =>
+        logError(log"Exception parsing Spark event log: ${MDC(PATH, sourceName)} " +
+          log"at line ${MDC(LINE_NUM, lineNumber)}", jpe)


Thanks for adding the location. Passing jpe here duplicates the stack trace, though, since every caller already logs or retries the same IOException:

UI loads: createHybridStore, createDiskStore and createInMemoryStore (FsHistoryProvider L1598, L1672, L1701) retry once on IOException, so one malformed line now adds 2 ERROR stack traces per request (4 with the hybrid store), where the current code logs none at ERROR (only the INFO "trying again"). Failed loads are not cached, so this repeats on every refresh.

Listing (L1032) and compaction (L1224) log the same exception again at ERROR.

Could we drop the throwable and keep only the source and line (optionally with jpe.getOriginalMessage)? The new test only checks the message, so it would still pass.

Removed the throwable from the location diagnostic and kept the same exception rethrow. The tests now assert a single location diagnostic with getThrown == null for both parsing and mapping errors.

dongjoon-hyun · 2026-09-28T05:55:38Z

+  }
+
  /**
   * Test replaying compressed spark history file that internally throws an EOFException.  To


About the "existing compressed/end-to-end replay tests" in the PR description: they don't really check the replayed events (pre-existing, not from this PR).

testApplicationReplay, used by "End-to-end replay" and "End-to-end replay with compression", compares each event with itself, JsonProtocolSuite.assertEquals(e1, e1) (L408). SPARK-28770 changed it from (e1, e2), which looks like a typo. It also zips without a size check and ignores the result of replay (L396), so it passes even if nothing is replayed.

This test only checks the event count (L300) and also ignores the result of replay (L298).

So nothing checks the content of a multi-buffer LZ4/Snappy/ZSTD log read through the new reader. Could we assert the result, compare originalEvents.size with replayedEvents.size and use assertEquals(e1, e2) (and pick the current app's log instead of the latest file in the shared directory), here or in a small test-only follow-up? Otherwise, please adjust the claim in the description.

Confirmed the pre-existing assertion gaps. I have narrowed the testing description; we will fix the replay result, event count/content assertions and log-file selection in a separate test follow-up.

dongjoon-hyun · 2026-09-28T05:55:38Z

      sourceName: String,
      maybeTruncated: Boolean,
      eventsFilter: ReplayEventsFilter): Boolean = {
+    replayEntries(lines.zipWithIndex, sourceName, maybeTruncated, eventsFilter)


Two notes for the tail-reparse follow-up (FsHistoryProvider L1104-L1127), which calls this overload:

There, lines starts after the byte skip and the dropped partial line, so the new ERROR at L208-L211 prints the full path with a skip-relative number. For example, a truncated SparkListenerApplicationEnd line at physical line 12,004 of a completed app is reported as at line 3959. Before this PR, JSON errors on this path had no bus-level line number. Please cover it in the follow-up, together with the existing Malformed line #N and the truncation WARN, which are also skip-relative there, or label the source on that path, e.g. s"$path (lines counted after skipping $target bytes)".

Pre-existing: if target lands inside a multi-byte UTF-8 sequence (e.g. CJK SQL text or job descriptions, which Jackson writes as raw UTF-8), source.next() (L1124) throws MalformedInputException before replay is called. mergeApplicationListing only logs it, and since the placeholder LogInfo was already written with the same file size, shouldReloadLog never retries it, so the completed app stays missing from the listing (and target is deterministic, so a restart fails the same way). Replaying through boundedLines after the skip, as planned, would throw the same way because of its strict decoder, so the follow-up should drop the bytes up to the first \n before decoding. I reproduced it at the unit level, not yet on a real SHS.

Confirmed the skip-relative numbering and reproduced source.next() failing when the offset lands inside a UTF-8 sequence. We will cover both in the tail-reparse follow-up, including discarding the partial line as raw bytes before decoding and correcting the diagnostic location. The SHS listing/retry behavior still needs an integration regression test.

dongjoon-hyun · 2026-09-28T05:55:38Z

+      sourceName: String): Iterator[(String, Int)] = {
    // Fail on malformed input like Source.getLines() does instead of replacing it.
    val decoder = StandardCharsets.UTF_8.newDecoder()
      .onMalformedInput(CodingErrorAction.REPORT)


Pre-existing and preserved by this PR (Source.getLines() with Codec.UTF8 also fails on malformed input), but related to the truncation handling: if the last, unterminated line of an in-progress or crashed app's log ends inside a multi-byte UTF-8 sequence, this decoder throws MalformedInputException from reader.read() at EOF. That happens in lineEntries.hasNext, outside the inner try, so it never reaches the maybeTruncated tolerance for JsonParseException; it is rethrown as an IOException, and the UI rebuild fails after its retry (permanently for a crashed app). The same log cut at an ASCII boundary replays fine.

Compressed in-progress logs expose lengths at compression block boundaries (128 KiB blocks for zstd, 32 KiB for lz4) rather than at code point boundaries, so with CJK-heavy content this happened in 3/80 (zstd) and 5/80 (lz4) simulated cuts; uncompressed local files were always cut at code point boundaries in my runs. A follow-up could treat a MalformedInputException raised after the underlying stream hit EOF as a truncated last line when maybeTruncated is set.

Reproduced: an ASCII-truncated final JSON line is tolerated with maybeTruncated, while a partial UTF-8 code point at EOF throws MalformedInputException. We will handle this separately, including compressed-log regression coverage.

dongjoon-hyun · 2026-09-28T05:55:38Z

+   * Keep the maximum buffered character count close to the original 512 MiB / 2 cap.
   */
-  val DEFAULT_MAX_LINE_LENGTH: Int = 512 * 1024 * 1024
+  val DEFAULT_MAX_LINE_LENGTH: Int = 256 * 1024 * 1024


nit: This duplicates the 256m default of History.EVENT_LOG_MAX_LINE_LENGTH, and this PR had to change both. Deriving it, e.g. History.EVENT_LOG_MAX_LINE_LENGTH.defaultValue.get.toInt (like ChunkedByteBuffer does with BUFFER_WRITE_CHUNK_SIZE; the API is the same down to branch-3.5), keeps them in sync. Also, "the original 512 MiB / 2 cap" refers to the unreleased code that this PR replaces, so it will be hard to interpret after the merge. Could the comment describe the current bound instead (a skipped line retains at most this many chars plus a possible trailing CR while it is drained) and leave the history to the PR description?

The reader default now comes from History.EVENT_LOG_MAX_LINE_LENGTH.defaultValue. I also replaced the historical comment with a description of the shared default.

…aces

dongjoon-hyun

Thank you for the updates, @viirya. I reviewed the latest commit (691a1d7) again.

The new byte accounting looks correct to me. A verbatim port of boundedLines and BoundedLineBuffer matched a byte-level reference model (split on \n, strip one trailing CR, keep a line iff its UTF-8 content is at most the limit, physical indices) on more than 60 million generated inputs, covering 1-4 byte code points, CR/CRLF/EOF endings, exact-limit lines, lines split across the decoder buffer and malformed bytes. It also matched on randomized inputs with Scala 2.12/2.13 on JDK 8, 17, 21 and 25. The 512 MiB clamp keeps the StringBuilder capacity around 604M chars on those JDKs, well below the UTF-16 limit.

I left the remaining comments inline, roughly in order of priority:

Release timing (History.scala L177): v4.2.1-rc1, cut today, already ships SPARK-59407 with the old semantics.
Binary compatibility (L42): SPARK-59407 removed the public no-arg constructor that every released ReplayListenerBus has, which breaks existing tools such as Sparklens and aws-emr-advisor.
Memory: the dead prefix is retained while an over-long line is drained (L271), and the buffer size is a step function of the limit (L260).
Test gaps: the constructor normalization is not exercised through replay (ReplayListenerSuite L180), and the stale-line test never parses a line (ReplayListenerSuite L233).
Diagnostics: the skip warning has no location or count (L128), and the generic branch still logs the whole line (L217).
Follow-up ideas: byte-level line splitting (L88) and per-line buffer allocation (L114).
Nits: zipWithIndex (L119), JIRA IDs in the new test names (ReplayListenerSuite L115), the 536870912 literal (History.scala L174), and the constructor scaladoc (L38).

dongjoon-hyun · 2026-09-28T18:36:25Z

+        "space. Values at or below 0, or above 536870912, use the maximum supported limit of " +
+        "536870912 bytes (512 MiB). This cap avoids JVM array-size limits but does not " +
+        "guarantee sufficient heap space. " +
        "Introduced in 4.3.0; also available in 3.5.10, 4.0.5, 4.1.4 and 4.2.1; and in " +


Following up on my earlier backport comment: v4.2.1-rc1 (cut 2026-09-28 06:11 UTC, about an hour before the latest commit here) already contains SPARK-59407 with the old semantics: .version("4.2.1"), the 512m default, the sb.length() * 2 check, and "Setting this to 0 or a negative value disables the limit". branch-4.2 has already moved to 4.2.2-SNAPSHOT without this fix. My earlier comment only covered branch-4.3, assuming nothing had shipped yet.

If 4.2.1 is finalized from RC1 and this PR is backported afterwards, a patch release (4.2.2) would change the unit, the default and the <= 0 meaning of a released config:

Default users: the same for ASCII, but stricter for non-ASCII-heavy lines (e.g. 268M -> 134M chars for 2-byte text, 268M -> ~89.5M chars for 3-byte text).

8k: an ApplicationStart line with 3,000 x U+4E2D (3,085 chars / 9,085 bytes) is replayed by RC1 but skipped here, so the app is missing from the listing.

0 / negative / 3g: RC1 never skips (sb.length() * 2 < Int.MaxValue is always true), while this PR clamps to 512 MiB, so there is no way to replay events above 512 MiB anymore.

An explicit 512m (RC1's documented default): RC1 keeps at most 268,435,456 chars and skips U+0101 followed by 600,000,000 ASCII chars with -Xmx1g, while this PR grows the builder to 603,979,774 chars (~1.2 GB as UTF-16) and fails with OutOfMemoryError on UI load or compaction.

Could we get this into branch-4.2 before 4.2.1 is finalized, e.g. by raising it on the RC1 vote so that it lands in RC2? Otherwise, we need a core-migration-guide.md entry for 4.2.1 -> 4.2.2 and should qualify "also available in ... 4.2.1" here and in monitoring.md.

Confirmed that v4.2.1-rc1 contains the old implementation and branch-4.2 is now 4.2.2-SNAPSHOT. I raised this on the Spark dev mailing list so we can decide whether to produce an RC2 with the reviewed fix before finalizing 4.2.1. I am holding off on migration changes pending that discussion.

dongjoon-hyun · 2026-09-28T18:36:25Z

+ *                      ending. Longer lines are drained, skipped and logged, bounding the
+ *                      memory replay can use when an event log is corrupt or unexpectedly large.
 */
 private[spark] class ReplayListenerBus(


Binary compatibility (pre-existing from SPARK-59407, but this PR is a good place to fix it before the maintenance releases): every released version (3.5.9, 4.0.4, 4.1.3, 4.2.0) declares class ReplayListenerBus extends SparkListenerBus with Logging, which has a public no-arg constructor. With (maxLineLength: Int = DEFAULT_MAX_LINE_LENGTH), scalac only emits <init>(I)V; default arguments don't generate a no-arg constructor.

The class is private[spark] and excluded from MiMa, but event-log tools already use it and would break on 3.5.10 / 4.0.5 / 4.1.4 / 4.2.1:

Sparklens: Class.forName("org.apache.spark.scheduler.ReplayListenerBus") + newInstance() (EventHistoryReporter.scala#L20-L21) -> InstantiationException.

aws-emr-advisor: getDeclaredConstructor().newInstance() (EmrSparkLogParser.scala#L75-L76) -> NoSuchMethodException, and class CustomReplayListenerBus extends ReplayListenerBus (CustomReplayListenerBus.scala#L25) -> NoSuchMethodError: 'void <init>()' when compiled against a released version.

The reflective cases fail even after recompiling. Could we add an auxiliary constructor?

def this() = this(ReplayListenerBus.DEFAULT_MAX_LINE_LENGTH)

It compiles on Scala 2.12 and 2.13 with or without the default argument, new ReplayListenerBus() is not ambiguous, and all in-repo call shapes still compile, including new ReplayListenerBus(maxLineLength = ...) and the anonymous subclass in StreamingQueryListenerSuite. If 4.2.1 ships from RC1, keeping the default argument as well also preserves $lessinit$greater$default$1 for code compiled against 4.2.1.

Confirmed with the compiled class: the no-argument constructor was missing. I restored it while retaining the default argument, and added a reflection regression test. The compiled class now exposes both constructors and the Scala default-argument method.

dongjoon-hyun · 2026-09-28T18:36:26Z

+        // The strict decoder emits valid surrogate pairs: count two bytes per half.
+        byteLength += (if (c < 0x80) 1 else if (c < 0x800 || Character.isSurrogate(c)) 2 else 3)
+        // Reserve one extra byte for a possible CR in the line ending.
+        if (byteLength <= bufferLimit) {


Once byteLength exceeds bufferLimit, the line is certain to be skipped, but the retained prefix stays reachable until the \n is read. With the default 256m, that is ~302 MB (Latin-1) or ~604 MB (UTF-16, i.e. any char >= U+0100 early in the line) of dead heap for the whole drain (~13 s per GB at the current read rate), overlapping with concurrent UI loads and compaction. In a -Xmx1g model, bus A draining U+0101 + ~868 MB of ASCII made bus B fail with OutOfMemoryError on a 200 MB line, and releasing the buffer on overflow made it pass.

This is pre-existing from SPARK-59407, but it's a small fix:

if (byteLength <= bufferLimit) { buffer.append(c) } else { // The line will be skipped; release the retained prefix while draining. buffer.setLength(0) buffer.trimToSize() }

After overflow byteLength >= maxLineLength + 2, so result() still returns None regardless of a trailing CR. It doesn't lower the peak, only how long the memory is held. The assertion at ReplayListenerSuite L164 (buffer.length == limit + 1) would become == 0.

The retained prefix is now cleared and trimmed once overflow makes the line unconditionally skippable. The buffer test checks both zero length and zero capacity after overflow; this reduces retention during draining, not the allocation peak.

dongjoon-hyun · 2026-09-28T18:36:26Z

+
+  /** Accumulates a bounded prefix of one decoded line, allowing a possible trailing CR. */
+  private[scheduler] class BoundedLineBuffer(maxLineLength: Int) {
+    private val buffer = new java.lang.StringBuilder()


FYI, because this StringBuilder starts at 16 and grows as 2c + 2, the retained buffer is a step function of the limit rather than proportional to it. For ASCII-heavy lines, every limit from 150,994,942 to 301,989,885 bytes (~144m to ~288m) ends with the same 301,989,886-char buffer as the 256m default, and ~288m to 512m all end with 603,979,774 chars.

So if an operator hits an OOM while a long line is drained and lowers the limit to 200m or 150m, nothing changes. With -Xmx800m, U+0101 followed by 300,000,000 ASCII chars failed at the same point (length 150,994,942, when the 302 MB UTF-16 array is copied into a 604 MB one) for 256m, 200m and 150m, while 143m and 128m skipped it. Could the docs mention that only limits below ~144m (e.g. 128m) reduce the buffer, or could the growth be capped at bufferLimit?

Confirmed the capacity steps from the JDK growth rule. I documented that lowering 256m to 200m or 150m may leave the same buffer capacity, with 128m as an example of a smaller step. Changing buffer growth remains part of the reader follow-up.

dongjoon-hyun · 2026-09-28T18:36:26Z

+    }
+  }
+
+  test("Replay constructor normalizes the effective line limit") {


Test coverage: this only reads effectiveMaxLineLength, and no test replays through a bus built with an out-of-range limit. So reverting ReplayListenerBus L114 to the raw parameter (new BoundedLineBuffer(maxLineLength), as in the base code) passes the whole suite, although new ReplayListenerBus(0) would then skip every non-empty line, and -1 / Int.MinValue every line. All production callers pass the already-normalized maxLineLength(conf), so there's no user impact today, but the hand-ported branch-3.5 backport could regress without a failing test. Could we replay one event through the 0, -1 and Int.MinValue buses in the first loop and assert that it arrives?

The constructor test now replays an event and checks delivery for every out-of-range argument in the loop, including 0, -1 and Int.MinValue.

dongjoon-hyun · 2026-09-28T18:36:26Z

        if (c == -1) {
          null
        } else {
+          val index = lineIndex


nit: This hand-rolls what zipWithIndex already does at L151. If the reader emits one Option per physical line, both overloads number lines the same way, and the manual counter, the tuple-typed state and the @tailrec recursion go away:

var warned = false Iterator.continually(reader.read()).takeWhile(_ != -1).map { first => val buffer = new BoundedLineBuffer(effectiveMaxLineLength) var c = first while (c != -1 && c != '\n') { buffer.append(c.toChar) c = reader.read() } val maybeLine = buffer.result() if (maybeLine.isEmpty && !warned) { logWarning(...) // unchanged warned = true } maybeLine }.zipWithIndex.collect { case (Some(line), index) => (line, index) }

A NextIterator[Option[String]] also works if relying on takeWhile's one-element look-ahead feels too implicit. I checked that this behaves the same as the current iterator on Scala 2.12 and 2.13, including the look-ahead from lineEntries.hasNext in the JsonParseException handler and exceptions thrown mid-line. Feel free to skip it if you'd rather keep the diff against SPARK-59407 small.

I would keep the current iterator in this fix to limit the backport diff. We can revisit the iterator structure with the reader follow-up.

dongjoon-hyun · 2026-09-28T18:36:26Z

-        val sb = new java.lang.StringBuilder()
-        var overLong = false
+      @tailrec private def fetchLine(): (String, Int) = {
+        val buffer = new BoundedLineBuffer(effectiveMaxLineLength)


For the block-read follow-up: a new BoundedLineBuffer (and a 16-char StringBuilder regrown by doubling) per line, including skipped lines, allocates ~4.3 bytes per input byte on the event logs under core/src/test/resources/spark-events, vs ~1.7 with one buffer per iterator that is reset for each line (Jackson readTree alone is ~3.9). Today the per-char BufferedReader.read() lock hides the CPU cost, but with a bulk reader it becomes ~10% of the read loop. Most of this is inherited from SPARK-59407 (this PR adds ~40 B per line), so it's fine to handle it there. A reused buffer would need a capacity cap so that it doesn't pin a huge array for the rest of the file.

Agreed to evaluate buffer reuse in the reader follow-up, including a retention cap after unusually large lines. I have not independently reproduced the allocation or throughput measurements.

dongjoon-hyun · 2026-09-28T18:36:26Z

    assert(eventMonster.loggedEvents(1) === JsonProtocol.sparkEventToJsonString(applicationEnd))
  }

+  test("Replay line limit uses UTF-8 bytes") {


nit: Since this PR fixes a specific issue and adds tests to an existing suite, could we prefix the new test names with the JIRA ID, e.g. test("SPARK-59804: Replay line limit uses UTF-8 bytes"), as the contributing guide suggests? The same applies to the other new tests (L128, L157, L169, L180, L194, L208, L233, L248, L267). It helps to trace failures across the backport branches.

Added the SPARK-59804 prefix to the new test names.

dongjoon-hyun · 2026-09-28T18:36:26Z

+        "the line ending. Longer lines are skipped with a warning, bounding the " +
+        "memory replay can use when an event log is corrupt or unexpectedly large. Buffer " +
+        "growth, UTF-16 storage and JSON parsing can require several times this limit in heap " +
+        "space. Values at or below 0, or above 536870912, use the maximum supported limit of " +


nit: This hard-codes 536870912 twice, duplicating ReplayListenerBus.MAX_LINE_LENGTH, similar to my earlier comment about DEFAULT_MAX_LINE_LENGTH. We could define the cap in History above EVENT_LOG_MAX_LINE_LENGTH (like DEFAULT_LOG_DIR), interpolate it with .doc(s"..."), and derive ReplayListenerBus.MAX_LINE_LENGTH from it. The other direction (referencing ReplayListenerBus.MAX_LINE_LENGTH from this doc) would create an initialization cycle, because the ReplayListenerBus object already reads History.EVENT_LOG_MAX_LINE_LENGTH.defaultValue during its initialization. monitoring.md would still need a manual update.

The maximum now lives in History, and both the config documentation and ReplayListenerBus derive it from that constant. monitoring.md retains the corresponding documented value.

dongjoon-hyun · 2026-09-28T18:36:26Z

- *                      materialized during replay. Longer lines are drained, skipped and
- *                      logged, bounding the memory replay can use when an event log is
- *                      corrupt or unexpectedly large.
+ * @param maxLineLength Maximum UTF-8 byte length of a single event log line, excluding its line


nit: Could the @param mention the normalization, e.g. "Values <= 0 or above ReplayListenerBus.MAX_LINE_LENGTH (512 MiB) use MAX_LINE_LENGTH"? Otherwise a caller passing 0 (RC1's "disables the limit" semantics) or 1 << 30 silently gets a 512 MiB limit.

The constructor scaladoc now describes the normalization of non-positive and above-maximum values to the 512 MiB limit.

… buffers

viirya · 2026-09-29T07:57:15Z

+        "limits but does not guarantee sufficient heap space. " +
        "Introduced in 4.3.0; also available in 3.5.10, 4.0.5, 4.1.4 and 4.2.1; and in " +
        "all versions after 4.3.0.")
      .version("4.3.0")


Note that I found the backport PRs of SPARK-59407 fills this .version field of this config with the version number in each branch, i.e., .version("4.1.4") on branch-4.1 backport, .version("3.5.10") on branch-3.5 backport.

It makes this config look a bit strange as on each version, this config has different version value on different branch. But what .version specifies should be the version that the config is added.

It seems because this is an improvement ticket and adds a new config, but it backports to many previous branches so the config in earlier versions looks strange.

I'm not sure how to deal with it when to backport this fix to these branches. Just backport this so all branches have "4.3.0" in .version of this config? Or leave the strange version unchanged? 🤔

@dongjoon-hyun @HyukjinKwon What do you think?

I think we have done this with 3.5.10 in this case so far IIRC

Do you mean to set .version("3.5.10") for all branches?

Actually, I'm not sure about that. Both different versions and 3.5.10 sounds misleading to me. However, if there is any precedence, I don't care.

Note that I found the backport PRs of SPARK-59407 fills this .version field of this config with the version number in each branch, i.e., .version("4.1.4") on branch-4.1 backport, .version("3.5.10") on branch-3.5 backport.

cc @holdenk for looping the author of that backport.

If not 3.5.10, what version string we should use in these branches? 4.3.0 as the doc said that it is introduced in 4.3.0?

HeartSaVioR · 2026-09-29T12:04:35Z

Shall we mark the ticket as a blocker (rather than minor) if we intend to block the release? Also does this warrant blocking Apache Spark 4.3.0 as well? If it does, please reply to the mail thread named "Blockers for Apache Spark 4.3.0 RC".

viirya · 2026-09-29T16:33:33Z

Shall we mark the ticket as a blocker (rather than minor) if we intend to block the release? Also does this warrant blocking Apache Spark 4.3.0 as well? If it does, please reply to the mail thread named "Blockers for Apache Spark 4.3.0 RC".

Makes sense. I set this as a blocker. I will reply the mail thread.

viirya · 2026-09-30T07:25:55Z

@HyukjinKwon @dongjoon-hyun Do you have any more comments except for the version of the config? Since this is set as a blocker, we should move this forward soon.

Thanks!

uros-b

Thank you @viirya and @dongjoon-hyun @HyukjinKwon!

uros-b · 2026-09-30T09:55:55Z

Regarding version, I think the convention is the lowest version that ships the config, identical on every branch including master. So 3.5.10, on all branches

viirya · 2026-09-30T17:10:18Z

Regarding version, I think the convention is the lowest version that ships the config, identical on every branch including master. So 3.5.10, on all branches

Actually I prefer not to change it because it is not what this bug fix proposes to fix. And, there seems no consensus on this.

I prefer to focus on the bug in this PR and leave the config version unchanged, so we can unblock the RCs quickly.

dongjoon-hyun · 2026-09-30T17:31:45Z

+1 for @viirya 's decision ("I prefer to focus on the bug in this PR and leave the config version unchanged").

viirya · 2026-09-30T17:34:42Z

Thanks @dongjoon-hyun ! Do you have any more comments on this PR?

…ing event log replay ### What changes were proposed in this pull request? Fix the UTF-8 byte limit and physical-line diagnostics introduced by SPARK-59407 (#58700): - Count event-log content in UTF-8 bytes, excluding line endings, while retaining only a bounded prefix of oversized lines. - Carry physical line indices across skipped lines, and log the source and line without a duplicate stack trace before rethrowing JSON parsing or mapping exceptions. - Reduce the default from `512m` to `256m`, keeping the worst-case buffered character count close to the previous implementation's effective default bound. - Clamp the effective UTF-8 content limit to `512m` in both configuration resolution and the reader constructor, and document the heap overhead. Derive the reader default from the History configuration. - Restore the public JVM no-argument constructor while retaining the Scala default argument. - Release retained prefixes once a line is known to be oversized, identify skipped-line locations, and limit malformed-line excerpts to 1,024 characters. - Document stepped buffer growth and centralize the maximum limit in History. The iterator-based replay API, strict UTF-8 decoding, existing truncated-last-line handling and ordinary input-stream IOException behavior are preserved. Independent follow-ups will address the unbounded compaction-filter and tail-reparse paths, tail-relative diagnostics and UTF-8 alignment, incomplete UTF-8 at EOF, bulk-read performance, and the pre-existing end-to-end test assertions. JIRA: https://issues.apache.org/jira/browse/SPARK-59804 Release coordination: 4.2.1 RC1 contains the previous implementation. This has been raised on the Spark dev mailing list to discuss an RC2 containing the reviewed fix; no migration entry or final release-version decision is made in this PR pending that discussion. ### Why are the changes needed? The original reader compares `StringBuilder.length() * 2` with a byte-valued setting. With an 8 KiB limit, it incorrectly skips a valid ASCII application-start event with a 6,000-character application name. UTF-8 multibyte characters and CRLF endings also produce inconsistent boundaries. Dropping oversized lines before `zipWithIndex` renumbers subsequent diagnostics. For example, a valid event on line 1, oversized lines 2 and 3, and an invalid event on line 4 report line 2. JSON processing exceptions also need an explicit diagnostic before they are rethrown to retain that physical location. Correcting the unit without adjusting the default permits roughly twice as many buffered characters. With a lazy line containing U+0101 followed by 600,000,000 ASCII characters, the UTF-8 reader with a 512 MiB limit runs out of memory in a separate 1 GiB heap JVM; the revised 256 MiB default successfully drains and skips that input. Buffer growth, UTF-16 storage and JSON parsing still require heap beyond the configured content limit. A separate `-Xmx3g` JVM also reproduces a JVM array-size failure with `600m` and 620,000,000 ASCII characters followed by U+0100, despite the line fitting within that configured limit. StringBuilder capacity growth followed by UTF-16 inflation fails independently of available heap. Clamping the effective content limit to `512m` successfully skips that input; it does not guarantee adequate heap for every workload. ### Does this PR introduce _any_ user-facing change? Yes. Events at or below the effective UTF-8 content limit are replayed, LF/CRLF endings do not consume the limit, and parsing diagnostics identify physical lines after skipped content. The default is reduced from `512m` to `256m` to keep its maximum buffered character count close to the previous effective bound. Non-positive values and values above `512m` now resolve to an effective maximum of 536,870,912 bytes (512 MiB), including direct constructor arguments. This clamps the content limit itself, so oversized lines are skipped instead of replayed as truncated prefixes. Smaller positive limits are unchanged. ### How was this patch tested? ``` build/sbt 'core/testOnly *ReplayListenerSuite' build/sbt 'core/Compile/scalastyle' 'core/Test/scalastyle' ``` All 19 suite tests and both scalastyle checks pass. Coverage includes the 8 KiB ASCII configuration case, raw 1- through 4-byte UTF-8 input with length and read-buffer boundary splits, line endings, directly observed buffer bounds, configuration normalization, event delivery through normalized constructors, reflective no-argument construction, buffer release, skipped-line locations, bounded malformed-line diagnostics, physical-line diagnostics without an attached throwable, JSON parsing and mapping exceptions, filtered iterators, truncation, and malformed UTF-8. The suite also runs its existing compressed/end-to-end tests, but their pre-existing assertion gaps mean they do not establish full replayed-event content equality; those test fixes remain a separate follow-up. Targeted mutation checks confirmed that removing or doubling the buffer cap, miscounting surrogate pairs, or removing JSON line diagnostics each makes its corresponding test fail. The oversized lazy-stream scenarios were also exercised in separate `-Xmx1g` and `-Xmx3g` JVMs as described above. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: OpenAI Codex (GPT-6) Closes #59074 from viirya/codex/replay-line-limits. Authored-by: Liang-Chi Hsieh <viirya@gmail.com> Signed-off-by: Liang-Chi Hsieh <viirya@gmail.com> (cherry picked from commit 719947a) Signed-off-by: Liang-Chi Hsieh <viirya@gmail.com>

…ing event log replay Fix the UTF-8 byte limit and physical-line diagnostics introduced by SPARK-59407 (#58700): - Count event-log content in UTF-8 bytes, excluding line endings, while retaining only a bounded prefix of oversized lines. - Carry physical line indices across skipped lines, and log the source and line without a duplicate stack trace before rethrowing JSON parsing or mapping exceptions. - Reduce the default from `512m` to `256m`, keeping the worst-case buffered character count close to the previous implementation's effective default bound. - Clamp the effective UTF-8 content limit to `512m` in both configuration resolution and the reader constructor, and document the heap overhead. Derive the reader default from the History configuration. - Restore the public JVM no-argument constructor while retaining the Scala default argument. - Release retained prefixes once a line is known to be oversized, identify skipped-line locations, and limit malformed-line excerpts to 1,024 characters. - Document stepped buffer growth and centralize the maximum limit in History. The iterator-based replay API, strict UTF-8 decoding, existing truncated-last-line handling and ordinary input-stream IOException behavior are preserved. Independent follow-ups will address the unbounded compaction-filter and tail-reparse paths, tail-relative diagnostics and UTF-8 alignment, incomplete UTF-8 at EOF, bulk-read performance, and the pre-existing end-to-end test assertions. JIRA: https://issues.apache.org/jira/browse/SPARK-59804 Release coordination: 4.2.1 RC1 contains the previous implementation. This has been raised on the Spark dev mailing list to discuss an RC2 containing the reviewed fix; no migration entry or final release-version decision is made in this PR pending that discussion. The original reader compares `StringBuilder.length() * 2` with a byte-valued setting. With an 8 KiB limit, it incorrectly skips a valid ASCII application-start event with a 6,000-character application name. UTF-8 multibyte characters and CRLF endings also produce inconsistent boundaries. Dropping oversized lines before `zipWithIndex` renumbers subsequent diagnostics. For example, a valid event on line 1, oversized lines 2 and 3, and an invalid event on line 4 report line 2. JSON processing exceptions also need an explicit diagnostic before they are rethrown to retain that physical location. Correcting the unit without adjusting the default permits roughly twice as many buffered characters. With a lazy line containing U+0101 followed by 600,000,000 ASCII characters, the UTF-8 reader with a 512 MiB limit runs out of memory in a separate 1 GiB heap JVM; the revised 256 MiB default successfully drains and skips that input. Buffer growth, UTF-16 storage and JSON parsing still require heap beyond the configured content limit. A separate `-Xmx3g` JVM also reproduces a JVM array-size failure with `600m` and 620,000,000 ASCII characters followed by U+0100, despite the line fitting within that configured limit. StringBuilder capacity growth followed by UTF-16 inflation fails independently of available heap. Clamping the effective content limit to `512m` successfully skips that input; it does not guarantee adequate heap for every workload. Yes. Events at or below the effective UTF-8 content limit are replayed, LF/CRLF endings do not consume the limit, and parsing diagnostics identify physical lines after skipped content. The default is reduced from `512m` to `256m` to keep its maximum buffered character count close to the previous effective bound. Non-positive values and values above `512m` now resolve to an effective maximum of 536,870,912 bytes (512 MiB), including direct constructor arguments. This clamps the content limit itself, so oversized lines are skipped instead of replayed as truncated prefixes. Smaller positive limits are unchanged. ``` build/sbt 'core/testOnly *ReplayListenerSuite' build/sbt 'core/Compile/scalastyle' 'core/Test/scalastyle' ``` All 19 suite tests and both scalastyle checks pass. Coverage includes the 8 KiB ASCII configuration case, raw 1- through 4-byte UTF-8 input with length and read-buffer boundary splits, line endings, directly observed buffer bounds, configuration normalization, event delivery through normalized constructors, reflective no-argument construction, buffer release, skipped-line locations, bounded malformed-line diagnostics, physical-line diagnostics without an attached throwable, JSON parsing and mapping exceptions, filtered iterators, truncation, and malformed UTF-8. The suite also runs its existing compressed/end-to-end tests, but their pre-existing assertion gaps mean they do not establish full replayed-event content equality; those test fixes remain a separate follow-up. Targeted mutation checks confirmed that removing or doubling the buffer cap, miscounting surrogate pairs, or removing JSON line diagnostics each makes its corresponding test fail. The oversized lazy-stream scenarios were also exercised in separate `-Xmx1g` and `-Xmx3g` JVMs as described above. Generated-by: OpenAI Codex (GPT-6) Closes #59074 from viirya/codex/replay-line-limits. Authored-by: Liang-Chi Hsieh <viirya@gmail.com> Signed-off-by: Liang-Chi Hsieh <viirya@gmail.com> (cherry picked from commit 719947a) Signed-off-by: Liang-Chi Hsieh <viirya@gmail.com>

viirya · 2026-09-30T18:28:16Z

Merge Summary:

merged into master 719947a
merged into branch-4.x 52c7fa1
merged into branch-4.3 cd32ffa
merged into branch-4.2 eabf40f
merged into branch-4.1 b4799fd
merged into branch-4.0 d99a2d9

Posted by merge_spark_pr.py

viirya · 2026-09-30T18:29:11Z

Thanks @dongjoon-hyun @HyukjinKwon @uros-b @HeartSaVioR !

Only branch-3.5 requires a manual backport. I will create a backport PR.

…s during event log replay ### What changes were proposed in this pull request? Backport #59074 (SPARK-59804, master commit 719947a) to `branch-3.5`. Adapt the replay diagnostics to the string-based logging API in Spark 3.5, preserving the messages and exception behavior. Add explicit filter parameter types where Scala 2.12 cannot infer them for overloaded replay calls. Retain the branch's `3.5.10` configuration version. The backport includes UTF-8 byte accounting, physical-line diagnostics, the 256 MiB default and 512 MiB maximum, the restored no-argument constructor, early release of skipped-line buffers, bounded malformed-line excerpts, documentation, and regression tests. Independent follow-ups from the original PR remain outside this backport. ### Why are the changes needed? `branch-3.5` contains SPARK-59407 (8defa3c), so it has the same byte-limit, diagnostic, and constructor regressions fixed by #59074. The master patch needs logging API adaptation on this branch. ### Does this PR introduce _any_ user-facing change? Yes, the same fixes as #59074: limits measure UTF-8 content bytes excluding line endings; the default is 256 MiB; non-positive and above-maximum settings resolve to 512 MiB. Replay preserves physical line numbers across skipped lines, bounds malformed-line excerpts, and restores the JVM no-argument constructor. ### How was this patch tested? On branch-3.5 with Java 17.0.16 and Scala 2.12.18, using the company Maven proxy: ``` build/sbt 'core/testOnly *ReplayListenerSuite' build/sbt 'core/Compile/scalastyle' 'core/Test/scalastyle' ``` All 19 tests and both scalastyle checks passed. The tests cover UTF-8 boundaries, normalized constructor replay, reflective no-argument construction, buffer release, physical-line diagnostics, bounded error excerpts, and IO exception handling. Existing compressed/end-to-end tests retain the assertion limitations documented in #59074; their independent follow-up remains outside this backport. Also checked the compiled class with `javap`: both constructors and the Scala default-argument method are present. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: OpenAI Codex (GPT-6) Closes #59166 from viirya/codex/replay-line-limits-35. Authored-by: Liang-Chi Hsieh <viirya@gmail.com> 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 #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>

…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>

…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-59804][UI] Fix UTF-8 byte limits and physical line numbers dur…

1df219a

…ing event log replay

dongjoon-hyun reviewed Sep 27, 2026

View reviewed changes

[SPARK-59804][UI] Address replay limit and diagnostics review

ab7c85d

HyukjinKwon reviewed Sep 28, 2026

View reviewed changes

[SPARK-59804][UI] Clarify optional replay line naming

aaee04e

dongjoon-hyun reviewed Sep 28, 2026

View reviewed changes

[SPARK-59804][UI] Cap replay line limits and avoid duplicate stack tr…

691a1d7

…aces

dongjoon-hyun reviewed Sep 28, 2026

View reviewed changes

[SPARK-59804][UI] Restore replay constructor and release skipped line…

1167bfa

… buffers

viirya commented Sep 29, 2026

View reviewed changes

HyukjinKwon approved these changes Sep 30, 2026

View reviewed changes

uros-b reviewed Sep 30, 2026

View reviewed changes

dongjoon-hyun approved these changes Sep 30, 2026

View reviewed changes

viirya closed this in 719947a Sep 30, 2026

viirya deleted the codex/replay-line-limits branch September 30, 2026 18:29

viirya mentioned this pull request Sep 30, 2026

[SPARK-59804][3.5][UI] Fix UTF-8 byte limits and physical line numbers during event log replay #59166

Closed

This was referenced Oct 9, 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

[SPARK-60158][UI] Avoid per-character synchronized reads in the bounded event log line reader #59360

Closed

Conversation

viirya commented Sep 27, 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!

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!

HyukjinKwon 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!

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!

viirya commented Sep 27, 2026 •

edited

Loading