Rename 'fail' to 'sendFailure' and other minor code style changes

Author: zsxwing

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

@@ -192,7 +192,7 @@ private[spark] trait RpcEndpoint {
    * Process messages from [[RpcEndpointRef.sendWithReply]] or [[RpcCallContext.replyWithSender)]]
    */
   def receiveAndReply(response: RpcCallContext): PartialFunction[Any, Unit] = {
-    case _ => response.fail(new SparkException(self + " won't reply anything"))
+    case _ => response.sendFailure(new SparkException(self + " won't reply anything"))
   }
 
   /**
@@ -374,5 +374,5 @@ private[spark] trait RpcCallContext {
   /**
    * Report a failure to the sender.
    */
-  def fail(e: Throwable): Unit
+  def sendFailure(e: Throwable): Unit
 }

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

@@ -20,9 +20,8 @@ package org.apache.spark.rpc.akka
 import java.net.URI
 import java.util.concurrent.{ConcurrentHashMap, CountDownLatch}
 
-import scala.concurrent.Await
+import scala.concurrent.{Await, Future}
 import scala.concurrent.duration._
-import scala.concurrent.Future
 import scala.language.postfixOps
 import scala.reflect.ClassTag
 import scala.util.control.NonFatal
@@ -139,7 +138,7 @@ private[spark] class AkkaRpcEnv private (
               val pf =
                 if (reply) {
                   endpoint.receiveAndReply(new RpcCallContext {
-                    override def fail(e: Throwable): Unit = {
+                    override def sendFailure(e: Throwable): Unit = {
                       s ! AkkaFailure(e)
                     }
 
@@ -329,9 +328,10 @@ private[akka] class AkkaRpcEndpointRef(
   override def sendWithReply[T: ClassTag](message: Any, timeout: FiniteDuration): Future[T] = {
     import scala.concurrent.ExecutionContext.Implicits.global
     actorRef.ask(AkkaMessage(message, true))(timeout).flatMap {
-      case AkkaMessage(message, reply) =>
+      case msg @ AkkaMessage(message, reply) =>
         if (reply) {
-          Future.failed(new SparkException("The sender cannot reply"))
+          logError(s"Receive $msg but the sender cannot reply")
+          Future.failed(new SparkException(s"Receive $msg but the sender cannot reply"))
         } else {
           Future.successful(message)
         }

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

@@ -469,7 +469,7 @@ abstract class RpcEnvSuite extends FunSuite with BeforeAndAfterAll {
       override val rpcEnv = env
 
       override def receiveAndReply(response: RpcCallContext) = {
-        case m => response.fail(new SparkException("Oops"))
+        case m => response.sendFailure(new SparkException("Oops"))
       }
     })
 
@@ -487,7 +487,7 @@ abstract class RpcEnvSuite extends FunSuite with BeforeAndAfterAll {
       override val rpcEnv = env
 
       override def receiveAndReply(response: RpcCallContext) = {
-        case msg: String => response.fail(new SparkException("Oops"))
+        case msg: String => response.sendFailure(new SparkException("Oops"))
       }
     })
 

core/src/test/scala/org/apache/spark/rpc/akka/AkkaRpcEnvSuite.scala

@@ -39,7 +39,7 @@ class AkkaRpcEnvSuite extends RpcEnvSuite {
       AkkaRpcEnv(RpcEnvConfig(conf, "test", "localhost", 12346, new SecurityManager(conf)))
     try {
       val newRef = newRpcEnv.setupEndpointRef("local", ref.address, "test_endpoint")
-      assert(s"akka.tcp://local@localhost:12345/user/test_endpoint" ===
+      assert("akka.tcp://local@localhost:12345/user/test_endpoint" ===
         newRef.asInstanceOf[AkkaRpcEndpointRef].actorRef.path.toString)
     } finally {
       newRpcEnv.shutdown()