HDDS-16382. Dedicated SCM client RPC timeout and retry for OM request critical path

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java

@@ -710,7 +710,8 @@ public final class OzoneConfigKeys {
       "hdds.scmclient.max.retry.timeout";
   public static final String HDDS_SCM_CLIENT_FAILOVER_MAX_RETRY =
       "hdds.scmclient.failover.max.retry";
-
+  public static final String HDDS_SCM_CLIENT_FAILOVER_RETRY_INTERVAL =
+      "hdds.scmclient.failover.retry.interval";
   public static final String OZONE_XCEIVER_CLIENT_METRICS_PERCENTILES_INTERVALS_SECONDS_KEY =
       "ozone.xceiver.client.metrics.percentiles.intervals.seconds";
 

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -602,6 +602,60 @@
       This config overrides Hadoop configuration "ipc.server.read.threadpool.size" for Ozone Manager.
     </description>
   </property>
+  <property>
+    <name>ozone.om.scmclient.location.rpc.timeout</name>
+    <value>30s</value>
+    <description>
+      RPC response timeout for OM's SCM block and container location
+      clients.
+    </description>
+  </property>
+  <property>
+    <name>ozone.om.scmclient.location.ipc.connect.timeout</name>
+    <value>5s</value>
+    <description>
+      TCP connect timeout for OM's SCM block and container location
+      clients. The value accepts a duration suffix such as ms, s, or m and is
+      converted to milliseconds for Hadoop IPC.
+    </description>
+  </property>
+  <property>
+    <name>ozone.om.scmclient.location.ipc.connect.max.retries.on.timeouts</name>
+    <value>0</value>
+    <description>
+      Maximum TCP connect retries after a timeout for OM's SCM block and
+      container location clients. A value of 0 immediately fails over to
+      another SCM peer.
+    </description>
+  </property>
+  <property>
+    <name>ozone.om.scmclient.location.failover.max.retry</name>
+    <value>3</value>
+    <description>
+      Maximum failover retries for OM's SCM block and container location
+      clients. The effective retry count is the larger of this value and the
+      quotient of ozone.om.scmclient.location.max.retry.timeout divided by
+      ozone.om.scmclient.location.failover.retry.interval.
+    </description>
+  </property>
+  <property>
+    <name>ozone.om.scmclient.location.max.retry.timeout</name>
+    <value>6s</value>
+    <description>
+      Duration used with ozone.om.scmclient.location.failover.retry.interval
+      to derive the failover retry count for OM's SCM block and container
+      location clients. This is not a limit on the total duration of an RPC
+      call.
+    </description>
+  </property>
+  <property>
+    <name>ozone.om.scmclient.location.failover.retry.interval</name>
+    <value>2s</value>
+    <description>
+      Delay between failover retries for OM's SCM block and container location
+      clients.
+    </description>
+  </property>
   <property>
     <name>ozone.om.http-address</name>
     <value>0.0.0.0:9874</value>

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/proxy/SCMBlockLocationFailoverProxyProvider.java

@@ -17,6 +17,7 @@
 
 package org.apache.hadoop.hdds.scm.proxy;
 
+import javax.net.SocketFactory;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
 import org.apache.hadoop.hdds.scm.ha.SCMNodeInfo;
 import org.apache.hadoop.hdds.scm.protocolPB.ScmBlockLocationProtocolPB;
@@ -32,7 +33,12 @@ public class SCMBlockLocationFailoverProxyProvider extends
       LoggerFactory.getLogger(SCMBlockLocationFailoverProxyProvider.class);
 
   public SCMBlockLocationFailoverProxyProvider(ConfigurationSource conf) {
-    super(ScmBlockLocationProtocolPB.class, conf, null);
+    this(conf, null);
+  }
+
+  public SCMBlockLocationFailoverProxyProvider(ConfigurationSource conf,
+      SocketFactory socketFactory) {
+    super(ScmBlockLocationProtocolPB.class, conf, null, socketFactory);
   }
 
   @Override
@@ -45,4 +51,3 @@ protected String getProtocolAddress(SCMNodeInfo scmNodeInfo) {
     return scmNodeInfo.getBlockClientAddress();
   }
 }
