SteNicholas opened a new pull request, #3718: URL: https://github.com/apache/celeborn/pull/3718
### What changes were proposed in this pull request? This PR extends the end-to-end shuffle integrity checks introduced in CELEBORN-894 (Spark-only) to Flink workloads, covering both the regular and the tiered (hybrid) read paths. When the check is enabled, the write side records a per subpartition CRC32 + byte count and the driver validates it against what the reader actually consumed, failing the read on a mismatch. - **Write side**: `FlinkShuffleClientImpl` hashes each push payload (the body after the batch header) into `PushState` via a zero-copy `ByteBuffer` view and reports the per-subpartition CRC32/bytes at `MapperEnd`, reusing the existing `crc32PerPartition` / `bytesWrittenPerPartition` plumbing. The constructor fails fast if the write-side `BATCH_HEADER_SIZE` ever diverges from the read-side `BufferUtils.HEADER_LENGTH_PREFIX`. - **Read side**: `RemoteBufferStreamReader` and `CelebornChannelBufferReader` accumulate the read CRC32/bytes through a shared `ReadIntegrityTracker` and report them at the last partition's stream end. The tracker owns the per-path framing/stripping and disables itself on any unexpected buffer shape (wrong component count, or a buffer shorter than the batch header) rather than risk a false mismatch. - **Driver side**: `ReadReducerPartitionEnd` is reused for MAP partitions, and `MapPartitionCommitHandler.finishPartition` combines the recorded write-side checksums over the consumed subpartition range, failing closed on a mismatch or missing metadata. - Add zero-copy `ByteBuffer` overloads to `CelebornCRC32`, `CommitMetadata` and `PushState`. - Minor: drop a stray (cosmetic, no-op) unary plus in `handleReducerPartitionEnd`'s failure branch and update the client config doc. ### Why are the changes needed? CELEBORN-894 added end-to-end integrity verification only for Spark. Flink workloads — including hybrid/tiered shuffle — had no equivalent guard, so silent shuffle data corruption (bit flips, truncation, mis-framing) could go undetected and surface as wrong results rather than a failed task. This PR brings the same write-vs-read checksum/byte-count validation to Flink so such corruption fails the read instead of being silently consumed. ### Does this PR resolve a correctness bug? - [ ] Yes ### Does this PR introduce _any_ user-facing change? - [x] Yes The existing `celeborn.client.shuffle.integrityCheck.enabled` config now also applies to Flink (previously Spark-only); its documentation is updated accordingly. The default remains `false`, so there is no behavior change unless the check is explicitly enabled. ### How was this patch tested? Added unit and integration tests: - `CelebornCRC32Test` / `CommitMetadataTest`: the new `ByteBuffer` overloads (single and split header/data), order-independence, and corruption / byte-count-mismatch detection. - `MapPartitionCommitHandlerTest`: `finishPartition` success and all failure branches (no metadata, missing map partition, out-of-bounds range, checksum mismatch, byte-count mismatch), concurrent recording, and the expired-shuffle race. - `ReadIntegrityTrackerTest`: report-once / disable semantics and per-path framing for both the regular and tiered read paths. - `RemoteBufferStreamReaderTest`: the stream-end-after-close race (a failed report must not notify the failure listener on a closed channel). - `CelebornBufferStreamTest`: the `hasRemainingPartitions` location-index boundary. - `WordCountTest` (`WordCountTestWithIntegrityCheck`) and `HybridShuffleWordCountTest`: end-to-end Flink runs with the check enabled, on both the regular and hybrid shuffle paths. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
