Rename RpcResponse to RpcCallContext

Author: zsxwing

core/src/main/scala/org/apache/spark/rpc/RpcEnv.scala

@@ -182,17 +182,17 @@ private[spark] trait RpcEndpoint {
   }
 
   /**
-   * Process messages from [[RpcEndpointRef.send]] or [[RpcResponse.reply)]]
+   * Process messages from [[RpcEndpointRef.send]] or [[RpcCallContext.reply)]]
    */
   def receive: PartialFunction[Any, Unit] = {
     case _ =>
       // network events will be passed here by default, so do nothing by default to avoid noise.
   }
 
   /**
-   * Process messages from [[RpcEndpointRef.sendWithReply]] or [[RpcResponse.replyWithSender)]]
+   * Process messages from [[RpcEndpointRef.sendWithReply]] or [[RpcCallContext.replyWithSender)]]
    */
-  def receiveAndReply(response: RpcResponse): PartialFunction[Any, Unit] = {
+  def receiveAndReply(response: RpcCallContext): PartialFunction[Any, Unit] = {
     case _ => response.fail(new SparkException(self + " won't reply anything"))
   }
 
@@ -281,7 +281,7 @@ private[spark] trait RpcEndpointRef {
    * Fire-and-forget semantics.
    *
    * The receiver will reply to sender's [[RpcEndpoint.receive]] or [[RpcEndpoint.receiveAndReply]]
-   * depending on which one of [[RpcResponse.reply]]s is called.
+   * depending on which one of [[RpcCallContext.reply]]s is called.
    */
   def sendWithReply(message: Any, sender: RpcEndpointRef): Unit
 
@@ -351,7 +351,7 @@ private[spark] case class NetworkErrorEvent(address: RpcAddress, cause: Throwabl
 /**
  * A callback that [[RpcEndpoint]] can use it to send back a message or failure.
  */
-private[spark] trait RpcResponse {
+private[spark] trait RpcCallContext {
 
   /**
    * Reply a message to the sender. If the sender is [[RpcEndpoint]], its [[RpcEndpoint.receive]]

core/src/main/scala/org/apache/spark/rpc/akka/AkkaRpcEnv.scala

@@ -151,7 +151,7 @@ private[spark] class AkkaRpcEnv private (
               val s = sender()
               val pf =
                 if (reply) {
-                  endpoint.receiveAndReply(new RpcResponse {
+                  endpoint.receiveAndReply(new RpcCallContext {
                     override def fail(e: Throwable): Unit = {
                       s ! AkkaFailure(e)
                     }

core/src/main/scala/org/apache/spark/scheduler/OutputCommitCoordinator.scala

@@ -20,7 +20,7 @@ package org.apache.spark.scheduler
 import scala.collection.mutable
 
 import org.apache.spark._
-import org.apache.spark.rpc.{RpcResponse, RpcEndpointRef, RpcEnv, RpcEndpoint}
+import org.apache.spark.rpc.{RpcCallContext, RpcEndpointRef, RpcEnv, RpcEndpoint}
 
 private sealed trait OutputCommitCoordinationMessage extends Serializable
 
@@ -156,7 +156,7 @@ private[spark] object OutputCommitCoordinator {
       override val rpcEnv: RpcEnv, outputCommitCoordinator: OutputCommitCoordinator)
     extends RpcEndpoint with Logging {
 
-    override def receiveAndReply(response: RpcResponse) = {
+    override def receiveAndReply(response: RpcCallContext) = {
       case AskPermissionToCommitOutput(stage, partition, taskAttempt) =>
         response.reply(
           outputCommitCoordinator.handleAskPermissionToCommit(stage, partition, taskAttempt))

core/src/test/scala/org/apache/spark/rpc/RpcEnvSuite.scala

@@ -93,7 +93,7 @@ abstract class RpcEnvSuite extends FunSuite with BeforeAndAfterAll {
     val endpoint = new RpcEndpoint {
       override val rpcEnv = env
 
-      override def receiveAndReply(response: RpcResponse) = {
+      override def receiveAndReply(response: RpcCallContext) = {
         case "Hello" => response.reply(self)
         case "Echo" => response.reply("Echo")
       }
@@ -109,7 +109,7 @@ abstract class RpcEnvSuite extends FunSuite with BeforeAndAfterAll {
     val rpcEndpointRef = env.setupEndpoint("ask-locally", new RpcEndpoint {
       override val rpcEnv = env
 
-      override def receiveAndReply(response: RpcResponse) = {
+      override def receiveAndReply(response: RpcCallContext) = {
         case msg: String => {
           response.reply(msg)
         }
@@ -123,7 +123,7 @@ abstract class RpcEnvSuite extends FunSuite with BeforeAndAfterAll {
     env.setupEndpoint("ask-remotely", new RpcEndpoint {
       override val rpcEnv = env
 
-      override def receiveAndReply(response: RpcResponse) = {
+      override def receiveAndReply(response: RpcCallContext) = {
         case msg: String => {
           response.reply(msg)
         }
@@ -146,7 +146,7 @@ abstract class RpcEnvSuite extends FunSuite with BeforeAndAfterAll {
     env.setupEndpoint("ask-timeout", new RpcEndpoint {
       override val rpcEnv = env
 
-      override def receiveAndReply(response: RpcResponse) = {
+      override def receiveAndReply(response: RpcCallContext) = {
         case msg: String => {
           Thread.sleep(100)
           response.reply(msg)
@@ -175,17 +175,17 @@ abstract class RpcEnvSuite extends FunSuite with BeforeAndAfterAll {
     val pongRef = env.setupEndpoint("pong", new RpcEndpoint {
       override val rpcEnv = env
 
-      override def receiveAndReply(response: RpcResponse) = {
+      override def receiveAndReply(response: RpcCallContext) = {
         case Ping(id) => response.replyWithSender(Pong(id), self)
       }
     })
 
     val pingRef = env.setupEndpoint("ping", new RpcEndpoint {
       override val rpcEnv = env
 
-      var requester: RpcResponse = _
+      var requester: RpcCallContext = _
 
-      override def receiveAndReply(response: RpcResponse) = {
+      override def receiveAndReply(response: RpcCallContext) = {
         case Start => {
           requester = response
           pongRef.sendWithReply(Ping(1), self)
@@ -430,7 +430,7 @@ abstract class RpcEnvSuite extends FunSuite with BeforeAndAfterAll {
     val endpointRef = env.setupEndpoint("sendWithReply", new RpcEndpoint {
       override val rpcEnv = env
 
-      override def receiveAndReply(response: RpcResponse) = {
+      override def receiveAndReply(response: RpcCallContext) = {
         case m => response.reply("ack")
       }
     })
@@ -446,7 +446,7 @@ abstract class RpcEnvSuite extends FunSuite with BeforeAndAfterAll {
     env.setupEndpoint("sendWithReply-remotely", new RpcEndpoint {
       override val rpcEnv = env
 
-      override def receiveAndReply(response: RpcResponse) = {
+      override def receiveAndReply(response: RpcCallContext) = {
         case m => response.reply("ack")
       }
     })
@@ -468,7 +468,7 @@ abstract class RpcEnvSuite extends FunSuite with BeforeAndAfterAll {
     val endpointRef = env.setupEndpoint("sendWithReply-error", new RpcEndpoint {
       override val rpcEnv = env
 
-      override def receiveAndReply(response: RpcResponse) = {
+      override def receiveAndReply(response: RpcCallContext) = {
         case m => response.fail(new SparkException("Oops"))
       }
     })
@@ -486,7 +486,7 @@ abstract class RpcEnvSuite extends FunSuite with BeforeAndAfterAll {
     env.setupEndpoint("sendWithReply-remotely-error", new RpcEndpoint {
       override val rpcEnv = env
 
-      override def receiveAndReply(response: RpcResponse) = {
+      override def receiveAndReply(response: RpcCallContext) = {
         case msg: String => response.fail(new SparkException("Oops"))
       }
     })