HIVE-12679: Allow users to be able to specify an implementation of IMetaStoreClient via HiveConf

common/src/java/org/apache/hadoop/hive/conf/HiveConf.java

@@ -829,6 +829,9 @@ public static enum ConfVars {
     HADOOPNUMREDUCERS("mapreduce.job.reduces", -1, "", true),
 
     // Metastore stuff. Be sure to update HiveConf.metaVars when you add something here!
+    METASTORE_CLIENT_FACTORY_CLASS("hive.metastore.client.factory.class",
+        "org.apache.hadoop.hive.ql.metadata.SessionHiveMetaStoreClientFactory",
+        "The name of the factory class that produces objects implementing the IMetaStoreClient interface."),
     METASTOREDBTYPE("hive.metastore.db.type", "DERBY", new StringSet("DERBY", "ORACLE", "MYSQL", "MSSQL", "POSTGRES"),
         "Type of database used by the metastore. Information schema & JDBCStorageHandler depend on it."),
     /**

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

@@ -215,6 +215,7 @@
 import org.apache.hadoop.mapred.InputFormat;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.util.StringUtils;
+import org.apache.hadoop.util.ReflectionUtils;
 import org.apache.hive.common.util.HiveVersionInfo;
 import org.apache.hive.common.util.TxnIdUtils;
 import org.apache.thrift.TException;
@@ -5129,11 +5130,25 @@ public HiveMetaHook getHook(
       }
     };
 
-    if (conf.getBoolVar(ConfVars.METASTORE_FASTPATH)) {
-      return new SessionHiveMetaStoreClient(conf, hookLoader, allowEmbedded);
-    } else {
-      return RetryingMetaStoreClient.getProxy(conf, hookLoader, metaCallTimeMap,
-          SessionHiveMetaStoreClient.class.getName(), allowEmbedded);
+    return createMetaStoreClientFactory(conf)
+        .createMetaStoreClient(conf, hookLoader, allowEmbedded, metaCallTimeMap);
+  }
+
+  private static HiveMetaStoreClientFactory createMetaStoreClientFactory(HiveConf conf) throws
+          MetaException {
+    String metaStoreClientFactoryClassName = conf.getVar(HiveConf.ConfVars.METASTORE_CLIENT_FACTORY_CLASS);
+
+    try {
+      Class<? extends HiveMetaStoreClientFactory> factoryClass =
+              conf.getClassByName(metaStoreClientFactoryClassName)
+                      .asSubclass(HiveMetaStoreClientFactory.class);
+      return ReflectionUtils.newInstance(factoryClass, conf);
+    } catch (Exception e) {
+      String errorMessage = String.format(
+              "Unable to instantiate a metastore client factory %s due to: %s",
+              metaStoreClientFactoryClassName, e);
+      LOG.error(errorMessage, e);
+      throw new MetaException(errorMessage);
     }
   }
 

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

@@ -0,0 +1,55 @@
+/*
+ * 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.hive.ql.metadata;
+
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.HiveMetaHookLoader;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+
+/**
+ * Abstract factory that defines an interface for other factories that produce concrete
+ * MetaStoreClient objects.
+ *
+ */
+public interface HiveMetaStoreClientFactory {
+
+  /**
+   * A method for producing IMetaStoreClient objects.
+   *
+   * The implementation returned by this method must throw a MetaException if allowEmbedded = true
+   * and it does not support embedded mode.
+   *
+   * @param conf
+   *          Hive Configuration.
+   * @param hookLoader
+   *          Hook for handling events related to tables.
+   * @param allowEmbedded
+   *          Flag indicating the implementation must run in-process, e.g. for unit testing or
+   *          "fast path".
+   * @param metaCallTimeMap
+   *          A container for storing entry and exit timestamps of IMetaStoreClient method
+   *          invocations.
+   * @return IMetaStoreClient An implementation of IMetaStoreClient.
+   * @throws MetaException
+   */
+  IMetaStoreClient createMetaStoreClient(HiveConf conf, HiveMetaHookLoader hookLoader,
+      boolean allowEmbedded, ConcurrentHashMap<String, Long> metaCallTimeMap) throws MetaException;
+}

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

@@ -0,0 +1,52 @@
+/*
+ * 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.hive.ql.metadata;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+import org.apache.hadoop.hive.metastore.HiveMetaHookLoader;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.RetryingMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+
+/**
+ * Default MetaStoreClientFactory for Hive which produces SessionHiveMetaStoreClient objects.
+ *
+ */
+public final class SessionHiveMetaStoreClientFactory implements HiveMetaStoreClientFactory {
+
+  @Override
+  public IMetaStoreClient createMetaStoreClient(HiveConf conf, HiveMetaHookLoader hookLoader,
+      boolean allowEmbedded,
+      ConcurrentHashMap<String, Long> metaCallTimeMap) throws MetaException {
+
+    checkNotNull(conf, "conf cannot be null!");
+    checkNotNull(hookLoader, "hookLoader cannot be null!");
+    checkNotNull(metaCallTimeMap, "metaCallTimeMap cannot be null!");
+
+    if (conf.getBoolVar(ConfVars.METASTORE_FASTPATH)) {
+      return new SessionHiveMetaStoreClient(conf, hookLoader, allowEmbedded);
+    } else {
+      return RetryingMetaStoreClient.getProxy(conf, hookLoader, metaCallTimeMap,
+          SessionHiveMetaStoreClient.class.getName(), allowEmbedded);
+    }
+  }
+}

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

@@ -19,6 +19,8 @@
 package org.apache.hadoop.hive.ql.metadata;
 
 import static org.apache.hadoop.hive.metastore.Warehouse.DEFAULT_DATABASE_NAME;
+import static org.junit.Assert.assertThat;
+import static org.hamcrest.CoreMatchers.instanceOf;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -34,6 +36,7 @@
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
 import org.apache.hadoop.hive.metastore.PartitionDropOptions;
 import org.apache.hadoop.hive.metastore.Warehouse;
 import org.apache.hadoop.hive.metastore.api.Database;
@@ -798,6 +801,42 @@ public void testHiveRefreshOnConfChange() throws Throwable{
     assertTrue(prevHiveObj != newHiveObj);
   }
 
+  @Test
+  public void testLoadingHiveMetaStoreClientFactory() throws Throwable {
+    String factoryClassName = SessionHiveMetaStoreClientFactory.class.getName();
+    HiveConf conf = new HiveConf();
+    conf.setVar(ConfVars.METASTORE_CLIENT_FACTORY_CLASS, factoryClassName);
+    // Make sure we instantiate the embedded version
+    // so the implementation chosen is SessionHiveMetaStoreClient, not a retryable version of it.
+    conf.setBoolVar(ConfVars.METASTORE_FASTPATH, true);
+    // The current object was constructed in setUp() before we got here
+    // so clean that up so we can inject our own dummy implementation of IMetaStoreClient
+    Hive.closeCurrent();
+    Hive hive = Hive.get(conf);
+    IMetaStoreClient tmp = hive.getMSC();
+    assertNotNull("getMSC() failed.", tmp);
+    assertThat("Invalid default client implementation created.", tmp,
+        instanceOf(SessionHiveMetaStoreClient.class));
+  }
+
+  @Test
+  public void testLoadingInvalidHiveMetaStoreClientFactory() throws Throwable {
+    // Intentionally invalid class
+    String factoryClassName = String.class.getName();
+    HiveConf conf = new HiveConf();
+    conf.setVar(HiveConf.ConfVars.METASTORE_CLIENT_FACTORY_CLASS, factoryClassName);
+    // The current object was constructed in setUp() before we got here
+    // so clean that up so we can inject our own dummy implementation of IMetaStoreClient
+    Hive.closeCurrent();
+    Hive hive = Hive.get(conf);
+    try {
+      hive.getMSC();
+      fail("getMSC() was expected to throw MetaException.");
+    } catch (Exception e) {
+      assertTrue("getMSC() failed, which IS expected.", true);
+    }
+  }
+
   // shamelessly copied from Path in hadoop-2
   private static final String SEPARATOR = "/";
   private static final char SEPARATOR_CHAR = '/';