-

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/proxy/SCMContainerLocationFailoverProxyProvider.java

@@ -17,6 +17,8 @@
 
 package org.apache.hadoop.hdds.scm.proxy;
 
+import jakarta.annotation.Nullable;
+import javax.net.SocketFactory;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
 import org.apache.hadoop.hdds.scm.ha.SCMNodeInfo;
 import org.apache.hadoop.hdds.scm.protocolPB.StorageContainerLocationProtocolPB;
@@ -32,14 +34,14 @@ public class SCMContainerLocationFailoverProxyProvider extends
   private static final Logger LOG =
       LoggerFactory.getLogger(SCMContainerLocationFailoverProxyProvider.class);
 
-  /**
-   * Construct SCMContainerLocationFailoverProxyProvider.
-   * If userGroupInformation is not null, use the passed ugi, else obtain
-   * from {@link UserGroupInformation#getCurrentUser()}
-   */
   public SCMContainerLocationFailoverProxyProvider(ConfigurationSource conf,
       UserGroupInformation userGroupInformation) {
-    super(StorageContainerLocationProtocolPB.class, conf, userGroupInformation);
+    this(conf, userGroupInformation, null);
+  }
+
+  public SCMContainerLocationFailoverProxyProvider(ConfigurationSource conf,
+      UserGroupInformation userGroupInformation, @Nullable SocketFactory socketFactory) {
+    super(StorageContainerLocationProtocolPB.class, conf, userGroupInformation, socketFactory);
   }
 
   @Override

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/proxy/SCMFailoverProxyProviderBase.java

@@ -19,6 +19,7 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.net.InetAddresses;
+import jakarta.annotation.Nullable;
 import java.io.IOException;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
@@ -32,6 +33,7 @@
 import java.util.Optional;
 import java.util.OptionalInt;
 import java.util.stream.Collectors;
+import javax.net.SocketFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hdds.HddsUtils;
 import org.apache.hadoop.hdds.conf.ConfigurationException;
@@ -48,6 +50,7 @@
 import org.apache.hadoop.ipc_.ProtobufRpcEngine;
 import org.apache.hadoop.ipc_.RPC;
 import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.net.StandardSocketFactory;
 import org.apache.hadoop.ozone.OzoneConfigKeys;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.slf4j.Logger;
@@ -88,6 +91,8 @@ public abstract class SCMFailoverProxyProviderBase<T> implements FailoverProxyPr
   private final long retryInterval;
 
   private final UserGroupInformation ugi;
+  @Nullable
+  private final SocketFactory socketFactory;
 
   private String updatedLeaderNodeID = null;
 
@@ -99,15 +104,37 @@ public abstract class SCMFailoverProxyProviderBase<T> implements FailoverProxyPr
    */
   private final boolean resolveOnFailureEnabled;
 
