Improved operation interceptor instantiation (not yet merged upstream) (#210)

Co-authored-by: Ray Mattingly <rmattingly@hubspot.com>

Author: 2 people

hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionImplementation.java

@@ -257,8 +257,6 @@ public class ConnectionImplementation implements ClusterConnection, Closeable {
 
   private final RetryingCallerInterceptor interceptor;
 
-  private final OperationInterceptorFactory operationInterceptorFactory;
-
   /**
    * Cluster registry of basic info such as clusterid and meta region location.
    */
@@ -339,7 +337,6 @@ public class ConnectionImplementation implements ClusterConnection, Closeable {
 
     this.stats = ServerStatisticTracker.create(conf);
     this.interceptor = new RetryingCallerInterceptorFactory(conf).build();
-    this.operationInterceptorFactory = createOperationInterceptorFactory(conf);
 
     this.backoffPolicy = ClientBackoffPolicyFactory.create(conf);
 
@@ -373,7 +370,7 @@ public class ConnectionImplementation implements ClusterConnection, Closeable {
         connectionAttributes);
       this.rpcControllerFactory = RpcControllerFactory.instantiate(conf);
       this.rpcCallerFactory = RpcRetryingCallerFactory.instantiate(conf, connectionConfig,
-        interceptor, this.stats, this.metrics, operationInterceptorFactory);
+        interceptor, this.stats, this.metrics);
       this.asyncProcess = new AsyncProcess(this, conf, rpcCallerFactory, rpcControllerFactory);
 
       // Do we publish the status?
@@ -2344,7 +2341,7 @@ public TableState getTableState(TableName tableName) throws IOException {
   @Override
   public RpcRetryingCallerFactory getNewRpcRetryingCallerFactory(Configuration conf) {
     return RpcRetryingCallerFactory.instantiate(conf, connectionConfig, this.interceptor,
-      this.stats, metrics, createOperationInterceptorFactory(conf));
+      this.stats, metrics);
   }
 
   @Override
@@ -2422,18 +2419,4 @@ public String getClusterId() {
     }
     return null;
   }
-
-  private static OperationInterceptorFactory createOperationInterceptorFactory(Configuration conf) {
-    String clazz = conf.get(OperationInterceptorFactory.HBASE_CLIENT_OPERATION_INTERCEPTOR_IMPL);
-    if (clazz == null || clazz.isEmpty()) {
-      return OperationInterceptorFactory.NO_OP;
-    }
-    try {
-      Class<? extends OperationInterceptorFactory> factoryClass =
-        conf.getClassByName(clazz).asSubclass(OperationInterceptorFactory.class);
-      return ReflectionUtils.newInstance(factoryClass, conf);
-    } catch (ClassNotFoundException e) {
-      throw new RuntimeException("Failed to load OperationInterceptorFactory class: " + clazz, e);
-    }
-  }
 }

hbase-client/src/main/java/org/apache/hadoop/hbase/client/RpcRetryingCallerFactory.java

@@ -21,13 +21,17 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.util.ReflectionUtils;
 import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Factory to create an {@link RpcRetryingCaller}
  */
 @InterfaceAudience.Private
 public class RpcRetryingCallerFactory {
 
+  private static final Logger LOG = LoggerFactory.getLogger(RpcRetryingCallerFactory.class);
+
   /** Configuration key for a custom {@link RpcRetryingCaller} */
   public static final String CUSTOM_CALLER_CONF_KEY = "hbase.rpc.callerfactory.class";
   private final ConnectionConfiguration connectionConf;
@@ -38,12 +42,12 @@ public class RpcRetryingCallerFactory {
 
   public RpcRetryingCallerFactory(Configuration conf, ConnectionConfiguration connectionConf) {
     this(conf, connectionConf, RetryingCallerInterceptorFactory.NO_OP_INTERCEPTOR, null,
-      OperationInterceptorFactory.NO_OP);
+      createOperationInterceptorFactory(conf));
   }
 
   public RpcRetryingCallerFactory(Configuration conf, ConnectionConfiguration connectionConf,
     RetryingCallerInterceptor interceptor, MetricsConnection metrics) {
-    this(conf, connectionConf, interceptor, metrics, OperationInterceptorFactory.NO_OP);
+    this(conf, connectionConf, interceptor, metrics, createOperationInterceptorFactory(conf));
   }
 
   public RpcRetryingCallerFactory(Configuration conf, ConnectionConfiguration connectionConf,
@@ -90,21 +94,23 @@ public static RpcRetryingCallerFactory instantiate(Configuration configuration,
   public static RpcRetryingCallerFactory instantiate(Configuration configuration,
     ConnectionConfiguration connectionConf, MetricsConnection metrics) {
     return instantiate(configuration, connectionConf,
-      RetryingCallerInterceptorFactory.NO_OP_INTERCEPTOR, null, metrics);
+      RetryingCallerInterceptorFactory.NO_OP_INTERCEPTOR, null, metrics,
+      createOperationInterceptorFactory(configuration));
   }
 
   public static RpcRetryingCallerFactory instantiate(Configuration configuration,
     ConnectionConfiguration connectionConf, ServerStatisticTracker stats,
     MetricsConnection metrics) {
     return instantiate(configuration, connectionConf,
-      RetryingCallerInterceptorFactory.NO_OP_INTERCEPTOR, stats, metrics);
+      RetryingCallerInterceptorFactory.NO_OP_INTERCEPTOR, stats, metrics,
+      createOperationInterceptorFactory(configuration));
   }
 
   public static RpcRetryingCallerFactory instantiate(Configuration configuration,
     ConnectionConfiguration connectionConf, RetryingCallerInterceptor interceptor,
     ServerStatisticTracker stats, MetricsConnection metrics) {
     return instantiate(configuration, connectionConf, interceptor, stats, metrics,
-      OperationInterceptorFactory.NO_OP);
+      createOperationInterceptorFactory(configuration));
   }
 
   public static RpcRetryingCallerFactory instantiate(Configuration configuration,
@@ -125,4 +131,20 @@ public static RpcRetryingCallerFactory instantiate(Configuration configuration,
     }
     return factory;
   }
+
+  private static OperationInterceptorFactory createOperationInterceptorFactory(Configuration conf) {
+    String clazz = conf.get(OperationInterceptorFactory.HBASE_CLIENT_OPERATION_INTERCEPTOR_IMPL);
+    if (clazz == null || clazz.isEmpty()) {
+      return OperationInterceptorFactory.NO_OP;
+    }
+    try {
+      Class<? extends OperationInterceptorFactory> factoryClass =
+        conf.getClassByName(clazz).asSubclass(OperationInterceptorFactory.class);
+      return ReflectionUtils.newInstance(factoryClass, conf);
+    } catch (ClassNotFoundException e) {
+      LOG.warn("Failed to load OperationInterceptorFactory class: {}, using NO_OP instead", clazz,
+        e);
+      return OperationInterceptorFactory.NO_OP;
+    }
+  }
 }