Cleanup

network/shuffle/src/main/java/org/apache/spark/network/sasl/SaslBootstrap.java

@@ -55,12 +55,10 @@ public void doBootstrap(TransportClient client) {
 
       while (!saslClient.isComplete()) {
         SaslMessage msg = new SaslMessage(appId, payload);
-        logger.info("Sending msg {} {}", appId, payload.length);
         ByteBuf buf = Unpooled.buffer(msg.encodedLength());
         msg.encode(buf);
 
         byte[] response = client.sendRpcSync(buf.array(), conf.saslRTTimeout());
-        logger.info("Got response {} {}", appId, response.length);
         payload = saslClient.response(response);
       }
     } finally {

network/shuffle/src/main/java/org/apache/spark/network/sasl/SaslMessage.java

@@ -25,7 +25,7 @@
 /**
  * Encodes a Sasl-related message which is attempting to authenticate using some credentials tagged
  * with the given appId. This appId allows a single SaslRpcHandler to multiplex different
- * applications who may be using different sets of credentials.
+ * applications which may be using different sets of credentials.
  */
 class SaslMessage implements Encodable {
 
@@ -51,7 +51,7 @@ public void encode(ByteBuf buf) {
     buf.writeByte(TAG_BYTE);
     byte[] idBytes = appId.getBytes(Charsets.UTF_8);
     buf.writeInt(idBytes.length);
-    buf.writeBytes(appId.getBytes(Charsets.UTF_8));
+    buf.writeBytes(idBytes);
     buf.writeInt(payload.length);
     buf.writeBytes(payload);
   }

network/shuffle/src/main/java/org/apache/spark/network/sasl/SaslRpcHandler.java

@@ -43,10 +43,13 @@
 public class SaslRpcHandler extends RpcHandler {
   private final Logger logger = LoggerFactory.getLogger(SaslRpcHandler.class);
 
+  /** RpcHandler we will delegate for authenticated connections. */
   private final RpcHandler delegate;
+
+  /** Class which provides secret keys which are shared by server and client on a per-app basis. */
   private final SecretKeyHolder secretKeyHolder;
 
-  // TODO: Invalidate channels that have closed!
+  /** Maps each channel to its SASL authentication state. */
   private final ConcurrentMap<TransportClient, SparkSaslServer> channelAuthenticationMap;
 
   public SaslRpcHandler(RpcHandler delegate, SecretKeyHolder secretKeyHolder) {
@@ -92,4 +95,3 @@ public void connectionTerminated(TransportClient client) {
     }
   }
 }
-

network/shuffle/src/main/java/org/apache/spark/network/sasl/SecretKeyHolder.java

@@ -18,7 +18,7 @@
 package org.apache.spark.network.sasl;
 
 /**
- * Interface for getting a secret key associated with some appId.
+ * Interface for getting a secret key associated with some application.
  */
 public interface SecretKeyHolder {
   /**

network/shuffle/src/main/java/org/apache/spark/network/sasl/SparkSaslServer.java

@@ -80,7 +80,6 @@ public SparkSaslServer(String secretKeyId, SecretKeyHolder secretKeyHolder) {
       this.saslServer = Sasl.createSaslServer(DIGEST, null, DEFAULT_REALM, SASL_PROPS,
         new DigestCallbackHandler());
     } catch (SaslException e) {
-      System.exit(0);
       throw Throwables.propagate(e);
     }
   }