Fix the code style and add docs

Author: zsxwing

core/src/main/scala/org/apache/spark/deploy/worker/WorkerWatcher.scala

@@ -53,7 +53,7 @@ private[spark] class WorkerWatcher(override val rpcEnv: RpcEnv, workerUrl: Strin
   def exitNonZero() = if (isTesting) isShutDown = true else System.exit(-1)
 
   override def receive = {
-    case e => logWarning(s"Received unexpected actor system event: $e")
+    case e => logWarning(s"Received unexpected message: $e")
   }
 
   override def onConnected(remoteAddress: RpcAddress): Unit = {

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

@@ -293,6 +293,7 @@ private[spark] abstract class RpcEndpointRef(@transient conf: SparkConf)
   /**
    * Send a message to the corresponding [[RpcEndpoint]] and get its result within a default
    * timeout, or throw a SparkException if this fails even after the default number of retries.
+   * Because this method retries, the message handling in the receiver side should be idempotent.
    *
    * Note: this is a blocking action which may cost a lot of time,  so don't call it in an message
    * loop of [[RpcEndpoint]].
@@ -306,7 +307,8 @@ private[spark] abstract class RpcEndpointRef(@transient conf: SparkConf)
   /**
    * Send a message to the corresponding [[RpcEndpoint.receive]] and get its result within a
    * specified timeout, throw a SparkException if this fails even after the specified number of
-   * retries.
+   * retries. Because this method retries, the message handling in the receiver side should be
+   * idempotent.
    *
    * Note: this is a blocking action which may cost a lot of time, so don't call it in an message
    * loop of [[RpcEndpoint]].
@@ -350,12 +352,16 @@ private[spark] abstract class RpcEndpointRef(@transient conf: SparkConf)
   /**
    * Send a message to the corresponding [[RpcEndpoint.receiveAndReply)]] and return a `Future` to
    * receive the reply within a default timeout.
+   *
+   * This method only sends the message once and never retries.
    */
   def sendWithReply[T: ClassTag](message: Any): Future[T] = sendWithReply(message, defaultTimeout)
 
   /**
    * Send a message to the corresponding [[RpcEndpoint.receiveAndReply)]] and return a `Future` to
    * receive the reply within the specified timeout.
+   *
+   * This method only sends the message once and never retries.
    */
   def sendWithReply[T: ClassTag](message: Any, timeout: FiniteDuration): Future[T]
 

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

@@ -133,15 +133,18 @@ private[spark] class AkkaRpcEnv private[akka] (
           safelyCall(endpoint) {
             processMessage(endpoint, m, sender)
           }
+
         case AkkaFailure(e) =>
           try {
             endpoint.onError(e)
           } catch {
             case NonFatal(e) => logError(s"Ignore error: ${e.getMessage}", e)
           }
+
         case message: Any => {
           logWarning(s"Unknown message: $message")
         }
+
       }
 
       override def postStop(): Unit = {
@@ -160,9 +163,9 @@ private[spark] class AkkaRpcEnv private[akka] (
 
   private def processMessage(endpoint: RpcEndpoint, m: AkkaMessage, _sender: ActorRef): Unit = {
     val message = m.message
-    val reply = m.reply
+    val needReply = m.needReply
     val pf =
-      if (reply) {
+      if (needReply) {
         endpoint.receiveAndReply(new RpcCallContext {
           override def sendFailure(e: Throwable): Unit = {
             _sender ! AkkaFailure(e)
@@ -185,7 +188,7 @@ private[spark] class AkkaRpcEnv private[akka] (
       }
     } catch {
       case NonFatal(e) =>
-        if (reply) {
+        if (needReply) {
           // If the sender asks a reply, we should send the error back to the sender
           _sender ! AkkaFailure(e)
         } else {
@@ -241,7 +244,7 @@ private[spark] class AkkaRpcEnv private[akka] (
     actorSystem.awaitTermination()
   }
 
-  override def toString = s"${getClass.getSimpleName}($actorSystem)"
+  override def toString: String = s"${getClass.getSimpleName}($actorSystem)"
 }
 
 private[spark] class AkkaRpcEnvFactory extends RpcEnvFactory {
@@ -308,9 +311,9 @@ private[akka] class AkkaRpcEndpointRef(
 /**
  * A wrapper to `message` so that the receiver knows if the sender expects a reply.
  * @param message
- * @param reply if the sender expects a reply message
+ * @param needReply if the sender expects a reply message
  */
-private[akka] case class AkkaMessage(message: Any, reply: Boolean)
+private[akka] case class AkkaMessage(message: Any, needReply: Boolean)
 
 /**
  * A reply with the failure error from the receiver to the sender

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

@@ -192,7 +192,6 @@ abstract class RpcEnvSuite extends FunSuite with BeforeAndAfterAll {
       }
     }
     val rpcEndpointRef = env.setupEndpoint("start-stop-test", endpoint)
-    rpcEndpointRef.send("message")
     env.stop(rpcEndpointRef)
     stopLatch.await(10, TimeUnit.SECONDS)
     assert(List("start", "stop") === calledMethods)