Code review feedback.

Author: rxin

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

@@ -43,6 +43,7 @@
 import org.apache.spark.network.TransportContext;
 import org.apache.spark.network.server.TransportChannelHandler;
 import org.apache.spark.network.util.IOMode;
+import org.apache.spark.network.util.JavaUtils;
 import org.apache.spark.network.util.NettyUtils;
 import org.apache.spark.network.util.TransportConf;
 
@@ -59,14 +60,14 @@
 public class TransportClientFactory implements Closeable {
 
   /** A simple data structure to track the pool of clients between two peer nodes. */
-  private class ClientPool {
+  private static class ClientPool {
     TransportClient[] clients;
     Object[] locks;
 
-    public ClientPool() {
-      clients = new TransportClient[numConnectionsPerPeer];
-      locks = new Object[numConnectionsPerPeer];
-      for (int i = 0; i < numConnectionsPerPeer; i++) {
+    public ClientPool(int size) {
+      clients = new TransportClient[size];
+      locks = new Object[size];
+      for (int i = 0; i < size; i++) {
         locks[i] = new Object();
       }
     }
@@ -123,7 +124,7 @@ public TransportClient createClient(String remoteHost, int remotePort) throws IO
     // Create the ClientPool if we don't have it yet.
     ClientPool clientPool = connectionPool.get(address);
     if (clientPool == null) {
-      connectionPool.putIfAbsent(address, new ClientPool());
+      connectionPool.putIfAbsent(address, new ClientPool(numConnectionsPerPeer));
       clientPool = connectionPool.get(address);
     }
 
@@ -219,11 +220,7 @@ public void close() {
         TransportClient client = clientPool.clients[i];
         if (client != null) {
           clientPool.clients[i] = null;
-          try {
-            client.close();
-          } catch (RuntimeException e) {
-            logger.warn("Ignoring exception during close", e);
-          }
+          JavaUtils.closeQuietly(client);
         }
       }
     }

network/common/src/test/java/org/apache/spark/network/TransportClientFactorySuite.java

@@ -65,38 +65,10 @@ public void tearDown() {
   /**
    * Request a bunch of clients to a single server to test
    * we create up to maxConnections of clients.
+   *
+   * If concurrent is true, create multiple threads to create clients in parallel.
    */
-  private void testClientReuse(final int maxConnections) throws IOException {
-    TransportConf conf = new TransportConf(new ConfigProvider() {
-      @Override
-      public String get(String name) {
-        if (name.equals("spark.shuffle.io.numConnectionsPerPeer")) {
-          return Integer.toString(maxConnections);
-        } else {
-          throw new NoSuchElementException();
-        }
-      }
-    });
-
-    RpcHandler rpcHandler = new NoOpRpcHandler();
-    TransportContext context = new TransportContext(conf, rpcHandler);
-    TransportClientFactory factory = context.createClientFactory();
-    HashSet<TransportClient> clients = new HashSet<TransportClient>();
-    for (int i = 0; i < maxConnections * 10; i++) {
-      TransportClient client = factory.createClient(TestUtils.getLocalHost(), server1.getPort());
-      assert(client.isActive());
-      clients.add(client);
-    }
-
-    assert(clients.size() == maxConnections);
-  }
-
-  /**
-   * Request a bunch of clients to a single server to test
-   * we create up to maxConnections of clients. This is a parallel
-   * version of testClientReuse.
-   */
-  private void testClientReuseConcurrent(final int maxConnections)
+  private void testClientReuse(final int maxConnections, boolean concurrent)
     throws IOException, InterruptedException {
     TransportConf conf = new TransportConf(new ConfigProvider() {
       @Override
@@ -133,7 +105,12 @@ public void run() {
           }
         }
       };
-      attempts[i].run();
+
+      if (concurrent) {
+        attempts[i].start();
+      } else {
+        attempts[i].run();
+      }
     }
 
     // Wait until all the threads complete.
@@ -143,22 +120,26 @@ public void run() {
 
     assert(failed.get() == 0);
     assert(clients.size() == maxConnections);
+
+    for (TransportClient client : clients) {
+      client.close();
+    }
   }
 
   @Test
-  public void reuseClientsUpToConfigVariable() throws IOException {
-    testClientReuse(1);
-    testClientReuse(2);
-    testClientReuse(3);
-    testClientReuse(4);
+  public void reuseClientsUpToConfigVariable() throws Exception {
+    testClientReuse(1, false);
+    testClientReuse(2, false);
+    testClientReuse(3, false);
+    testClientReuse(4, false);
   }
 
   @Test
   public void reuseClientsUpToConfigVariableConcurrent() throws Exception {
-    testClientReuseConcurrent(1);
-    testClientReuseConcurrent(2);
-    testClientReuseConcurrent(3);
-    testClientReuseConcurrent(4);
+    testClientReuse(1, true);
+    testClientReuse(2, true);
+    testClientReuse(3, true);
+    testClientReuse(4, true);
   }
 
   @Test