ewoodbury opened a new pull request, #3879:
URL: https://github.com/apache/celeborn/pull/3879
### What changes were proposed in this pull request?
Move DFS replica dedup from the worker (at close) to the driver (after
commit):
- Worker: `DfsTierWriter.closeStreams` no longer deletes its copy when the
peer's `.success` exists, so both copies are always reported in `CommitFiles`.
Remove the now-unused `DiskFileInfo.getDfsPeerWriterSuccessPath` and
`Utils.getPeerPath`.
- Worker: fail the commit when a DFS copy's data stream or index cannot be
closed, instead of reporting an unreadable copy as committed. `.success` is
written last, and failing to create it is only logged.
- Worker: in `Controller.commitFiles`, treat a non-empty file with no
storage info as failed. DFS writers no longer return null here; this is a
safety check, since an id missing from every list is what caused this bug.
- Driver: after `checkDataLost` passes, leave out of the published file
group any replica whose primary also committed, with both copies on HDFS, and
queue `DestroyWorkerSlots` for it on a separate pool in `LifecycleManager`.
Applies to `ReducePartitionCommitHandler` and `MapPartitionCommitHandler`. If
either side failed, nothing is removed.
### Why are the changes needed?
With DFS storage and `celeborn.client.push.replicate.enabled=true`, the
worker that closes a partition second sees the peer's `.success`, deletes its
own copy, and leaves the id out of its `CommitFiles` response. If the peer's
commit then times out, the peer reports every id failed. `checkDataLost` only
flags an id that failed on both sides, so the partition gets no location and
reducers silently read nothing.
A peer's `.success` only means the peer wrote its file, not that its commit
succeeded; only the driver sees both results. The dedup came from ISSUE-440
(#466). `closeStreams` also runs after a failed flush, so a failed copy writes
`.success` and its healthy peer deletes itself, which is a second way to hit
the same loss.
The docs recommend replication off with HDFS, but users enable it, S3/OSS
support is new in 0.6.0, the loss is silent, and
`celeborn.client.shuffle.integrityCheck.enabled` is off by default.
Notes for reviewers:
- Storage: HDFS keeps one copy after commit, as today; the driver queues the
release right after commit, and a failed release only keeps the replica until
shuffle expiry. With `MEMORY,HDFS`, the release depends on CELEBORN-2485
(`StorageManager.removeFileInfo` skips DFS files when the shuffle has memory
files). S3/OSS keep the same objects as today: the old delete ran before
`completeMultipartUpload`, so it only removed the empty placeholder. Each
replica now also gets an index and `.success`.
- Reads: `DfsPartitionReader` opens streams through the location's worker,
so after an HDFS replica is released, losing the primary worker means a fetch
failure and stage rerun, as with today's dedup. S3/OSS now publish both copies,
so reads can fail over.
- Rolling upgrade: upgrade workers first. New workers with an old client
keep both HDFS copies until expiry. Old workers still drop a copy at close, so
a new client never releases anything for them.
- CELEBORN-2341 touches the same commit timeout path; replying with the
partial result narrows the window but does not close it.
### Does this PR resolve a correctness bug?
- [x] Yes
Fixes silent shuffle data loss with DFS storage and replication.
### Does this PR introduce _any_ user-facing change?
- [x] Yes
Deleting the second HDFS copy moves from the worker to the driver, S3/OSS
now keep and publish both copies, and a DFS copy whose index cannot be written
now fails the commit. No new configuration.
### How was this patch tested?
- `HdfsReplicaReleaseClusterSuite` (mini cluster, HDFS simulated with
`RawLocalFileSystem`):
- The primary's commit times out (`COMMIT_FILE_EXCEPTION`); the partition
is published from the replica and reads back intact. On `main` this fails with
an empty file group.
- When both copies commit, the replica is left out of the file group and
its files are deleted. This also passes on `main`; it checks that HDFS storage
behavior is unchanged.
- `TierWriterSuite`: the writer keeps its copy when the peer `.success`
exists; fails close when its index cannot be written; still commits when only
`.success` fails; and reports the original flush failure when the index also
fails. The first three fail on `main`.
- `HdfsReplicaReleaseSuite`: replica selection across storage types, failed
sides, missing storage info, and unallocated workers.
- Client and worker module tests, `spotless:check`, and Scala 2.13
(`-Pspark-4.0`) test-compile. `DataPushQueueSuiteJ` and
`ChunkFetchIntegrationSuiteJ` (with its SSL subclass) were excluded; they hang
or fail on unmodified `main` locally.
- Not tested against real HDFS or S3/OSS.
--
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]
