[VL] Fix false-EOF probe in ColumnarCachedBatchSerializer.read trailing markers

[yaooqinn]

What changes were proposed in this pull request?

ColumnarCachedBatchSerializer.read used input.available() > 0 as
an EOF probe before reading the trailing hasStats / hasSchema
boolean markers. This PR replaces the probe with a direct readBoolean()
inside a narrow try/catch on KryoException whose message starts
with "Buffer underflow", which is the actual signal Kryo emits at
true end-of-input.

The V1 wire back-compat behavior (absent trailing booleans read as
null, not throw) is preserved — and is locked by the existing test
ColumnarCachedBatchKryoSuite#"V1 wire (no trailing hasStats/hasSchema booleans) reads as stats=null/schema=null".

Why are the changes needed?

Kryo Input.available() returns
(limit - position) + underlyingStream.available(). The JDK
InputStream.available() contract permits any implementation to
return 0 even when more data follows — BufferedInputStream over
shuffle-spill / network chunk boundaries routinely does so. When the
Kryo buffer is drained AND the underlying stream reports 0 at the
trailing-boolean byte position, the probe falsely concludes EOF,
skips hasStats, and the next readClassAndObject interprets the
stats payload (which contains the schema JSON) as a class name.

Production stack we observed (truncated):

ClassNotFoundException: {"type":"struct","fields":[...
  at DefaultClassResolver.readName(DefaultClassResolver.java:154)
  at DefaultClassResolver.readClass(DefaultClassResolver.java:133)
  at Kryo.readClass(Kryo.java:693)
  at Kryo.readClassAndObject(Kryo.java:804)
  at ColumnarCachedBatchSerializer$$anon$2.hasNext(...)

The length-bound require(... <= maxLen ...) guard from
491070bf34 is preserved — it remains useful against
NegativeArraySizeException / oversized allocation and is
orthogonal to the EOF-probe fix.

Why catch KryoException by message prefix?

Kryo 4.x (the version on Gluten's classpath via kryo-shaded) has
no dedicated KryoBufferUnderflowException subclass — that was added
in Kryo 5.x. The message-prefix check is the narrowest filter we can
apply on 4.x without swallowing real corruption (e.g. a
ClassNotFoundException wrapped during readClassAndObject, which
does not start with "Buffer underflow").

Does this PR introduce any user-facing change?

No (bug fix in a build that has not shipped to any tagged release).

How was this patch tested?

New ColumnarCachedBatchKryoBoundaryProbeBugSuite:

• "multi-batch deserialize survives boundary-aligned trailing-boolean probe"
  — deterministic repro of the production stack using a
  1-byte-per-read InputStream that returns available() == 0 at
  the trailing-boolean position. Fails on the pre-patch code, passes
  after the fix.

Existing ColumnarCachedBatchKryoSuite#"V1 wire ..." continues to
pass, locking the V1 silent-null back-compat contract.

Verified locally on -Pspark-4.1 -Pscala-2.13; affected suites
(Framed, BuildFilter, StatsBlob, ShipBlockerMarshal,
IntFamilyMarshal) all green (23/23). CI 60/60 green on
da460551e7.

---

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

[liuneng1994]

PR description contradicts the code

Cross-referencing the existing locked test in ColumnarCachedBatchKryoSuite.scala:

test("V1 wire (no trailing hasStats/hasSchema booleans) reads as stats=null/schema=null") {
  ...
  assert(read.stats === null, "absent trailing hasStats must read as null, not throw")
  assert(read.schema === null, "absent trailing hasSchema must read as null, not throw")
}

This contract requires V1 compat to be preserved — and the new try/catch (KryoException if isBufferUnderflow) => false is exactly what satisfies it. But the PR description says the opposite:

description claims	code / tests actually
"V1 wire era … never released … guard protects nothing"	try/catch preserves V1 compat; locked by existing "V1 wire …" test
"A truncated stream now surfaces KryoException rather than silently returning null"	truncated stream is caught and returns null stats — same silent behavior
Cites BoundaryProbeBugSuite#"truncated wire … fails fast, not silently"	new file only contains 1 test ("multi-batch deserialize survives …") — the cited test doesn't exist

Suggested fix: keep the code as-is (preserving V1 compat is correct), and trim the description down to honestly what the patch does — "fix the false-EOF probe in Input.available() so the trailing booleans are read across chunk boundaries; V1 compat is now achieved via try/catch on KryoException instead". Drop the "Spark cached blocks live and die within a single application instance" paragraph and the cite to a non-existent test.