HIVE-29620: RetryingMetaStoreClient uses HiveMetaStoreClientBuilder t…

iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/client/HiveRESTCatalogClient.java

@@ -68,10 +68,6 @@ public class HiveRESTCatalogClient extends BaseMetaStoreClient {
 
   private RESTCatalog restCatalog;
 
-  public HiveRESTCatalogClient(Configuration conf, boolean allowEmbedded) {
-    this(conf);
-  }
-
   public HiveRESTCatalogClient(Configuration conf) {
     super(conf);
     reconnect();

ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java

@@ -6094,8 +6094,7 @@ public HiveMetaHook getHook(
       }
     };
 
-    HiveMetaStoreClientBuilder msClientBuilder = new HiveMetaStoreClientBuilder(conf)
-        .newClient(allowEmbedded)
+    HiveMetaStoreClientBuilder msClientBuilder = new HiveMetaStoreClientBuilder(conf, allowEmbedded)
         .enhanceWith(client ->
             HiveMetaStoreClientWithLocalCache.newClient(conf, client))
         .enhanceWith(client ->

ql/src/java/org/apache/hadoop/hive/ql/metadata/HiveMetaStoreClientWithLocalCache.java

@@ -41,7 +41,7 @@
 import org.apache.hadoop.hive.metastore.api.PartitionsByExprRequest;
 import org.apache.hadoop.hive.metastore.api.PartitionsStatsRequest;
 import org.apache.hadoop.hive.metastore.api.Table;
-import org.apache.hadoop.hive.metastore.client.MetaStoreClientWrapper;
+import org.apache.hadoop.hive.metastore.client.FilterMetaStoreClient;
 import org.apache.hadoop.hive.metastore.client.utils.HiveMetaStoreClientUtils;
 import org.apache.hadoop.hive.ql.io.AcidUtils;
 import org.apache.hadoop.hive.ql.metadata.client.MetaStoreClientCacheUtils;
@@ -76,7 +76,7 @@
  * It helps to reduce the time spent in compilation by using HS2 memory more effectively, and it allows to
  * improve HMS throughput for multi-tenant workloads by reducing the number of calls it needs to serve.
  */
-public class HiveMetaStoreClientWithLocalCache extends MetaStoreClientWrapper {
+public class HiveMetaStoreClientWithLocalCache extends FilterMetaStoreClient {
 
   private static final Logger LOG = LoggerFactory.getLogger(HiveMetaStoreClientWithLocalCache.class);
 

ql/src/java/org/apache/hadoop/hive/ql/metadata/SessionHiveMetaStoreClient.java

@@ -95,7 +95,7 @@
 import org.apache.hadoop.hive.metastore.api.UniqueConstraintsRequest;
 import org.apache.hadoop.hive.metastore.api.UniqueConstraintsResponse;
 import org.apache.hadoop.hive.metastore.cache.CachedStore;
-import org.apache.hadoop.hive.metastore.client.MetaStoreClientWrapper;
+import org.apache.hadoop.hive.metastore.client.FilterMetaStoreClient;
 import org.apache.hadoop.hive.metastore.client.ThriftHiveMetaStoreClient;
 import org.apache.hadoop.hive.metastore.client.builder.PartitionBuilder;
 import org.apache.hadoop.hive.metastore.client.utils.HiveMetaStoreClientUtils;
@@ -155,7 +155,7 @@
  * so the readers of the objects in these maps should have the most recent view of the object.
  * But again, could be fragile.
  */
-public class SessionHiveMetaStoreClient extends MetaStoreClientWrapper {
+public class SessionHiveMetaStoreClient extends FilterMetaStoreClient {
   private static final Logger LOG = LoggerFactory.getLogger(SessionHiveMetaStoreClient.class);
 
   private volatile Warehouse wh = null;

ql/src/test/org/apache/hadoop/hive/ql/metadata/TestHiveMetaStoreClient.java

@@ -36,7 +36,7 @@
 import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
 import org.apache.hadoop.hive.metastore.api.Table;
 import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
-import org.apache.hadoop.hive.metastore.client.MetaStoreClientWrapper;
+import org.apache.hadoop.hive.metastore.client.FilterMetaStoreClient;
 import org.apache.hadoop.hive.metastore.client.HookEnabledMetaStoreClient;
 import org.apache.hadoop.hive.metastore.client.SynchronizedMetaStoreClient;
 import org.apache.hadoop.hive.metastore.client.ThriftHiveMetaStoreClient;
@@ -66,7 +66,7 @@
  *  As we support more APIs, we should add them here with appropriate test cases.
  *
  */
-public class TestHiveMetaStoreClient extends MetaStoreClientWrapper {
+public class TestHiveMetaStoreClient extends FilterMetaStoreClient {
 
   private static IMetaStoreClient createUnderlyingClient(Configuration conf) throws MetaException {
     IMetaStoreClient thriftClient = new ThriftHiveMetaStoreClient(conf, true);

standalone-metastore/metastore-client/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClient.java

@@ -28,7 +28,7 @@
 import org.apache.hadoop.hive.metastore.api.Partition;
 import org.apache.hadoop.hive.metastore.api.Table;
 import org.apache.hadoop.hive.metastore.api.Type;
-import org.apache.hadoop.hive.metastore.client.MetaStoreClientWrapper;
+import org.apache.hadoop.hive.metastore.client.FilterMetaStoreClient;
 import org.apache.hadoop.hive.metastore.client.ThriftHiveMetaStoreClient;
 import org.apache.hadoop.hive.metastore.client.builder.HiveMetaStoreClientBuilder;
 import org.apache.thrift.TException;
@@ -45,7 +45,7 @@
  */
 @InterfaceAudience.Public
 @InterfaceStability.Evolving
-public class HiveMetaStoreClient extends MetaStoreClientWrapper implements IMetaStoreClient, AutoCloseable {
+public class HiveMetaStoreClient extends FilterMetaStoreClient implements IMetaStoreClient, AutoCloseable {
   public static final String MANUALLY_INITIATED_COMPACTION = "manual";
   public static final String RENAME_PARTITION_MAKE_COPY = "renamePartitionMakeCopy";
 
@@ -61,7 +61,7 @@ public HiveMetaStoreClient(Configuration conf, HiveMetaHookLoader hookLoader) th
 
   public HiveMetaStoreClient(Configuration conf, HiveMetaHookLoader hookLoader, Boolean allowEmbedded)
     throws MetaException {
-    this(conf, hookLoader, new HiveMetaStoreClientBuilder(conf).newClient(allowEmbedded).build());
+    this(conf, hookLoader, new HiveMetaStoreClientBuilder(conf, allowEmbedded).build());
   }
 
   private HiveMetaStoreClient(Configuration conf, HiveMetaHookLoader hookLoader,
@@ -75,8 +75,7 @@ private HiveMetaStoreClient(Configuration conf, HiveMetaHookLoader hookLoader,
 
   private static IMetaStoreClient createUnderlyingClient(Configuration conf, HiveMetaHookLoader hookLoader,
       IMetaStoreClient baseMetaStoreClient) {
-    return new HiveMetaStoreClientBuilder(conf)
-       .client(baseMetaStoreClient)
+    return new HiveMetaStoreClientBuilder(conf, baseMetaStoreClient)
        .withHooks(hookLoader)
        .threadSafe()
        .build();

standalone-metastore/metastore-client/src/main/java/org/apache/hadoop/hive/metastore/IMetaStoreClient.java

@@ -65,11 +65,12 @@ default void setHiveAddedJars(String addedJars) {
 
   /**
    * Returns true if the current client is using an in process metastore (local metastore).
+   * Default false, as in real production the client should always connect to a remote meta service
    *
    * @return
    */
-  default boolean isLocalMetaStore(){
-    throw new UnsupportedOperationException("MetaStore client does not support checking if metastore is local");
+  default boolean isLocalMetaStore() {
+    return false;
   }
 
   /**

standalone-metastore/metastore-client/src/main/java/org/apache/hadoop/hive/metastore/RetryingMetaStoreClient.java

@@ -33,9 +33,11 @@
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 
+import org.apache.commons.lang3.ClassUtils;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hive.common.classification.RetrySemantics;
+import org.apache.hadoop.hive.metastore.client.builder.HiveMetaStoreClientBuilder;
 import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
 import org.apache.hadoop.hive.metastore.conf.MetastoreConf.ConfVars;
 import org.apache.hadoop.hive.metastore.utils.JavaUtils;
@@ -69,15 +71,7 @@ public class RetryingMetaStoreClient implements InvocationHandler {
   private final Map<String, Long> metaCallTimeMap;
   private final long connectionLifeTimeInMillis;
   private long lastConnectionTime;
-  private boolean localMetaStore;
-
-
-  protected RetryingMetaStoreClient(Configuration conf, Class<?>[] constructorArgTypes,
-      Object[] constructorArgs, Map<String, Long> metaCallTimeMap,
-      Class<? extends IMetaStoreClient> msClientClass) throws MetaException {
-    this(conf, metaCallTimeMap, () ->
-        JavaUtils.newInstance(msClientClass, constructorArgTypes, constructorArgs));
-  }
+  private final boolean localMetaStore;
 
   protected RetryingMetaStoreClient(Configuration conf, Map<String, Long> metaCallTimeMap,
         Supplier<? extends IMetaStoreClient> msClient) throws MetaException {
@@ -95,12 +89,11 @@ protected RetryingMetaStoreClient(Configuration conf, Map<String, Long> metaCall
     this.connectionLifeTimeInMillis = MetastoreConf.getTimeVar(conf,
         ConfVars.CLIENT_SOCKET_LIFETIME, TimeUnit.MILLISECONDS);
     this.lastConnectionTime = System.currentTimeMillis();
-    String msUri = MetastoreConf.getVar(conf, ConfVars.THRIFT_URIS);
-    localMetaStore = (msUri == null) || msUri.trim().isEmpty();
 
     SecurityUtils.reloginExpiringKeytabUser();
 
     this.base = msClient.get();
+    this.localMetaStore = base.isLocalMetaStore();
 
     LOG.info("RetryingMetaStoreClient proxy=" + base.getClass() + " ugi=" + this.ugi
         + " retries=" + this.retryLimit + " delay=" + this.retryDelaySeconds
@@ -109,9 +102,7 @@ protected RetryingMetaStoreClient(Configuration conf, Map<String, Long> metaCall
 
   public static IMetaStoreClient getProxy(
       Configuration hiveConf, boolean allowEmbedded) throws MetaException {
-    return getProxy(hiveConf, new Class[]{Configuration.class, HiveMetaHookLoader.class, Boolean.class},
-        new Object[]{hiveConf, null, allowEmbedded}, null, HiveMetaStoreClient.class.getName()
-    );
+    return new HiveMetaStoreClientBuilder(hiveConf, allowEmbedded).withRetry(null).build();
   }
 
   @VisibleForTesting
@@ -123,13 +114,14 @@ public static IMetaStoreClient getProxy(Configuration hiveConf, HiveMetaHookLoad
   public static IMetaStoreClient getProxy(Configuration hiveConf, HiveMetaHookLoader hookLoader,
       Map<String, Long> metaCallTimeMap, String mscClassName, boolean allowEmbedded)
           throws MetaException {
-
-    return getProxy(hiveConf,
-        new Class[] {Configuration.class, HiveMetaHookLoader.class, Boolean.class},
-        new Object[] {hiveConf, hookLoader, allowEmbedded},
-        metaCallTimeMap,
-        mscClassName
-    );
+    String origClientImpl = MetastoreConf.getVar(hiveConf, ConfVars.METASTORE_CLIENT_IMPL);
+    try {
+      MetastoreConf.setVar(hiveConf, ConfVars.METASTORE_CLIENT_IMPL, mscClassName);
+      return new HiveMetaStoreClientBuilder(hiveConf, allowEmbedded)
+          .withHooks(hookLoader).withRetry(metaCallTimeMap).build();
+    } finally {
+      MetastoreConf.setVar(hiveConf, ConfVars.METASTORE_CLIENT_IMPL, origClientImpl);
+    }
   }
 
   /**
@@ -148,26 +140,18 @@ public static IMetaStoreClient getProxy(Configuration hiveConf, Class<?>[] const
   public static IMetaStoreClient getProxy(Configuration hiveConf, Class<?>[] constructorArgTypes,
       Object[] constructorArgs, Map<String, Long> metaCallTimeMap,
       String mscClassName) throws MetaException {
-
     @SuppressWarnings("unchecked")
     Class<? extends IMetaStoreClient> baseClass =
         JavaUtils.getClass(mscClassName, IMetaStoreClient.class);
-
-    RetryingMetaStoreClient handler =
-        new RetryingMetaStoreClient(hiveConf, constructorArgTypes, constructorArgs,
-            metaCallTimeMap, baseClass);
-    return getProxy(baseClass.getInterfaces(), handler);
+    IMetaStoreClient baseClient = JavaUtils.newInstance(baseClass, constructorArgTypes, constructorArgs);
+    return new HiveMetaStoreClientBuilder(hiveConf, baseClient).withRetry(metaCallTimeMap).build();
   }
 
   public static IMetaStoreClient getProxy(Configuration hiveConf, Map<String, Long> metaCallTimeMap,
       IMetaStoreClient msClient) throws MetaException {
     RetryingMetaStoreClient handler =
         new RetryingMetaStoreClient(hiveConf, metaCallTimeMap, () -> msClient);
-    return getProxy(msClient.getClass().getInterfaces(), handler);
-  }
-
-  private static IMetaStoreClient getProxy(Class<?>[] interfaces,
-      RetryingMetaStoreClient handler) {
+    Class<?>[] interfaces = ClassUtils.getAllInterfaces(msClient.getClass()).toArray(new Class[0]);
     return (IMetaStoreClient) Proxy.newProxyInstance(
             RetryingMetaStoreClient.class.getClassLoader(), interfaces, handler);
   }

standalone-metastore/metastore-client/src/main/java/org/apache/hadoop/hive/metastore/client/MetaStoreClientWrapper.java → standalone-metastore/metastore-client/src/main/java/org/apache/hadoop/hive/metastore/client/FilterMetaStoreClient.java

@@ -36,11 +36,11 @@
 import java.util.List;
 import java.util.Map;
 
-public abstract class MetaStoreClientWrapper extends BaseMetaStoreClient {
+public abstract class FilterMetaStoreClient extends BaseMetaStoreClient {
 
   protected final IMetaStoreClient delegate;
 
-  public MetaStoreClientWrapper(IMetaStoreClient delegate, Configuration conf) {
+  public FilterMetaStoreClient(IMetaStoreClient delegate, Configuration conf) {
     super(conf);
     this.delegate = delegate;
   }

standalone-metastore/metastore-client/src/main/java/org/apache/hadoop/hive/metastore/client/HookEnabledMetaStoreClient.java

@@ -18,7 +18,6 @@
 
 package org.apache.hadoop.hive.metastore.client;
 
-import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hive.common.TableName;
 import org.apache.hadoop.hive.metastore.DefaultHiveMetaHook;
@@ -55,7 +54,7 @@
 import static org.apache.hadoop.hive.common.AcidConstants.SOFT_DELETE_TABLE;
 import static org.apache.hadoop.hive.metastore.utils.MetaStoreUtils.getDefaultCatalog;
 
-public class HookEnabledMetaStoreClient extends MetaStoreClientWrapper {
+public class HookEnabledMetaStoreClient extends FilterMetaStoreClient {
   private final HiveMetaHookLoader hookLoader;
 
   private static final Logger LOG = LoggerFactory.getLogger(HookEnabledMetaStoreClient.class);

standalone-metastore/metastore-client/src/main/java/org/apache/hadoop/hive/metastore/client/SynchronizedMetaStoreClient.java

@@ -32,7 +32,7 @@
  * The reflection logic originally comes from {@link HiveMetaStoreClient}.
  * This should be used by multi-thread applications unless all the underlying layers are thread-safe.
  */
-public class SynchronizedMetaStoreClient extends MetaStoreClientWrapper implements IMetaStoreClient {
+public class SynchronizedMetaStoreClient extends FilterMetaStoreClient implements IMetaStoreClient {
   public static SynchronizedMetaStoreClient newClient(Configuration conf, IMetaStoreClient delegate) {
     return new SynchronizedMetaStoreClient(conf, delegate);
   }

standalone-metastore/metastore-client/src/main/java/org/apache/hadoop/hive/metastore/client/builder/HiveMetaStoreClientBuilder.java

@@ -19,41 +19,49 @@
 package org.apache.hadoop.hive.metastore.client.builder;
 
 import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.commons.lang3.reflect.ConstructorUtils;
+import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hive.metastore.HiveMetaHookLoader;
+import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
 import org.apache.hadoop.hive.metastore.IMetaStoreClient;
 import org.apache.hadoop.hive.metastore.RetryingMetaStoreClient;
 import org.apache.hadoop.hive.metastore.api.MetaException;
 import org.apache.hadoop.hive.metastore.client.HookEnabledMetaStoreClient;
 import org.apache.hadoop.hive.metastore.client.SynchronizedMetaStoreClient;
 import org.apache.hadoop.hive.metastore.client.ThriftHiveMetaStoreClient;
 import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
-import org.apache.hadoop.hive.metastore.utils.JavaUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.lang.reflect.Constructor;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Function;
 
 public class HiveMetaStoreClientBuilder {
   private static final Logger LOG = LoggerFactory.getLogger(HiveMetaStoreClientBuilder.class);
+  private static final Map<Class<? extends IMetaStoreClient>, MetaStoreClientFactory>
+      CLIENT_FACTORIES = new ConcurrentHashMap<>();
 
   private final Configuration conf;
   private IMetaStoreClient client;
 
-  public HiveMetaStoreClientBuilder(Configuration conf) {
-    this.conf = Objects.requireNonNull(conf);
-  }
-
-  public HiveMetaStoreClientBuilder newClient(boolean allowEmbedded) throws MetaException {
+  public HiveMetaStoreClientBuilder(Configuration configuration, boolean allowEmbedded) throws MetaException {
+    this.conf = new Configuration(Objects.requireNonNull(configuration));
+    boolean isHiveClient = HiveMetaStoreClient.class.getName().equals(
+        MetastoreConf.getVar(conf, MetastoreConf.ConfVars.METASTORE_CLIENT_IMPL));
+    if (isHiveClient) {
+      // Prevent stack overflow as HiveMetaStoreClient calls HiveMetaStoreClientBuilder to build the underlying client
+      MetastoreConf.setVar(conf, MetastoreConf.ConfVars.METASTORE_CLIENT_IMPL, ThriftHiveMetaStoreClient.class.getName());
+    }
     this.client = createClient(conf, allowEmbedded);
-    return this;
   }
 
-  public HiveMetaStoreClientBuilder client(IMetaStoreClient client) {
-    this.client = client;
-    return this;
+  public HiveMetaStoreClientBuilder(Configuration conf, IMetaStoreClient client) {
+    this.conf =  Objects.requireNonNull(conf);
+    this.client = Objects.requireNonNull(client);
   }
 
   public HiveMetaStoreClientBuilder enhanceWith(Function<IMetaStoreClient, IMetaStoreClient> wrapperFunction) {
@@ -81,16 +89,16 @@ public IMetaStoreClient build() {
   }
 
   private static IMetaStoreClient createClient(Configuration conf, boolean allowEmbedded) throws MetaException {
-    Class<? extends IMetaStoreClient> mscClass = MetastoreConf.getClass(
-        conf, MetastoreConf.ConfVars.METASTORE_CLIENT_IMPL,
-        ThriftHiveMetaStoreClient.class, IMetaStoreClient.class);
-    LOG.info("Using {} as a base MetaStoreClient", mscClass.getName());
-
-    IMetaStoreClient baseMetaStoreClient = null;
     try {
-      baseMetaStoreClient = JavaUtils.newInstance(mscClass,
-          new Class[]{Configuration.class, boolean.class},
-          new Object[]{conf, allowEmbedded});
+      Class<? extends IMetaStoreClient> mscClass = MetastoreConf.getClass(
+          conf, MetastoreConf.ConfVars.METASTORE_CLIENT_IMPL,
+          ThriftHiveMetaStoreClient.class, IMetaStoreClient.class);
+      LOG.info("Using {} as a base MetaStoreClient", mscClass.getName());
+      MetaStoreClientFactory factory = CLIENT_FACTORIES.get(mscClass);
+      if (factory == null) {
+        CLIENT_FACTORIES.put(mscClass, factory = new MetaStoreClientFactory(mscClass));
+      }
+      return factory.createClient(conf, allowEmbedded);
     } catch (Throwable t) {
       // Reflection by JavaUtils will throw RuntimeException, try to get real MetaException here.
       Throwable rootCause = ExceptionUtils.getRootCause(t);
@@ -100,7 +108,34 @@ private static IMetaStoreClient createClient(Configuration conf, boolean allowEm
         throw new MetaException(rootCause.getMessage());
       }
     }
+  }
+
+   private static class MetaStoreClientFactory {
+    private Constructor<? extends IMetaStoreClient> bestMatchingCtr;
+    private Function<Pair<Configuration, Boolean>, Object[]> argsTransformer;
+
+    MetaStoreClientFactory(Class<? extends IMetaStoreClient> mscClass) {
+      Constructor<? extends IMetaStoreClient> candidate =
+          ConstructorUtils.getMatchingAccessibleConstructor(mscClass, Configuration.class, boolean.class);
+      if (candidate != null) {
+        this.bestMatchingCtr = candidate;
+        this.argsTransformer = args -> new Object[] {args.getLeft(), (boolean) args.getRight()};
+      } else if ((candidate = ConstructorUtils.getMatchingAccessibleConstructor(mscClass, Configuration.class,
+          HiveMetaHookLoader.class, Boolean.class)) != null) {
+        this.bestMatchingCtr = candidate;
+        this.argsTransformer = args ->
+            new Object[] {args.getLeft(), null, Boolean.valueOf(args.getRight())};
+      } else if ((candidate = ConstructorUtils.getMatchingAccessibleConstructor(mscClass, Configuration.class)) != null) {
+        this.bestMatchingCtr = candidate;
+        this.argsTransformer = args -> new Object[] {args.getLeft()};
+      }
+      if (bestMatchingCtr == null) {
+        throw new RuntimeException("No matching constructor found for this IMetaStoreClient " + mscClass);
+      }
+    }
 
-    return baseMetaStoreClient;
+    IMetaStoreClient createClient(Configuration conf, boolean allowEmbedded) throws Exception {
+      return bestMatchingCtr.newInstance(argsTransformer.apply(Pair.of(conf, allowEmbedded)));
+    }
   }
 }