bug tod

apache · Nov 11, 2024 · 5d628b9 · 5d628b9
1 parent 2d18124
commit 5d628b9
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 1 deletion.
diff --git a/...park/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReader.scala b/...park/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReader.scala
@@ -261,6 +261,7 @@ class CelebornShuffleReader[K, C](
       if (handle.numMappers > 0) {
         val startFetchWait = System.nanoTime()
         var inputStream: CelebornInputStream = streams.get(partitionId)
+        // todo bug fix:  split后，fetch时inputStream一直为空
         while (inputStream == null) {
           if (exceptionRef.get() != null) {
             exceptionRef.get() match {

diff --git a/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala b/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala
@@ -650,9 +650,10 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
     }
 
     // First, request to get allocated slots from Primary
-    val numGroupTask = math.ceil(numMappers.toDouble / groupMapTaskGroupSize).toInt
+    var numGroupTask = 1
     var groupNumPartitions = numPartitions
     if (partitionType.getValue.equals(PartitionType.REDUCE.getValue) && groupMapTaskEnabled) {
+      numGroupTask = math.ceil(numMappers.toDouble / groupMapTaskGroupSize).toInt
       groupNumPartitions = numPartitions * numGroupTask
     }