Fixed two NPEs.

Author: rxin

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

@@ -66,6 +66,9 @@ private class ClientPool {
     public ClientPool() {
       clients = new TransportClient[numConnectionsPerPeer];
       locks = new Object[numConnectionsPerPeer];
+      for (int i = 0; i < numConnectionsPerPeer; i++) {
+        locks[i] = new Object();
+      }
     }
   }
 
@@ -120,7 +123,8 @@ 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) {
-      clientPool = connectionPool.putIfAbsent(address, new ClientPool());
+      connectionPool.putIfAbsent(address, new ClientPool());
+      clientPool = connectionPool.get(address);
     }
 
     int clientIndex = rand.nextInt(numConnectionsPerPeer);