(celeborn) branch branch-0.5 updated: [CELEBORN-1663][FOLLOWUP] Only register appShuffleDeterminate if stage using celeborn for shuffle

[nicholasjiang]

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

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


The following commit(s) were added to refs/heads/branch-0.5 by this push:
     new c3c5ee091 [CELEBORN-1663][FOLLOWUP] Only register 
appShuffleDeterminate if stage using celeborn for shuffle
c3c5ee091 is described below

commit c3c5ee091ccfab12cad8e52b92837a01bb1d4b31
Author: Sanskar Modi <sanskarmod...@gmail.com>
AuthorDate: Tue Oct 22 14:28:12 2024 +0800

    [CELEBORN-1663][FOLLOWUP] Only register appShuffleDeterminate if stage 
using celeborn for shuffle
---
 .../org/apache/spark/shuffle/celeborn/SparkShuffleManager.java   | 9 +++++----
 1 file changed, 5 insertions(+), 4 deletions(-)

diff --git 
a/client-spark/spark-2/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java
 
b/client-spark/spark-2/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java
index aa5549184..5be2b5c26 100644
--- 
a/client-spark/spark-2/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java
+++ 
b/client-spark/spark-2/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java
@@ -116,16 +116,17 @@ public class SparkShuffleManager implements 
ShuffleManager {
     appUniqueId = SparkUtils.appUniqueId(dependency.rdd().context());
     initializeLifecycleManager(appUniqueId);
 
-    lifecycleManager.registerAppShuffleDeterminate(
-        shuffleId,
-        dependency.rdd().getOutputDeterministicLevel() != 
DeterministicLevel.INDETERMINATE());
-
     if (fallbackPolicyRunner.applyAllFallbackPolicy(
         lifecycleManager, dependency.partitioner().numPartitions())) {
       logger.warn("Fallback to SortShuffleManager!");
       sortShuffleIds.add(shuffleId);
       return sortShuffleManager().registerShuffle(shuffleId, numMaps, 
dependency);
     } else {
+      lifecycleManager.registerAppShuffleDeterminate(
+          shuffleId,
+          !DeterministicLevel.INDETERMINATE()
+              .equals(dependency.rdd().getOutputDeterministicLevel()));
+
       return new CelebornShuffleHandle<>(
           appUniqueId,
           lifecycleManager.getHost(),