(incubator-celeborn) branch branch-0.4 updated: [CELEBORN-1233] Treat unfound PartitionLocation as failed in Controller#commitFiles

[zh...@apache.org]

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

zhouky pushed a commit to branch branch-0.4
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git


The following commit(s) were added to refs/heads/branch-0.4 by this push:
     new d5ec9ae8f [CELEBORN-1233] Treat unfound PartitionLocation as failed in Controller#commitFiles
d5ec9ae8f is described below

commit d5ec9ae8f699da61f8526e4cd2c3e957d8bea70d
Author: zky.zhoukeyong <zk...@alibaba-inc.com>
AuthorDate: Thu Jan 18 11:29:44 2024 +0800

    [CELEBORN-1233] Treat unfound PartitionLocation as failed in Controller#commitFiles
    
    ### What changes were proposed in this pull request?
    I tested 1T TPCDS with the following Celeborn 8-worker cluster setup:
    1. Workers have fixed ports for rpc/push/replicate
    2. `spark.celeborn.client.spark.fetch.throwsFetchFailure` is enabled
    3. graceful shutdown is enabled
    
    I randomly kill -9 and ./sbin/stop-worker.sh (both graceful shutdown and non-graceful shutdown) some workers and start it immediately. Then I encountered result incorrectness with low probability (1 out of 99 queries).
    
    After digging into it, I found the reason is as follows:
    1. At time T1, all workers are serving shuffle 602
    2. At time T2, I run stop-worker.sh for worker2, and then run kill -9 and start worker1. Since the workers are configured with fixed ports, clients think they are OK and Master let them re-register, which will also success. And worker2 is clean in memory.
    4. At time T3, push requests to worker2 fails and revives on worker1, so worker1 has reservation for shuffle 602. Then I start worker2.
    5. At time T4, LifecycleManager sends CommitFiles to all workers, on worker1, it just logs that some PartitionLocations
        don't exist and ignores them.
    6. CommitFiles success, but worker1 loses some data before restarting, and no error happens.
    
    The following snapshot shows the process.
    
    ![image](https://github.com/apache/incubator-celeborn/assets/948245/9ef1a1ff-bb26-420a-929c-70c9476ec700)
    
    This PR fixes this by treating unfound PartitionLocations as failed when handling CommitFiles.
    
    ### Why are the changes needed?
    ditto
    
    ### Does this PR introduce _any_ user-facing change?
    No.
    
    ### How was this patch tested?
    Manual test
    
    Closes #2235 from waitinfuture/1233.
    
    Authored-by: zky.zhoukeyong <zk...@alibaba-inc.com>
    Signed-off-by: zky.zhoukeyong <zk...@alibaba-inc.com>
    (cherry picked from commit 749a0fa43974d5f2b8f0ce9f7b61e032fb8f8aea)
    Signed-off-by: zky.zhoukeyong <zk...@alibaba-inc.com>
---
 .../test/scala/org/apache/celeborn/client/WithShuffleClientSuite.scala | 2 --
 .../scala/org/apache/celeborn/service/deploy/worker/Controller.scala   | 3 ++-
 2 files changed, 2 insertions(+), 3 deletions(-)

diff --git a/client/src/test/scala/org/apache/celeborn/client/WithShuffleClientSuite.scala b/client/src/test/scala/org/apache/celeborn/client/WithShuffleClientSuite.scala
index 5077f5192..88247d9b6 100644
--- a/client/src/test/scala/org/apache/celeborn/client/WithShuffleClientSuite.scala
+++ b/client/src/test/scala/org/apache/celeborn/client/WithShuffleClientSuite.scala
@@ -174,8 +174,6 @@ trait WithShuffleClientSuite extends CelebornFunSuite {
     // task number incr to numMappers + 1
     shuffleClient.registerMapPartitionTask(shuffleId, numMappers, mapId, attemptId + 1, 9)
     shuffleClient.mapPartitionMapperEnd(shuffleId, mapId, attemptId, numMappers, 1)
-    // retry
-    shuffleClient.mapPartitionMapperEnd(shuffleId, mapId, attemptId, numMappers, 1)
     // another attempt
     shuffleClient.mapPartitionMapperEnd(
       shuffleId,
diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala
index f2548032a..9351bf5c8 100644
--- a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala
+++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala
@@ -295,7 +295,8 @@ private[deploy] class Controller(
                   }
 
                 if (location == null) {
-                  logWarning(s"Get Partition Location for $shuffleKey $uniqueId but didn't exist.")
+                  logError(s"Get Partition Location for $shuffleKey $uniqueId but didn't exist, treat as failed.")
+                  failedIds.add(uniqueId)
                   return
                 }