Address comments

Author: aarondav

core/src/main/scala/org/apache/spark/network/netty/NettyBlockTransferService.scala

@@ -17,8 +17,6 @@
 
 package org.apache.spark.network.netty
 
-import org.apache.spark.network.util.TransportConf
-
 import scala.collection.JavaConversions._
 import scala.concurrent.{Future, Promise}
 
@@ -27,7 +25,7 @@ import org.apache.spark.network._
 import org.apache.spark.network.buffer.ManagedBuffer
 import org.apache.spark.network.client.{TransportClientBootstrap, RpcResponseCallback, TransportClientFactory}
 import org.apache.spark.network.netty.NettyMessages.{OpenBlocks, UploadBlock}
-import org.apache.spark.network.sasl.{SecretKeyHolder, SaslRpcHandler, SaslBootstrap}
+import org.apache.spark.network.sasl.{SaslRpcHandler, SaslClientBootstrap}
 import org.apache.spark.network.server._
 import org.apache.spark.network.shuffle.{BlockFetchingListener, OneForOneBlockFetcher}
 import org.apache.spark.serializer.JavaSerializer
@@ -56,7 +54,7 @@ class NettyBlockTransferService(conf: SparkConf, securityManager: SecurityManage
         (nettyRpcHandler, None)
       } else {
         (new SaslRpcHandler(nettyRpcHandler, securityManager),
-          Some(new SaslBootstrap(transportConf, conf.getAppId, securityManager)))
+          Some(new SaslClientBootstrap(transportConf, conf.getAppId, securityManager)))
       }
     }
     transportContext = new TransportContext(transportConf, rpcHandler)

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

@@ -130,6 +130,7 @@ public void initChannel(SocketChannel ch) {
     });
 
     // Connect to the remote server
+    long preConnect = System.currentTimeMillis();
     ChannelFuture cf = bootstrap.connect(address);
     if (!cf.awaitUninterruptibly(conf.connectionTimeoutMs())) {
       throw new RuntimeException(
@@ -142,25 +143,30 @@ public void initChannel(SocketChannel ch) {
     assert client != null : "Channel future completed successfully with null client";
 
     // Execute any client bootstraps synchronously before marking the Client as successful.
+    long preBootstrap = System.currentTimeMillis();
     logger.debug("Connection to {} successful, running bootstraps...", address);
     try {
       for (TransportClientBootstrap clientBootstrap : clientBootstraps) {
         clientBootstrap.doBootstrap(client);
       }
-    } catch (Exception e) { // catch Exception as the bootstrap may be written in Scala
-      logger.error("Exception while bootstrapping client", e);
+    } catch (Exception e) { // catch non-RuntimeExceptions too as bootstrap may be written in Scala
+      long bootstrapTime = System.currentTimeMillis() - preBootstrap;
+      logger.error("Exception while bootstrapping client after " + bootstrapTime + " ms", e);
       client.close();
       throw Throwables.propagate(e);
     }
+    long postBootstrap = System.currentTimeMillis();
 
-    logger.debug("Successfully executed {} bootstraps for {}", clientBootstraps.size(), address);
     // Successful connection & bootstrap -- in the event that two threads raced to create a client,
     // use the first one that was put into the connectionPool and close the one we made here.
     TransportClient oldClient = connectionPool.putIfAbsent(address, client);
     if (oldClient == null) {
+      logger.debug("Successfully created connection to {} after {} ms ({} ms spent in bootstraps)",
+        address, postBootstrap - preConnect, postBootstrap - preBootstrap);
       return client;
     } else {
-      logger.debug("Two clients were created concurrently, second one will be disposed.");
+      logger.debug("Two clients were created concurrently after {} ms, second will be disposed.",
+        postBootstrap - preConnect);
       client.close();
       return oldClient;
     }

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

@@ -30,14 +30,14 @@
  * Bootstraps a {@link TransportClient} by performing SASL authentication on the connection. The
  * server should be setup with a {@link SaslRpcHandler} with matching keys for the given appId.
  */
-public class SaslBootstrap implements TransportClientBootstrap {
-  private final Logger logger = LoggerFactory.getLogger(SaslBootstrap.class);
+public class SaslClientBootstrap implements TransportClientBootstrap {
+  private final Logger logger = LoggerFactory.getLogger(SaslClientBootstrap.class);
 
   private final TransportConf conf;
   private final String appId;
   private final SecretKeyHolder secretKeyHolder;
 
-  public SaslBootstrap(TransportConf conf, String appId, SecretKeyHolder secretKeyHolder) {
+  public SaslClientBootstrap(TransportConf conf, String appId, SecretKeyHolder secretKeyHolder) {
     this.conf = conf;
     this.appId = appId;
     this.secretKeyHolder = secretKeyHolder;
@@ -48,6 +48,7 @@ public SaslBootstrap(TransportConf conf, String appId, SecretKeyHolder secretKey
    * challenge-response tokens until we either successfully authenticate or throw an exception
    * due to mismatch.
    */
+  @Override
   public void doBootstrap(TransportClient client) {
     SparkSaslClient saslClient = new SparkSaslClient(appId, secretKeyHolder);
     try {

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

@@ -43,7 +43,7 @@
 public class SaslRpcHandler extends RpcHandler {
   private final Logger logger = LoggerFactory.getLogger(SaslRpcHandler.class);
 
-  /** RpcHandler we will delegate for authenticated connections. */
+  /** RpcHandler we will delegate to for authenticated connections. */
   private final RpcHandler delegate;
 
   /** Class which provides secret keys which are shared by server and client on a per-app basis. */

network/shuffle/src/test/java/org/apache/spark/network/sasl/SaslIntegrationSuite.java

@@ -96,7 +96,7 @@ public void afterEach() {
   public void testGoodClient() {
     clientFactory = context.createClientFactory(
       Lists.<TransportClientBootstrap>newArrayList(
-        new SaslBootstrap(conf, "app-id", new TestSecretKeyHolder("good-key"))));
+        new SaslClientBootstrap(conf, "app-id", new TestSecretKeyHolder("good-key"))));
 
     TransportClient client = clientFactory.createClient(TestUtils.getLocalHost(), server.getPort());
     String msg = "Hello, World!";
@@ -108,7 +108,7 @@ public void testGoodClient() {
   public void testBadClient() {
     clientFactory = context.createClientFactory(
       Lists.<TransportClientBootstrap>newArrayList(
-        new SaslBootstrap(conf, "app-id", new TestSecretKeyHolder("bad-key"))));
+        new SaslClientBootstrap(conf, "app-id", new TestSecretKeyHolder("bad-key"))));
 
     try {
       // Bootstrap should fail on startup.
@@ -146,7 +146,7 @@ public void testNoSaslServer() {
     TransportContext context = new TransportContext(conf, handler);
     clientFactory = context.createClientFactory(
       Lists.<TransportClientBootstrap>newArrayList(
-        new SaslBootstrap(conf, "app-id", new TestSecretKeyHolder("key"))));
+        new SaslClientBootstrap(conf, "app-id", new TestSecretKeyHolder("key"))));
     TransportServer server = context.createServer();
     try {
       clientFactory.createClient(TestUtils.getLocalHost(), server.getPort());