rename

Author: turboFei

client-spark/spark-2/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java

@@ -109,11 +109,11 @@ private void initializeLifecycleManager(String appId) {
             lifecycleManager.registerShuffleTrackerCallback(
                 shuffleId -> mapOutputTracker.unregisterAllMapOutput(shuffleId));
 
-            lifecycleManager.registerBroadcastGetReducerFileGroupResponse(
+            lifecycleManager.registerBroadcastGetReducerFileGroupResponseCallback(
                 (shuffleId, getReducerFileGroupResponse) ->
                     SparkUtils.serializeGetReducerFileGroupResponse(
                         shuffleId, getReducerFileGroupResponse));
-            lifecycleManager.registerInvalidatedBroadcastGetReducerFileGroupResponse(
+            lifecycleManager.registerInvalidatedBroadcastCallback(
                 shuffleId -> SparkUtils.invalidateSerializedGetReducerFileGroupResponse(shuffleId));
           }
         }

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

@@ -151,11 +151,11 @@ private void initializeLifecycleManager(String appId) {
             lifecycleManager.registerShuffleTrackerCallback(
                 shuffleId -> SparkUtils.unregisterAllMapOutput(mapOutputTracker, shuffleId));
 
-            lifecycleManager.registerBroadcastGetReducerFileGroupResponse(
+            lifecycleManager.registerBroadcastGetReducerFileGroupResponseCallback(
                 (shuffleId, getReducerFileGroupResponse) ->
                     SparkUtils.serializeGetReducerFileGroupResponse(
                         shuffleId, getReducerFileGroupResponse));
-            lifecycleManager.registerInvalidatedBroadcastGetReducerFileGroupResponse(
+            lifecycleManager.registerInvalidatedBroadcastCallback(
                 shuffleId -> SparkUtils.invalidateSerializedGetReducerFileGroupResponse(shuffleId));
           }
         }

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

@@ -1840,20 +1840,20 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
     cancelShuffleCallback = Some(callback)
   }
 
-  @volatile private var broadcastGetReducerFileGroupResponse
+  @volatile private var broadcastGetReducerFileGroupResponseCallback
       : Option[java.util.function.BiFunction[Integer, GetReducerFileGroupResponse, Array[Byte]]] =
     None
-  def registerBroadcastGetReducerFileGroupResponse(call: java.util.function.BiFunction[
+  def registerBroadcastGetReducerFileGroupResponseCallback(call: java.util.function.BiFunction[
     Integer,
     GetReducerFileGroupResponse,
     Array[Byte]]): Unit = {
-    broadcastGetReducerFileGroupResponse = Some(call)
+    broadcastGetReducerFileGroupResponseCallback = Some(call)
   }
 
-  @volatile private var invalidatedBroadcastGetReducerFileGroupResponse: Option[Consumer[Integer]] =
+  @volatile private var invalidatedBroadcastCallback: Option[Consumer[Integer]] =
     None
-  def registerInvalidatedBroadcastGetReducerFileGroupResponse(call: Consumer[Integer]): Unit = {
-    invalidatedBroadcastGetReducerFileGroupResponse = Some(call)
+  def registerInvalidatedBroadcastCallback(call: Consumer[Integer]): Unit = {
+    invalidatedBroadcastCallback = Some(call)
   }
 
   def invalidateLatestMaxLocsCache(shuffleId: Int): Unit = {
@@ -1893,14 +1893,14 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
   def broadcastGetReducerFileGroupResponse(
       shuffleId: Int,
       response: GetReducerFileGroupResponse): Option[Array[Byte]] = {
-    broadcastGetReducerFileGroupResponse match {
+    broadcastGetReducerFileGroupResponseCallback match {
       case Some(c) => Option(c.apply(shuffleId, response))
       case _ => None
     }
   }
 
   private def invalidatedBroadcastGetReducerFileGroupResponse(shuffleId: Int): Unit = {
-    invalidatedBroadcastGetReducerFileGroupResponse match {
+    invalidatedBroadcastCallback match {
       case Some(c) => c.accept(shuffleId)
       case _ =>
     }