[CELEBORN-2140] Add logs for track unfinished maptasks

client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/HashBasedShuffleWriter.java

@@ -379,6 +379,12 @@ private void close() throws IOException, InterruptedException {
 
     long waitStartTime = System.nanoTime();
     shuffleClient.mapperEnd(shuffleId, mapId, encodedAttemptId, numMappers, numPartitions);
+    logger.debug(
+        "Finish mapEnd RPC for shuffleId={} mapId={} attemptNumber={} numMappers={}",
+        shuffleId,
+        mapId,
+        taskContext.attemptNumber(),
+        numMappers);
     writeMetrics.incWriteTime(System.nanoTime() - waitStartTime);
 
     BlockManagerId bmId = SparkEnv.get().blockManager().shuffleServerId();

client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedShuffleWriter.java

@@ -382,6 +382,12 @@ private void close() throws IOException {
 
     long waitStartTime = System.nanoTime();
     shuffleClient.mapperEnd(shuffleId, mapId, encodedAttemptId, numMappers, numPartitions);
+    logger.debug(
+        "Finish mapEnd RPC for shuffleId={} mapId={} attemptNumber={} numMappers={}",
+        shuffleId,
+        mapId,
+        taskContext.attemptNumber(),
+        numMappers);
     writeMetrics.incWriteTime(System.nanoTime() - waitStartTime);
   }
 

client/src/main/scala/org/apache/celeborn/client/ClientUtils.scala

@@ -18,8 +18,9 @@
 package org.apache.celeborn.client
 
 import org.apache.celeborn.common.CelebornConf
+import org.apache.celeborn.common.internal.Logging
 
-object ClientUtils {
+object ClientUtils extends Logging {
 
   /**
    * Check if all the mapper attempts are finished. If any of the attempts is not finished, return false.
@@ -29,13 +30,11 @@ object ClientUtils {
    * @param attempts The mapper finished attemptId array. An attempt ID of -1 indicates that the mapper is not finished.
    * @return True if all mapper attempts are finished, false otherwise.
    */
-  def areAllMapperAttemptsFinished(attempts: Array[Int]): Boolean = {
-    var i = attempts.length - 1
-    while (i >= 0) {
-      if (attempts(i) < 0) {
-        return false
-      }
-      i -= 1
+  def areAllMapperAttemptsFinished(attempts: Array[Int], shuffleId: Int): Boolean = {
+    val length = attempts.length - 1
+    if (length >= 0 && attempts.exists(_ < 0)) {
+      val unfinishedTasks = attempts.filter(_ < 0).mkString(", ")
+      logDebug(s"For shuffle $shuffleId, exists unfinished map task $unfinishedTasks")
     }
     true
   }

client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala

@@ -994,7 +994,9 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
     }
 
     def areAllMapTasksEnd(shuffleId: Int): Boolean = {
-      ClientUtils.areAllMapperAttemptsFinished(commitManager.getMapperAttempts(shuffleId))
+      ClientUtils.areAllMapperAttemptsFinished(
+        commitManager.getMapperAttempts(shuffleId),
+        shuffleId)
     }
 
     shuffleIds.synchronized {

client/src/main/scala/org/apache/celeborn/client/commit/ReducePartitionCommitHandler.scala

@@ -304,7 +304,7 @@ class ReducePartitionCommitHandler(
           }
         }
         // Mapper with this attemptId finished, also check all other mapper finished or not.
-        (true, ClientUtils.areAllMapperAttemptsFinished(attempts))
+        (true, ClientUtils.areAllMapperAttemptsFinished(attempts, shuffleId))
       } else {
         // Mapper with another attemptId finished, skip this request
         (false, false)