wang-haihua opened a new pull request, #3837:
URL: https://github.com/apache/celeborn/pull/3837
### What changes were proposed in this pull request?
Add a `committed` flag on `TierWriterBase`, set in `close()` right before
`notifyFileCommitted()` (exactly when the file becomes durable and fetchable).
`PartitionDataWriter.notifyError` then skips `destroy()` when the writer is
already committed, so a disk `CRITICAL_ERROR` broadcast no longer deletes
committed files.
The rollback path (`Controller.destroyWriters`) and shuffle-expiry cleanup
(`StorageManager.cleanFileInternal`) call `destroy()` directly rather than
through `notifyError`, so they are unaffected and still remove committed files
when the shuffle actually ends.
### Why are the changes needed?
`DeviceMonitor` broadcasts `CRITICAL_ERROR` to every writer registered on a
disk:
```
ObservedDevice.notifyObserversOnError(CRITICAL_ERROR)
-> PartitionDataWriter.notifyError -> destroy(IOException)
-> TierWriterBase.destroy -> cleanLocalOrDfsFiles()
-> DiskFileInfo.deleteAllFiles(null) // deletes data + index + sorted
files
```
A `LocalTierWriter` registers as an observer at file creation and, before
this fix, is removed from the observer set only on the `destroy` path — never
when the file commits. So a committed writer stays registered for the entire
remaining lifetime of the shuffle. If the disk crosses the `CRITICAL_ERROR`
threshold during that window, the broadcast destroys the committed writer and
deletes its data/index/sorted files out from under in-flight reducers, causing
`FileNotFoundException` on fetch and genuine shuffle data loss. The broadcast
hits every observer on the disk, so a committed file is deleted merely for
sharing a disk with unrelated failing writes.
See [CELEBORN-2452](https://issues.apache.org/jira/browse/CELEBORN-2452) for
full analysis.
### Does this PR resolve a correctness bug?
- [x] Yes
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
- Traced every `destroy()` caller to confirm the guard belongs at the
`notifyError` entry (device-error broadcast) and not inside `destroy()`, which
is shared with the rollback/expiry paths that must still delete committed files.
- Verified `worker` (with dependent modules) `test-compile` passes,
confirming the Scala `@volatile var committed` getter is callable from the Java
`PartitionDataWriter`.
--
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]
