Fix DistributionData percentile sketch data race (AIOOBE) and partial-read deserialization bug

[Zhangyx39]

Summary

DistributionData (LI Beam fork) holds a single mutable, non-thread-safe DataSketches UpdateDoublesSketch and mutated it in place with no synchronization. The per-record writer (DistributionCell.update on the task/mailbox thread) races concurrent readers:

• percentiles() / extractResult() during metric extraction (the AIOOBE site),
• writeObject() during accumulator serialization (TM→JM),
• combine(DistributionData) during JobManager accumulator merge.

A read (getQuantiles, which sizes a destination array from getN() then copies) concurrent with a write (update, which grows the sketch) overruns the array and throws:

java.lang.ArrayIndexOutOfBoundsException: Index N out of bounds for length N
    at org.apache.datasketches.quantiles.DoublesAuxiliary.populateFromDoublesSketch(...)
    at org.apache.datasketches.quantiles.DoublesSketch.getQuantiles(...)

DataSketches UpdateDoublesSketch is documented single-threaded, so this is a caller-side concurrency bug, independent of the datasketches version.

Fix — snapshot-under-lock

• Add a private final ReentrantLock lock guarding all sketch/primitive access. ReentrantLock (rather than synchronized on a field) is used because it is final + Serializable: it survives the TM→JM accumulator round-trip and always deserializes unlocked, so no transient re-init is needed. (Also satisfies ErrorProne SynchronizeOnNonFinalField under -Werror.)
• Writers guarded (O(1)/O(sketch)): update, combine(long), combine(long,long,long,long), combine(DistributionData), reset.
• Reader percentiles(): copies the bounded (k=256) sketch under the lock, then runs the expensive O(n) getQuantiles on the immutable snapshot outside the lock — so per-record writers never block on the quantile computation.
• combine(DistributionData): snapshots other under its own lock first (copying sketch bytes, not aliasing) → deadlock-free, no shared mutable sketch.
• writeObject guarded; sum/count/min/max getters guarded for consistent snapshots.

Additional fix (tightly coupled)

readObject used InputStream.read(byte[]) (reads up to len bytes), which returned a partial sketch image when the array spanned the stream buffer — corrupting the deserialized sketch and yielding wrong percentiles after the TM→JM transfer. Switched to readFully. This was caught by the new serialization parity test.

Testing

• Concurrency stress test (writers + readers + serializers): reproduces the exact ArrayIndexOutOfBoundsException (DoublesAuxiliary.populateFromDoublesSketch → getQuantiles) on the pre-fix code and passes with the fix — the regression guard.
• Serialization round-trip + populated/empty percentile + combine parity tests.
• All 8 DistributionDataTest tests pass; Spotless/ErrorProne clean. Pre-existing unrelated failures in DistributionCellTest/MetricsContainerImplTest were confirmed present on the untouched baseline.

Performance

Write path adds one uncontended monitor acquire (nanoseconds, lock-elidable). Read/serialize path holds the lock only for a bounded few-KB toByteArray() copy, not the O(n) getQuantiles. No measurable throughput regression expected — the fix removes a crash, not adds a bottleneck.