SteNicholas commented on PR #3670:
URL: https://github.com/apache/celeborn/pull/3670#issuecomment-4490883452
@xumingming, please take a look at the comments of claude code:
```
Code Review — [CELEBORN-2313] Extend E2E checked zone to batch assembly
point
(apache/celeborn#3670)
Overview
Moves CRC_M (per-partition checksum + byte-count) accumulation out of
ShuffleClientImpl.pushOrMergeData() (which runs on the async DataPusher
thread) into a new
explicit ShuffleClient.computeBatchCRC() invoked synchronously in the
writer thread, right after
batch assembly, at 7 dispatch sites (Hash spark-2/3: flushSendBuffer,
pushGiantRecord,
close()/closeWrite() residual; Sort pushGiantRecord; SortBasedPusher:
partition-change,
buffer-overflow, final flush). The old accumulation line in
pushOrMergeData() is removed;
ShuffleClient/DummyShuffleClient get the new method. +103/-6, 9 files.
Goal: widen the
integrity-checked zone to cover the assembly→async-dispatch window.
Correctness — the central concern is call-site completeness
- The single chokepoint becomes an invariant maintained by convention.
Previously every push
passed through pushOrMergeData(), so CRC_M provably covered exactly what
was sent. Now CRC_M
correctness depends on enumerating every assembled-data dispatch site
across 4 writers + the
pusher. A missed site → under-count → false integrity-mismatch task
failures; an extra/duplicated
site or a batch CRC'd then not sent (discard/abort) → over-count → same.
This is the dominant risk
and is asserted by inspection only. The enumerated sites look plausibly
complete (normal flush,
giant record, close residual, sort partition-change/overflow/final; Sort
normal path is covered
via the shared SortBasedPusher), but please double-check there is no other
path that reaches the
server with assembled user data (e.g., revive/split re-push reuses an
already-CRC'd buffer = OK;
confirm nothing else calls pushData/mergeData with fresh data).
- ⚠️ attemptId encoding consistency. Hash writers pass encodedAttemptId;
SortBasedPusher passes
attemptNumber. computeBatchCRC does getPushState(makeMapKey(shuffleId,
mapId, attemptId)). If that
key doesn't match the key used by the actual push/commit path for the
same batch, CRC_M is
accumulated into a different PushState than the one read at commit →
integrity check silently
broken or always-mismatching. Verify both encodings resolve to the same
PushState (high priority).
- ⚠️ Thread-safety / visibility. Accumulation moves from the DataPusher
thread to the writer
thread, while PushState.addDataWithOffsetAndLength was previously
single-threaded on the pusher.
Confirm: (a) SortBasedPusher.pushData can't run concurrently with another
path mutating the same
PushState (e.g., spill-triggered flush), and (b) there is a happens-before
between writer-thread
accumulation and the commit-time getCRC32PerPartition read (likely OK
since the writer also drives
mapperEnd, but it should be explicit/confirmed).
- Buffer ordering is correct: computeBatchCRC is placed immediately before
addTask/mergeData/pushData at every site, so the snapshot reflects the
bytes about to be
dispatched (and any later mutation is exactly what this PR intends to
catch). Partition/offset
args line up with the following dispatch call at each site.
- CRC is still over uncompressed assembled bytes (same content the old
code saw before the
compression block), so semantics vs. the reducer-side check are preserved.
Code quality / style
- 7 duplicated computeBatchCRC(shuffleId, mapId, <attempt>, partition,
buf, 0, size) calls are
fragile — a future new dispatch path will silently omit it. Consider
funnelling writer-side
dispatch through a small helper (crcThenAddTask(...) /
crcThenPushOrMerge(...)) so the CRC and the
dispatch can't drift apart, restoring much of the lost single-point
property without giving up
the wider zone.
- Good: the ShuffleClient.computeBatchCRC Javadoc states the invariant
("sole CRC accumulation
path; pushOrMergeData does not compute CRC"). Keep that — it's the
contract the whole change rests
on.
- The PR description is honest about the elegance trade-off; appreciated.
Worth adding an inline
comment at each site (or near pushOrMergeData) pointing to the enumerated
sites so the invariant
is discoverable.
- Disabled path is correctly early-returned (if
(!shuffleIntegrityCheckEnabled) return;), so
default-off users pay nothing.
Performance
- CRC32 now runs on the synchronous writer critical path instead of
overlapping on the DataPusher
thread. For large shuffles with integrity check enabled this removes
pipelining and may measurably
reduce write throughput. It's gated by shuffleIntegrityCheckEnabled
(default off → no impact),
but for users who enable it this is a real regression that should be
benchmarked and called out as
a conscious trade-off.
Test coverage — thin for the risk profile
- testComputeBatchCRCAccumulatesCorrectly is a solid unit test for the
primitive: multi-batch,
multi-partition accumulation verified against an independent
CommitMetadata recomputation,
including per-partition separation. Good.
- Missing the high-risk coverage: (1) no test that each writer path (giant
/ normal flush / close
residual / sort overflow/partition-change/final) actually calls
computeBatchCRC exactly once for
the bytes it sends; (2) no end-to-end assertion that writer-side CRC_M
equals a recomputation over
the actually pushed bytes (the regression that a moved/missed site would
introduce); (3) no test
for the attemptId-keying consistency; (4) no disabled-path test. Recommend
at least a writer-level
test (Hash and Sort, integrity enabled) asserting accumulated
per-partition CRC matches
independent recomputation across the giant/normal/close paths.
Security
- N/A — internal integrity mechanism, no new external input handling.
Verdict
Reasonable, well-motivated correctness enhancement with a clean primitive
and honest
documentation. No definitive bug found, but it trades a structural
guarantee for a convention
spread over 7 sites. Before merge I'd want: (1) confirmation of
attemptId/PushState key
consistency (esp. SortBasedPusher), (2) confirmation of call-site
completeness and PushState
thread-safety/visibility, (3) writer-level/E2E tests that would fail if a
site is missed or
mis-keyed, and (4) a benchmark note on the writer-thread CRC cost.
Consider the dispatch-helper
refactor to harden the invariant.
```
--
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]
