make asyncCalllPermitsTimeoutMs configurable

Author: hfutatzhanghb

Diff for: hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonConfigurationKeys.java

@@ -355,6 +355,9 @@ public class CommonConfigurationKeys extends CommonConfigurationKeysPublic {
   public static final String  IPC_CLIENT_ASYNC_CALLS_MAX_KEY =
       "ipc.client.async.calls.max";
   public static final int     IPC_CLIENT_ASYNC_CALLS_MAX_DEFAULT = 100;
+  public static final String  IPC_CLIENT_ASYNC_CALLS_PERMITS_ACQUIRE_TIMEOUT_MS_KEY =
+      "ipc.client.async.calls.permits.acquire.timeout.ms";
+  public static final int     IPC_CLIENT_ASYNC_CALLS_PERMITS_ACQUIRE_TIMEOUT_MS_DEFAULT = 1000;
   public static final String  IPC_CLIENT_FALLBACK_TO_SIMPLE_AUTH_ALLOWED_KEY = "ipc.client.fallback-to-simple-auth-allowed";
   public static final boolean IPC_CLIENT_FALLBACK_TO_SIMPLE_AUTH_ALLOWED_DEFAULT = false;
 

Diff for: hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java

@@ -204,6 +204,7 @@ public static Object getExternalHandler() {
   private final byte[] clientId;
   private final int maxAsyncCalls;
   private final AtomicInteger asyncCallCounter = new AtomicInteger(0);
+  private final int asyncCalllPermitsTimeoutMs;
   private final ConcurrentMap<ConnectionId, Semaphore> asyncCallCounters =
       new ConcurrentHashMap<>();
 
@@ -1382,6 +1383,9 @@ public Client(Class<? extends Writable> valueClass, Configuration conf,
     this.maxAsyncCalls = conf.getInt(
         CommonConfigurationKeys.IPC_CLIENT_ASYNC_CALLS_MAX_KEY,
         CommonConfigurationKeys.IPC_CLIENT_ASYNC_CALLS_MAX_DEFAULT);
+    this.asyncCalllPermitsTimeoutMs = conf.getInt(
+        CommonConfigurationKeys.IPC_CLIENT_ASYNC_CALLS_PERMITS_ACQUIRE_TIMEOUT_MS_KEY,
+        CommonConfigurationKeys.IPC_CLIENT_ASYNC_CALLS_PERMITS_ACQUIRE_TIMEOUT_MS_DEFAULT);
   }
 
   /**
@@ -1479,8 +1483,8 @@ private void checkAsyncCall(ConnectionId remoteId) throws IOException {
         if (LOG.isDebugEnabled()) {
           LOG.debug("Acquiring lock for connectionId {}", remoteId);
         }
-        // TODO timeout param configurable.
-        boolean isAcquired = asyncPermits.tryAcquire(1000, TimeUnit.MILLISECONDS);
+        boolean isAcquired = asyncPermits.tryAcquire(asyncCalllPermitsTimeoutMs,
+            TimeUnit.MILLISECONDS);
         if (!isAcquired) {
           String errMsg = String.format(
               "Exceeded limit of max asynchronous calls: %d, " +