Remove 'sendWithReply(message: Any, sender: RpcEndpointRef): Unit'

Author: zsxwing

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

@@ -326,15 +326,6 @@ private[spark] abstract class RpcEndpointRef(@transient conf: SparkConf)
    */
   def send(message: Any): Unit
 
-  /**
-   * Send a message to the corresponding [[RpcEndpoint.receiveAndReply]] asynchronously.
-   * Fire-and-forget semantics.
-   *
-   * The receiver will reply to sender's [[RpcEndpoint.receive]] or [[RpcEndpoint.receiveAndReply]]
-   * depending on which one of [[RpcCallContext.reply]]s is called.
-   */
-  def sendWithReply(message: Any, sender: RpcEndpointRef): Unit
-
   /**
    * Send a message to the corresponding [[RpcEndpoint.receiveAndReply)]] and return a `Future` to
    * receive the reply within a default timeout.

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

@@ -291,17 +291,6 @@ private[akka] class AkkaRpcEndpointRef(
     actorRef ! AkkaMessage(message, false)
   }
 
-  override def sendWithReply(message: Any, sender: RpcEndpointRef): Unit = {
-    implicit val actorSender: ActorRef =
-      if (sender == null) {
-        Actor.noSender
-      } else {
-        require(sender.isInstanceOf[AkkaRpcEndpointRef])
-        sender.asInstanceOf[AkkaRpcEndpointRef].actorRef
-      }
-    actorRef ! AkkaMessage(message, true)
-  }
-
   override def sendWithReply[T: ClassTag](message: Any, timeout: FiniteDuration): Future[T] = {
     import scala.concurrent.ExecutionContext.Implicits.global
     actorRef.ask(AkkaMessage(message, true))(timeout).flatMap {

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

@@ -171,39 +171,6 @@ abstract class RpcEnvSuite extends FunSuite with BeforeAndAfterAll {
     }
   }
 
-  test("ping pong") {
-    val pongRef = env.setupEndpoint("pong", new RpcEndpoint {
-      override val rpcEnv = env
-
-      override def receiveAndReply(context: RpcCallContext) = {
-        case Ping(id) => context.sender.sendWithReply(Pong(id), self)
-      }
-    })
-
-    val pingRef = env.setupEndpoint("ping", new RpcEndpoint {
-      override val rpcEnv = env
-
-      var requester: RpcCallContext = _
-
-      override def receiveAndReply(context: RpcCallContext) = {
-        case Start => {
-          requester = context
-          pongRef.sendWithReply(Ping(1), self)
-        }
-        case p @ Pong(id) => {
-          if (id < 10) {
-            context.sender.sendWithReply(Ping(id + 1), self)
-          } else {
-            requester.reply(p)
-          }
-        }
-      }
-    })
-
-    val reply = pingRef.askWithReply[Pong](Start)
-    assert(Pong(10) === reply)
-  }
-
   test("onStart and onStop") {
     val stopLatch = new CountDownLatch(1)
     val calledMethods = mutable.ArrayBuffer[String]()