HDFS-13274. Modify patch based on comment

Author: zengqiang.xu

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/ConnectionContext.java

@@ -20,6 +20,7 @@
 import java.net.InetSocketAddress;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hdfs.NameNodeProxiesClient.ProxyAndInfo;
 import org.apache.hadoop.ipc.RPC;
 import org.apache.hadoop.util.Time;
@@ -56,9 +57,11 @@ public class ConnectionContext {
   /** The maximum number of requests that this connection can handle concurrently. **/
   private final int maxConcurrencyPerConn;
 
-  public ConnectionContext(ProxyAndInfo<?> connection, int maxConcurrencyPerConn) {
+  public ConnectionContext(ProxyAndInfo<?> connection, Configuration conf) {
     this.client = connection;
-    this.maxConcurrencyPerConn = maxConcurrencyPerConn;
+    this.maxConcurrencyPerConn = conf.getInt(
+        RBFConfigKeys.DFS_ROUTER_MAX_CONCURRENCY_PER_CONNECTION_KEY,
+        RBFConfigKeys.DFS_ROUTER_MAX_CONCURRENCY_PER_CONNECTION_DEFAULT);
   }
 
   /**

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/ConnectionPool.java

@@ -109,9 +109,6 @@ public class ConnectionPool {
   /** Enable using multiple physical socket or not. **/
   private final boolean enableMultiSocket;
 
-  /** Max Concurrency of each connection. */
-  private final int maxConcurrencyPerConn;
-
   /** Map for the protocols and their protobuf implementations. */
   private final static Map<Class<?>, ProtoImpl> PROTO_MAP = new HashMap<>();
   static {
@@ -159,9 +156,6 @@ protected ConnectionPool(Configuration config, String address,
     this.enableMultiSocket = conf.getBoolean(
         RBFConfigKeys.DFS_ROUTER_NAMENODE_ENABLE_MULTIPLE_SOCKET_KEY,
         RBFConfigKeys.DFS_ROUTER_NAMENODE_ENABLE_MULTIPLE_SOCKET_DEFAULT);
-    this.maxConcurrencyPerConn = conf.getInt(
-        RBFConfigKeys.DFS_ROUTER_MAX_CONCURRENCY_PER_CONNECTION_KEY,
-        RBFConfigKeys.DFS_ROUTER_MAX_CONCURRENCY_PER_CONNECTION_DEFAULT);
 
     // Add minimum connections to the pool
     for (int i = 0; i < this.minSize; i++) {
@@ -403,8 +397,7 @@ public String getJSON() {
   public ConnectionContext newConnection() throws IOException {
     return newConnection(this.conf, this.namenodeAddress,
         this.ugi, this.protocol, this.enableMultiSocket,
-        this.socketIndex.incrementAndGet(),
-        this.maxConcurrencyPerConn);
+        this.socketIndex.incrementAndGet());
   }
 
   /**
@@ -419,16 +412,13 @@ public ConnectionContext newConnection() throws IOException {
    * @param ugi User context.
    * @param proto Interface of the protocol.
    * @param enableMultiSocket Enable multiple socket or not.
-   * @param maxConcurrencyPerConn The maximum number of requests that
-   *                              this connection can handle concurrently.
    * @return proto for the target ClientProtocol that contains the user's
    *         security context.
    * @throws IOException If it cannot be created.
    */
   protected static <T> ConnectionContext newConnection(Configuration conf,
       String nnAddress, UserGroupInformation ugi, Class<T> proto,
-      boolean enableMultiSocket, int socketIndex,
-      int maxConcurrencyPerConn) throws IOException {
+      boolean enableMultiSocket, int socketIndex) throws IOException {
     if (!PROTO_MAP.containsKey(proto)) {
       String msg = "Unsupported protocol for connection to NameNode: "
           + ((proto != null) ? proto.getName() : "null");
@@ -467,7 +457,7 @@ protected static <T> ConnectionContext newConnection(Configuration conf,
     Text dtService = SecurityUtil.buildTokenService(socket);
 
     ProxyAndInfo<T> clientProxy = new ProxyAndInfo<T>(client, dtService, socket);
-    return new ConnectionContext(clientProxy, maxConcurrencyPerConn);
+    return new ConnectionContext(clientProxy, conf);
   }
 
   private static <T> T newProtoClient(Class<T> proto, ProtoImpl classes,

hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestConnectionManager.java

@@ -343,7 +343,7 @@ private void testConnectionCleanup(float ratio, int totalConns,
     addConnectionsToPool(pool, totalConns - 1, activeConns - 1);
 
     // There are activeConn connections.
-    // We can clean up the pool
+    // We can cleanup the pool
     tmpConnManager.cleanup(pool);
     assertEquals(leftConns, pool.getNumConnections());
 
@@ -356,6 +356,6 @@ public void testUnsupportedProtoExceptionMsg() throws Exception {
         "Unsupported protocol for connection to NameNode: "
             + TestConnectionManager.class.getName(),
         () -> ConnectionPool.newConnection(conf, TEST_NN_ADDRESS, TEST_USER1,
-            TestConnectionManager.class, false, 0, 1));
+            TestConnectionManager.class, false, 0));
   }
 }