(celeborn) branch branch-0.7 updated: [CELEBORN-2452] Prevent device error broadcast from deleting committed shuffle files

[zaynt]

This is an automated email from the ASF dual-hosted git repository.

zaynt4606 pushed a commit to branch branch-0.7
in repository https://gitbox.apache.org/repos/asf/celeborn.git


The following commit(s) were added to refs/heads/branch-0.7 by this push:
     new 5d54008d62 [CELEBORN-2452] Prevent device error broadcast from 
deleting committed shuffle files
5d54008d62 is described below

commit 5d54008d62fe1c1858b1626b95bf2b0e8933ac0a
Author: wang-haihua <wang-hai...@foxmail.com>
AuthorDate: Mon Oct 12 21:18:36 2026 +0800

    [CELEBORN-2452] Prevent device error broadcast from deleting committed 
shuffle files
    
    ### 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  [...]
    
    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?
    
    - 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.
    
    Closes #3837 from wang-haihua/CELEBORN-2452-committed-file-deletion.
    
    Authored-by: wang-haihua <wang-hai...@foxmail.com>
    Signed-off-by: zhengtao <shuaizhentao....@alibaba-inc.com>
    
    (cherry picked from commit 4d7fb8e4c88e8b63cffe518e59b325f5eaf47ef3)
    Signed-off-by: zhengtao <shuaizhentao....@alibaba-inc.com>
    
    Co-Authored-By: Codex <nore...@openai.com>
    AI-Model: gpt-6
    AI-Contributed/Feature: 0/9
    AI-Contributed/UT: 0/0
---
 .../service/deploy/worker/storage/PartitionDataWriter.java         | 7 ++++++-
 .../apache/celeborn/service/deploy/worker/storage/TierWriter.scala | 2 ++
 2 files changed, 8 insertions(+), 1 deletion(-)

diff --git 
a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionDataWriter.java
 
b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionDataWriter.java
index cf7ee4e5f4..720423829b 100644
--- 
a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionDataWriter.java
+++ 
b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionDataWriter.java
@@ -264,7 +264,12 @@ public class PartitionDataWriter implements DeviceObserver 
{
   }
 
   @Override
-  public void notifyError(String mountPoint, DiskStatus diskStatus) {
+  public synchronized void notifyError(String mountPoint, DiskStatus 
diskStatus) {
+    // A committed file is already flushed and visible to fetch requests,
+    // a device error broadcast must not destroy it.
+    if (currentTierWriter.committed()) {
+      return;
+    }
     destroy(
         new IOException(
             "Destroy FileWriter "
diff --git 
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/TierWriter.scala
 
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/TierWriter.scala
index 6a39236fcd..5c55cb0790 100644
--- 
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/TierWriter.scala
+++ 
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/TierWriter.scala
@@ -65,6 +65,7 @@ abstract class TierWriterBase(
 
   @volatile var closed: Boolean = false
   @volatile private var destroyed: Boolean = false
+  @volatile var committed: Boolean = false
 
   takeBuffer()
 
@@ -144,6 +145,7 @@ abstract class TierWriterBase(
           logWarning(s"close file writer $this failed", e)
       }
     }
+    committed = true
     notifyFileCommitted()
 
     fileInfo.getFileLength