[SPARK-21701][CORE] Enable RPC client to use SO_RCVBUF, SO_SNDBUF and SO_BACKLOG in SparkConf

common/network-common/src/main/java/org/apache/spark/network/client/TransportClientFactory.java

@@ -210,6 +210,18 @@ private TransportClient createClient(InetSocketAddress address)
       .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, conf.connectionTimeoutMs())
       .option(ChannelOption.ALLOCATOR, pooledAllocator);
 
+    if (conf.backLog() > 0) {
+      bootstrap.option(ChannelOption.SO_BACKLOG, conf.backLog());
+    }
+
+    if (conf.receiveBuf() > 0) {
+      bootstrap.option(ChannelOption.SO_RCVBUF, conf.receiveBuf());
+    }
+
+    if (conf.sendBuf() > 0) {
+      bootstrap.option(ChannelOption.SO_SNDBUF, conf.sendBuf());
+    }
+
     final AtomicReference<TransportClient> clientRef = new AtomicReference<>();
     final AtomicReference<Channel> channelRef = new AtomicReference<>();
 

common/network-common/src/main/java/org/apache/spark/network/protocol/MessageEncoder.java

@@ -29,6 +29,52 @@
 /**
  * Encoder used by the server side to encode server-to-client responses.
  * This encoder is stateless so it is safe to be shared by multiple threads.
+ * <p></p>
+ * The following is the wire format that RPC message used, it is a typical
+ * header+payload structure, while the header contains body length which will
+ * be an indication for {@link org.apache.spark.network.util.TransportFrameDecoder}
+ * to build a complete message out of byte array for downstream {@link ChannelHandler}
+ * to use.
+ * <p></p>
+ * The underlying network I/O handles {@link MessageWithHeader} to transport message
+ * between peers. Below shows how RPC message looks like. The header is
+ * {@link MessageWithHeader#header}
+ * and the body is {@link MessageWithHeader#body}.
+ * <pre>
+ * Byte/      0       |       1       |       2       |       3       |
+ *     /              |               |               |               |
+ *     |0 1 2 3 4 5 6 7|0 1 2 3 4 5 6 7|0 1 2 3 4 5 6 7|0 1 2 3 4 5 6 7|
+ *     +---------------+---------------+---------------+---------------+
+ *     0/ Header                                                       /
+ *     /                                                               /
+ *     /                                                               /
+ *     /                                                               /
+ *     +---------------+---------------+---------------+---------------+
+ *     /  Body (a.k.a payload)                                         /
+ *     +---------------+---------------+---------------+---------------+
+ * </pre>
+ * The detailed header wire format is shown as below. Header consists of
+ * <ul>
+ *     <li>1. frame length: the total byte size of header+payload,
+ *     the length is an Integer value, which means maximum frame size would
+ *     be 16MB</li>
+ *     <li>2. message type: which subclass of {@link Message} are wrapped,
+ *     the length is {@link Message.Type#encodedLength()}</li>
+ *     <li>3. encoded message: some {@link Message} may provide additional
+ *     information and being included in the header</li>
+ * </ul>
+ * <pre>
+ * Byte/      0        |       1       |       2       |       3       |
+ *     /               |               |               |               |
+ *     |0 1 2 3 4 5 6 7|0 1 2 3 4 5 6 7|0 1 2 3 4 5 6 7|0 1 2 3 4 5 6 7|
+ *     +---------------+---------------+---------------+---------------+
+ *    0| frame length                                                  |
+ *     +---------------+---------------+---------------+---------------+
+ *    4| message type  |                                               |
+ *     +---------------+       encoded message                         |
+ *     |                                                               |
+ *     +---------------+---------------+---------------+---------------+
+ * </pre>
  */
 @ChannelHandler.Sharable
 public final class MessageEncoder extends MessageToMessageEncoder<Message> {

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

@@ -190,6 +190,31 @@ abstract class RpcEnvSuite extends SparkFunSuite with BeforeAndAfterAll {
     }
   }
 
+  test("ask a non-existent endpoint remotely") {
+    env.setupEndpoint("ask-non-existent-remotely", new RpcEndpoint {
+      override val rpcEnv = env
+
+      override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {
+        case msg: String =>
+          context.reply(msg)
+      }
+    })
+
+    val anotherEnv = createRpcEnv(new SparkConf(), "remote", 0, clientMode = true)
+    try {
+      val e = intercept[SparkException] {
+        anotherEnv.setupEndpointRef(env.address, "Oops")
+      }
+      // The SparkException cause should be a RpcEndpointNotFoundException
+      assert(e.getCause.isInstanceOf[RpcEndpointNotFoundException])
+      assert(e.getCause.getMessage.contains("Cannot find endpoint"))
+      assert(e.getCause.getMessage.contains(env.address.toString))
+    } finally {
+      anotherEnv.shutdown()
+      anotherEnv.awaitTermination()
+    }
+  }
+
   test("onStart and onStop") {
     val stopLatch = new CountDownLatch(1)
     val calledMethods = mutable.ArrayBuffer[String]()

core/src/test/scala/org/apache/spark/rpc/netty/NettyRpcEnvSuite.scala

@@ -35,15 +35,6 @@ class NettyRpcEnvSuite extends RpcEnvSuite with MockitoSugar {
     new NettyRpcEnvFactory().create(config)
   }
 
-  test("non-existent endpoint") {
-    val uri = RpcEndpointAddress(env.address, "nonexist-endpoint").toString
-    val e = intercept[SparkException] {
-      env.setupEndpointRef(env.address, "nonexist-endpoint")
-    }
-    assert(e.getCause.isInstanceOf[RpcEndpointNotFoundException])
-    assert(e.getCause.getMessage.contains(uri))
-  }
-
   test("advertise address different from bind address") {
     val sparkConf = new SparkConf()
     val config = RpcEnvConfig(sparkConf, "test", "localhost", "example.com", 0,