tested cluster & client mode

Author: John Lee

common/network-common/src/main/java/org/apache/spark/network/sasl/SaslRpcHandler.java

@@ -114,7 +114,7 @@ public void receive(TransportClient client, ByteBuffer message, RpcResponseCallb
     // method returns. This assumes that the code ensures, through other means, that no outbound
     // messages are being written to the channel while negotiation is still going on.
     if (saslServer.isComplete()) {
-      cient.setClientUser(saslServer.getUserName());
+      client.setClientUser(saslServer.getUserName());
       if (!SparkSaslServer.QOP_AUTH_CONF.equals(saslServer.getNegotiatedProperty(Sasl.QOP))) {
         logger.debug("SASL authentication successful for channel {}", client);
         complete(true);

core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala

@@ -544,7 +544,7 @@ private[netty] class RequestMessage(
     val senderAddress: RpcAddress,
     val receiver: NettyRpcEndpointRef,
     val content: Any,
-    val senderUserName: String = null){
+    val senderUserName: String = null) {
 
   /** Manually serialize [[RequestMessage]] to minimize the size. */
   def serialize(nettyEnv: NettyRpcEnv): ByteBuffer = {
@@ -590,7 +590,11 @@ private[netty] object RequestMessage {
     }
   }
 
-  def apply(nettyEnv: NettyRpcEnv, client: TransportClient, bytes: ByteBuffer): RequestMessage = {
+  def apply(
+      nettyEnv: NettyRpcEnv,
+      client: TransportClient,
+      bytes: ByteBuffer,
+      senderUserName: String = null): RequestMessage = {
     val bis = new ByteBufferInputStream(bytes)
     val in = new DataInputStream(bis)
     try {
@@ -602,7 +606,8 @@ private[netty] object RequestMessage {
         senderAddress,
         ref,
         // The remaining bytes in `bytes` are the message content.
-        nettyEnv.deserialize(client, bytes))
+        nettyEnv.deserialize(client, bytes),
+        senderUserName)
     } finally {
       in.close()
     }
@@ -653,12 +658,8 @@ private[netty] class NettyRpcHandler(
     val addr = client.getChannel().remoteAddress().asInstanceOf[InetSocketAddress]
     assert(addr != null)
     val clientAddr = RpcAddress(addr.getHostString, addr.getPort)
-    val requestMessage = RequestMessage(nettyEnv, client, message)
+    var requestMessage = RequestMessage(nettyEnv, client, message, client.getClientUser)
 
-    if (client.getClientUser != null) {
-      requestMessage = RequestMessage(requestMessage.senderAddress, requestMessage.receiver,
-        requestMessage.content, client.getClientUser)
-    }
     if (requestMessage.senderAddress == null) {
       // Create a new message with the socket address of the client as the sender.
       new RequestMessage(clientAddr, requestMessage.receiver, requestMessage.content,

resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnRMClient.scala

@@ -75,7 +75,7 @@ private[spark] class YarnRMClient extends Logging {
 
     logInfo("Registering the ApplicationMaster")
     synchronized {
-      var response = amClient.registerApplicationMaster(Utils.localHostName(), port, uiAddress)
+      var response = amClient.registerApplicationMaster(Utils.localHostName(), port, trackingUrl)
       registered = true
       masterkey = response.getClientToAMTokenMasterKey()
     }