+  public SCMFailoverProxyProviderBase(Class<T> protocol, ConfigurationSource conf,
+      UserGroupInformation userGroupInformation) {
+    this(protocol, conf, userGroupInformation, null);
+  }
+
   /**
    * Construct SCMFailoverProxyProviderBase.
+   * <p>
    * If userGroupInformation is not null, use the passed ugi, else obtain
    * from {@link UserGroupInformation#getCurrentUser()}
+   * <p>
+   * Additionally, we can optionally specify {@link SocketFactory} which can be used for use
+   * case when two clients require different timeout configuration. This is because Hadoop
+   * {@link org.apache.hadoop.ipc.ClientCache} uses {@link SocketFactory} as the cache key.
+   * The default {@link StandardSocketFactory#hashCode()} hashes to the class name, which
+   * means that all clients with the same socket factory share the same {@link org.apache.hadoop.ipc.Client}.
+   * However, sometimes we want to have separate client with different timeout. In that case,
+   * we need to have a separate {@link SocketFactory}.
+   * @param protocol SCM RPC protocol
+   * @param conf Configuration
+   * @param userGroupInformation custom ugi for this client. If not specified the implementation defaults to
+   *                             {@link UserGroupInformation#getCurrentUser()}
+   * @param socketFactory custom socket factory for the client. If not specified, the implement defaults to
+   *                             {@link NetUtils#getDefaultSocketFactory(Configuration)}
    */
   public SCMFailoverProxyProviderBase(Class<T> protocol, ConfigurationSource conf,
-      UserGroupInformation userGroupInformation) {
+      UserGroupInformation userGroupInformation,
+      @Nullable SocketFactory socketFactory) {
     this.protocolClass = protocol;
     this.conf = conf;
+    this.socketFactory = socketFactory;
 
     if (userGroupInformation == null) {
       try {
@@ -496,10 +523,12 @@ private T createSCMProxy(InetSocketAddress scmAddress) throws IOException {
     // retries on the same SCM in case of connection exception. This retry
     // policy essentially results in TRY_ONCE_THEN_FAIL.
     RetryPolicy connectionRetryPolicy = RetryPolicies.failoverOnNetworkException(0);
+    SocketFactory sockFactory = socketFactory == null ?
+        NetUtils.getDefaultSocketFactory(hadoopConf) : socketFactory;
     return RPC.getProtocolProxy(
         protocolClass,
         scmVersion, scmAddress, ugi,
-        hadoopConf, NetUtils.getDefaultSocketFactory(hadoopConf),
+        hadoopConf, sockFactory,
         (int)scmClientConfig.getRpcTimeOut(), connectionRetryPolicy).getProxy();
   }
 

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HAUtils.java

@@ -40,6 +40,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
+import javax.net.SocketFactory;
 import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.hdds.HddsConfigKeys;
 import org.apache.hadoop.hdds.HddsUtils;
@@ -131,31 +132,37 @@ public static boolean addSCM(OzoneConfiguration conf, AddSCMRequest request,
    */
   public static ScmBlockLocationProtocol getScmBlockClient(
       OzoneConfiguration conf) {
+    return getScmBlockClient(conf, null);
+  }
+
+  public static ScmBlockLocationProtocol getScmBlockClient(
+      OzoneConfiguration conf, SocketFactory socketFactory) {
     ScmBlockLocationProtocolClientSideTranslatorPB scmBlockLocationClient =
         new ScmBlockLocationProtocolClientSideTranslatorPB(
-            new SCMBlockLocationFailoverProxyProvider(conf), conf);
+            new SCMBlockLocationFailoverProxyProvider(conf, socketFactory),
+            conf);
     return TracingUtil
         .createProxy(scmBlockLocationClient, ScmBlockLocationProtocol.class,
             conf);
   }
 
   public static StorageContainerLocationProtocol getScmContainerClient(
       ConfigurationSource conf) {
-    SCMContainerLocationFailoverProxyProvider proxyProvider =
-        new SCMContainerLocationFailoverProxyProvider(conf, null);
-    StorageContainerLocationProtocol scmContainerClient =
-        TracingUtil.createProxy(
-            new StorageContainerLocationProtocolClientSideTranslatorPB(
-                proxyProvider), StorageContainerLocationProtocol.class, conf);
-    return scmContainerClient;
+    return getScmContainerClient(conf, null, null);
   }
 
   @VisibleForTesting
   public static StorageContainerLocationProtocol getScmContainerClient(
       ConfigurationSource conf, UserGroupInformation userGroupInformation) {
+    return getScmContainerClient(conf, userGroupInformation, null);
+  }
+
+  public static StorageContainerLocationProtocol getScmContainerClient(
+      ConfigurationSource conf, UserGroupInformation userGroupInformation,
+      SocketFactory socketFactory) {
     SCMContainerLocationFailoverProxyProvider proxyProvider =
         new SCMContainerLocationFailoverProxyProvider(conf,
-            userGroupInformation);
+            userGroupInformation, socketFactory);
     StorageContainerLocationProtocol scmContainerClient =
         TracingUtil.createProxy(
             new StorageContainerLocationProtocolClientSideTranslatorPB(

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/IdentitySocketFactory.java

@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.om;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.Socket;
+import javax.net.SocketFactory;
+import org.apache.hadoop.ipc.ClientCache;
+import org.apache.hadoop.net.StandardSocketFactory;
+
+/**
+ * Delegates socket creation while retaining identity-based equality.
+ *
+ * <p>Hadoop's {@link ClientCache} uses a {@link SocketFactory} as the cache key
+ * and deduplicates IPC clients when socket factories compare equal. Wrapping a
+ * factory in this class provides a distinct cache key so a client can use
+ * configuration that must not be shared with other Hadoop IPC clients.
+ *
+ * <p>Unlike {@link StandardSocketFactory}, this class intentionally does not
+ * implement {@link Object#equals(Object)} or {@link Object#hashCode()}.
+ * Retaining {@link Object}'s identity-based implementations ensures that each
+ * instance remains a distinct Hadoop IPC client cache key.
+ */
+final class IdentitySocketFactory extends SocketFactory {
+  private final SocketFactory delegate;
+
+  IdentitySocketFactory(SocketFactory delegate) {
+    this.delegate = delegate;
+  }
+
+  @Override
+  public Socket createSocket() throws IOException {
+    return delegate.createSocket();
+  }
+
+  @Override
+  public Socket createSocket(String host, int port) throws IOException {
+    return delegate.createSocket(host, port);
+  }
+
+  @Override
+  public Socket createSocket(String host, int port, InetAddress localHost,
+      int localPort) throws IOException {
+    return delegate.createSocket(host, port, localHost, localPort);
+  }
+
+  @Override
+  public Socket createSocket(InetAddress host, int port) throws IOException {
+    return delegate.createSocket(host, port);
+  }
+
+  @Override
+  public Socket createSocket(InetAddress address, int port,
+      InetAddress localAddress, int localPort) throws IOException {
+    return delegate.createSocket(address, port, localAddress, localPort);
+  }
+}

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OMConfigKeys.java

@@ -55,6 +55,34 @@ public final class OMConfigKeys {
   public static final int OZONE_OM_DB_MAX_OPEN_FILES_DEFAULT
       = -1;
 
+  public static final String OZONE_OM_SCM_LOCATION_CLIENT_RPC_TIMEOUT =
+      "ozone.om.scmclient.location.rpc.timeout";
+  public static final String OZONE_OM_SCM_LOCATION_CLIENT_RPC_TIMEOUT_DEFAULT =
+      "30s";
+  public static final String OZONE_OM_SCM_LOCATION_CLIENT_IPC_CONNECT_TIMEOUT =
+      "ozone.om.scmclient.location.ipc.connect.timeout";
+  public static final String
+      OZONE_OM_SCM_LOCATION_CLIENT_IPC_CONNECT_TIMEOUT_DEFAULT =
+      "5s";
+  public static final String
+      OZONE_OM_SCM_LOCATION_CLIENT_IPC_CONNECT_TIMEOUT_RETRIES =
+      "ozone.om.scmclient.location.ipc.connect.max.retries.on.timeouts";
+  public static final String
+      OZONE_OM_SCM_LOCATION_CLIENT_IPC_CONNECT_TIMEOUT_RETRIES_DEFAULT = "0";
+  public static final String OZONE_OM_SCM_LOCATION_CLIENT_FAILOVER_MAX_RETRY =
+      "ozone.om.scmclient.location.failover.max.retry";
+  public static final String
+      OZONE_OM_SCM_LOCATION_CLIENT_FAILOVER_MAX_RETRY_DEFAULT = "3";
+  public static final String OZONE_OM_SCM_LOCATION_CLIENT_MAX_RETRY_TIMEOUT =
+      "ozone.om.scmclient.location.max.retry.timeout";
+  public static final String
+      OZONE_OM_SCM_LOCATION_CLIENT_MAX_RETRY_TIMEOUT_DEFAULT = "6s";
+  public static final String
+      OZONE_OM_SCM_LOCATION_CLIENT_FAILOVER_RETRY_INTERVAL =
+      "ozone.om.scmclient.location.failover.retry.interval";
+  public static final String
+      OZONE_OM_SCM_LOCATION_CLIENT_FAILOVER_RETRY_INTERVAL_DEFAULT = "2s";
+
   public static final String OZONE_OM_INTERNAL_SERVICE_ID =
       "ozone.om.internal.service.id";