[CELEBORN-2452] Prevent device error broadcast from deleting committed shuffle files

[wang-haihua]

What changes were proposed in this pull request?

Add a committed flag on TierWriterBase, set in close() right before notifyFileCommitted(). 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 for full analysis.

Does this PR resolve a correctness bug?

• Yes

Does this PR introduce any user-facing change?

No.

How was this patch tested?

• Existing DiskReducePartitionDataWriterSuiteJ write/close tests pass, confirming the added committed flag in close() doesn't regress the normal commit path.
• The notifyError guard itself is verified by tracing all destroy() callers (device-error must not delete committed files; rollback/expiry still must) and by a successful worker test-compile.

[Copilot]

🟡 Changes recommended

The new notifyError committed-guard is not synchronized with close()/destroy(), so a device-error notification racing with close can still destroy a now-committed file.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

Prevents DeviceMonitor CRITICAL_ERROR broadcasts from deleting already-committed shuffle files by introducing a committed-state signal on tier writers and using it to gate error-triggered destruction, reducing the risk of FileNotFoundException and shuffle data loss during fetch.

Changes:

• Add a committed flag to TierWriterBase and set it during close() immediately before notifyFileCommitted().
• In PartitionDataWriter.notifyError, skip destroy() when the current writer is already committed.

File summaries

File	Description
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/TierWriter.scala	Adds and sets the committed lifecycle flag on tier writers during close/commit.
worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionDataWriter.java	Uses the committed flag to avoid destroying committed writers on device error broadcasts.

Review details

• Files reviewed: 2/2 changed files
• Comments generated: 1
• Review effort level: Lite

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[wang-haihua]

Fixed in a3b1722 by making notifyError synchronized. I think no deadlock risk: the only lock order is ObservedDevice → PartitionDataWriter; the destroy → unregisterFileWriter path removes from a ConcurrentHashMap.newKeySet and never takes the ObservedDevice lock. @SteNicholas PTAL, thanks.

[wang-haihua]

@SteNicholas @cxzl25 @pan3793 PTAL, thanks.