SteNicholas commented on PR #3689:
URL: https://github.com/apache/celeborn/pull/3689#issuecomment-4627674900
## Review — Encryption at Rest (Spark)
Nice work. The wiring is clean and the integration points are well chosen. I
traced the full write→store→read path and the design is sound. A few notes
below, mostly non-blocking suggestions and a couple of questions.
### What works well
- **Crypto ordering is correct**: compress-then-encrypt on push
(`ShuffleClientImpl.pushOrMergeData`) and decrypt-then-decompress on read
(`CelebornInputStream.fillBuffer`). Encrypting after compression preserves
compressibility, and the read path is symmetric.
- **Composes correctly with the end-to-end integrity check**: the write-side
checksum is computed over the raw uncompressed/unencrypted data
(`ShuffleClientImpl.java:1049-1051`) and the read-side over the
decrypted+decompressed data (`CelebornInputStream.java:870-871`). Both sides
checksum the same plaintext, so EAR +
`celeborn.client.shuffle.integrityCheck.enabled` work together.
- **Only the batch payload is encrypted** — the 16-byte batch header
(mapId/attemptId/batchId/length) stays plaintext, so worker-side
routing/sorting and dedup are unaffected.
- **Backward-compatible surface**: the 6-arg `ShuffleClient.get` overload is
preserved (MR/Tez/Flink callers unaffected); signature changes are confined to
internal `create()`/reader constructors with `Optional.empty()` defaults.
Confirmed the only `ShuffleClient` concrete subclasses are `ShuffleClientImpl`
and `DummyShuffleClient`, both updated.
- **Fresh IV per batch** via `CryptoStreamUtils` → semantic security across
batches.
### Suggestions / questions
1. **No end-to-end test of the actual wiring** (test coverage).
`SparkCryptoHandlerSuiteJ` covers the handler in isolation, but every
reader/writer test still passes `Optional.empty()`. The encrypt-on-push →
decrypt-on-read path — and crucially its interaction with compression and the
integrity check — has no automated coverage and relies on "tested in production
internally." A round-trip test through
`ShuffleClientImpl`/`CelebornInputStream` with a real handler (compression
on/off × integrity-check on/off) would protect this from regressions.
2. **Per-batch allocation churn on the hot path** (perf). In `fillBuffer`,
the crypto branch reassigns `compressedBuf = decrypted` / `rawDataBuf =
decrypted` to the fresh array returned by `decrypt()` on every batch,
discarding the buffers pre-allocated in `init()`. Combined with
`decrypt()`/`encrypt()` each allocating a `ByteArray{Input,Output}Stream` + a
new array per call, this adds steady allocation pressure per batch. Consider a
decrypt/encrypt API that writes into a reusable caller-provided buffer.
3. **PR description says "no user-facing change"** — this isn't quite
accurate. Users who already run with `spark.io.encryption.enabled=true` will
now have their Celeborn shuffle data encrypted (added CPU + ~IV/length overhead
per batch) where it previously was not. Worth calling out explicitly, and a
docs/config page for the feature would help (the design is in a Google Doc
that's not accessible to ASF reviewers — could it be folded into the CIP-22
doc?).
4. **Confidentiality vs. integrity** (informational). Spark's
`CryptoStreamUtils` default is AES/CTR, which is unauthenticated — EAR provides
confidentiality but not tamper-detection on its own, and the integrity check
that *would* catch tampering defaults to `false`. This matches Spark's own IO
encryption, so it's a reasonable choice, but a doc note (confidentiality ≠
integrity; pair with the integrity check where tamper-resistance matters) would
set expectations. The intentionally-loose assertion in
`testDecryptWithWrongKeyFails` reflects exactly this property.
5. **`commons-crypto` dependency placement** (question). It's added as a
compile dependency to `client/pom.xml`, but the `client` module itself doesn't
reference it — only `client-spark/common`'s `SparkCryptoHandler` uses
`CryptoStreamUtils` (which pulls `commons-crypto` transitively from
spark-core). Is the explicit `client` dependency there to feed the shaded
`<include>`s? If so, would declaring it in `client-spark/common` (where it's
actually used) be cleaner?
6. **`ShuffleClientImpl.cryptoHandler` is non-volatile** (minor). It's set
in `setupCryptoHandler` (under the `get()` synchronized block) but read on task
threads in `pushOrMergeData`/`readPartition` without synchronization. This
mirrors the existing `extension` field, so it's consistent — just flagging
`volatile` as a clarity improvement since reads happen post-init outside the
lock.
7. **`SparkCryptoHandler.encrypt` writes the plaintext `length` as a
cleartext int prefix** (minor). It's redundant (you could read to EOF on
decrypt) and leaks the exact plaintext size in the clear, though the ciphertext
length already approximates it. Fine to keep — just noting.
Overall this looks solid and well-scoped. The main thing I'd push on before
merge is item (1) — an integration test for the round-trip wiring.
--
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]
