You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@celeborn.apache.org by "ewoodbury (via GitHub)" <gi...@apache.org> on 2026/10/19 01:12:26 UTC
[PR] [CELEBORN-2467] Move DFS replica dedup from worker close to the driver after commit [celeborn]
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2467] Fix silent shuffle data loss when a replica's commit times out on OSS and HDFS storage [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas closed pull request #3879: [CELEBORN-2467] Fix silent shuffle data loss when a replica's commit times out on OSS and HDFS storage
URL: https://github.com/apache/celeborn/pull/3879
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2467] Fix silent shuffle data loss when a replica's commit times out on OSS and HDFS storage [celeborn]
Posted by "ewoodbury (via GitHub)" <gi...@apache.org>.
ewoodbury commented on code in PR #3879:
URL: https://github.com/apache/celeborn/pull/3879#discussion_r4331530132
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/TierWriter.scala:
##########
@@ -691,39 +699,36 @@ class DfsTierWriter(
}
}
+ // A copy whose data stream or index failed to close is unreadable, so it must not count as
+ // committed. S3/OSS upload completion failures are unchecked and already fail the commit.
+ override protected def failCommitOnCloseStreamsError: Boolean = true
Review Comment:
Thanks - addressed in: https://github.com/apache/celeborn/commit/86cd8f9a3c937660eb4c0ca345202b0ce21db561, so these stream-close failures get propagated, covered by a regression test
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2467] Fix silent shuffle data loss when a replica's commit times out on OSS and HDFS storage [celeborn]
Posted by "ewoodbury (via GitHub)" <gi...@apache.org>.
ewoodbury closed pull request #3879: [CELEBORN-2467] Fix silent shuffle data loss when a replica's commit times out on OSS and HDFS storage
URL: https://github.com/apache/celeborn/pull/3879
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2467] Fix silent shuffle data loss when a replica's commit times out on OSS and HDFS storage [celeborn]
Posted by "ewoodbury (via GitHub)" <gi...@apache.org>.
ewoodbury commented on PR #3879:
URL: https://github.com/apache/celeborn/pull/3879#issuecomment-6283329248
Pushed an update: For HDFS `cleanFile` now deletes first and only stops tracking the file once all of its files (data, index, .success, .sorted) are gone. Otherwise it logs a warning and throws, so `DestroyWorkerSlots` replies PARTIAL_SUCCESS and the driver retries. Shuffle expiry cleans it up if the retries also fail.
One change from my earlier comment: the writer is still dropped up front since it can only delete once. Retries delete through the file info instead. I added tests for repeated failures, a leftover index file, a successful retry, and expiry during a failed delete.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2467] Fix silent shuffle data loss when a replica's commit times out on OSS and HDFS storage [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #3879:
URL: https://github.com/apache/celeborn/pull/3879#discussion_r4324814006
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/TierWriter.scala:
##########
@@ -691,39 +699,36 @@ class DfsTierWriter(
}
}
+ // A copy whose data stream or index failed to close is unreadable, so it must not count as
+ // committed. S3/OSS upload completion failures are unchecked and already fail the commit.
+ override protected def failCommitOnCloseStreamsError: Boolean = true
Review Comment:
@ewoodbury, propagate HDFS append-close failures before releasing replicas. With `celeborn.worker.reuse.hdfs.outputStream.enabled=false` (the default), `hdfsStream` is already null here. Data streams are closed inside `DfsFlushTask.flush`, which still catches and logs `IOException` without notifying the writer, so this new flag does not cover the default HDFS write path.
I reproduced this on `f4048b1` by injecting an append-stream close failure with the data left unpersisted: `close()` returned `1024`, `committed=true`, the data file contained `0` bytes, and `.success` existed. If this happens to the primary while its replica commits successfully, the new driver-side selection can unpublish and delete the healthy replica, leaving the unreadable primary.
Please propagate per-flush stream-close failures through the flush notifier, preserving any earlier write failure, and add a regression test with stream reuse disabled.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2467] Fix silent shuffle data loss when a replica's commit times out on OSS and HDFS storage [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on PR #3879:
URL: https://github.com/apache/celeborn/pull/3879#issuecomment-6306869721
**Merge Summary:**
- merged into main https://github.com/apache/celeborn/commit/a3335bee
- merged into branch-0.7 https://github.com/apache/celeborn/commit/c857dd7a
*Posted by `merge_pr.py`*
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2467] Fix silent shuffle data loss when a replica's commit times out on OSS and HDFS storage [celeborn]
Posted by "ewoodbury (via GitHub)" <gi...@apache.org>.
ewoodbury commented on PR #3879:
URL: https://github.com/apache/celeborn/pull/3879#issuecomment-6282229467
Thanks @zaynt4606!
On the destroy, that's a good point, it makes sense to track these failure better. Looks like the update to cover this would be slim (<100 new lines) so I'll plan to do that: updating StorageManager.cleanFile so a failed delete is not reported as success. It'll keep the file info and writer, and throw so `DestroyWorkerSlots` replies PARTIAL_SUCCESS and the client retries. If the retries also fail, the file stays tracked and is removed at shuffle expiry.
Will add a failure injection test as well, covering the updated path as well, makes sense
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2467] Fix silent shuffle data loss when a replica's commit times out on OSS and HDFS storage [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #3879:
URL: https://github.com/apache/celeborn/pull/3879#discussion_r4324814006
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/TierWriter.scala:
##########
@@ -691,39 +699,36 @@ class DfsTierWriter(
}
}
+ // A copy whose data stream or index failed to close is unreadable, so it must not count as
+ // committed. S3/OSS upload completion failures are unchecked and already fail the commit.
+ override protected def failCommitOnCloseStreamsError: Boolean = true
Review Comment:
[P1] Propagate HDFS append-close failures before releasing replicas
With `celeborn.worker.reuse.hdfs.outputStream.enabled=false` (the default), `hdfsStream` is already null here. Data streams are closed inside `DfsFlushTask.flush`, which still catches and logs `IOException` without notifying the writer, so this new flag does not cover the default HDFS write path.
I reproduced this on `f4048b1` by injecting an append-stream close failure with the data left unpersisted: `close()` returned `1024`, `committed=true`, the data file contained `0` bytes, and `.success` existed. If this happens to the primary while its replica commits successfully, the new driver-side selection can unpublish and delete the healthy replica, leaving the unreadable primary.
Please propagate per-flush stream-close failures through the flush notifier, preserving any earlier write failure, and add a regression test with stream reuse disabled.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2467] Fix silent shuffle data loss when a replica's commit times out on OSS and HDFS storage [celeborn]
Posted by "ewoodbury (via GitHub)" <gi...@apache.org>.
ewoodbury commented on PR #3879:
URL: https://github.com/apache/celeborn/pull/3879#issuecomment-6304336749
Thanks @SteNicholas , good catch - Fixed so `DfsFlushTask.flush` now re-throws a stream-close failure instead of just logging it. If the write itself failed, that error is kept and the close error is added as suppressed. The flusher sets it on the notifier, so the writer's `close()` fails the commit, that copy is reported failed, and its replica is never released.
Added a regression test with stream reuse disabled, where the append stream drops the data and fails on close. Also rebased onto main to pick up the CI fix; only new change is the last commit: https://github.com/apache/celeborn/pull/3879/changes/86cd8f9a3c937660eb4c0ca345202b0ce21db561
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org