Pluggable HiveSession implementation

InMobi · Dec 24, 2013 · ed79ba3 · ed79ba3
1 parent c1ae856
commit ed79ba3
Show file tree

Hide file tree

Showing 3 changed files with 41 additions and 7 deletions.
diff --git a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
@@ -762,6 +762,9 @@ public static enum ConfVars {
     // By default, logs will be purged after a week
     HIVE_SERVER2_LOG_PURGE_DELAY("hive.server2.log.purge.delay", 7 * 1440 * 60 * 1000L),
 
+    // Hive session impl classes
+    HIVE_SESSION_IMPL_CLASSNAME("hive.session.impl.classname", null),
+    HIVE_SESSION_IMPL_WITH_UGI_CLASSNAME("hive.session.impl.withugi.classname", null),
     // HiveServer2 auth configuration
     HIVE_SERVER2_AUTHENTICATION("hive.server2.authentication", "NONE"),
     HIVE_SERVER2_KERBEROS_KEYTAB("hive.server2.authentication.kerberos.keytab", ""),

diff --git a/service/src/java/org/apache/hive/service/cli/CLIService.java b/service/src/java/org/apache/hive/service/cli/CLIService.java
@@ -32,15 +32,12 @@
 import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
 import org.apache.hadoop.hive.metastore.IMetaStoreClient;
-import org.apache.hadoop.hive.ql.TaskStatus;
 import org.apache.hadoop.hive.ql.metadata.Hive;
 import org.apache.hadoop.hive.ql.metadata.HiveException;
 import org.apache.hadoop.hive.shims.ShimLoader;
 import org.apache.hive.service.CompositeService;
 import org.apache.hive.service.ServiceException;
 import org.apache.hive.service.auth.HiveAuthFactory;
-import org.apache.hive.service.cli.operation.Operation;
-import org.apache.hive.service.cli.operation.SQLOperation;
 import org.apache.hive.service.cli.session.HiveSession;
 import org.apache.hive.service.cli.session.SessionManager;
 
@@ -110,6 +107,9 @@ public synchronized void stop() {
     super.stop();
   }
 
+  public SessionManager getSessionManager() {
+    return sessionManager;
+  }
 
   /* (non-Javadoc)
    * @see org.apache.hive.service.cli.ICLIService#openSession(java.lang.String, java.lang.String, java.util.Map)

diff --git a/service/src/java/org/apache/hive/service/cli/session/SessionManager.java b/service/src/java/org/apache/hive/service/cli/session/SessionManager.java
@@ -20,7 +20,11 @@
 
 import java.io.File;
 import java.io.IOException;
-import java.util.*;
+import java.lang.reflect.Constructor;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
@@ -55,6 +59,8 @@ public class SessionManager extends CompositeService {
   private ScheduledExecutorService logPurgerService;
   private File queryLogDir;
   private boolean isLogRedirectionEnabled;
+  private String sessionImplWithUGIclassName;
+  private String sessionImplclassName;
 
   public SessionManager() {
     super("SessionManager");
@@ -76,6 +82,8 @@ public synchronized void init(HiveConf hiveConf) {
         keepAliveTime, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(backgroundPoolQueueSize));
     backgroundOperationPool.allowCoreThreadTimeOut(true);
 
+    this.sessionImplclassName = hiveConf.getVar(ConfVars.HIVE_SESSION_IMPL_CLASSNAME);
+    this.sessionImplWithUGIclassName = hiveConf.getVar(ConfVars.HIVE_SESSION_IMPL_WITH_UGI_CLASSNAME);
     if (hiveConf.getBoolVar(ConfVars.HIVE_SERVER2_LOG_REDIRECTION_ENABLED)) {
       queryLogDir = new File(hiveConf.getVar(ConfVars.HIVE_SERVER2_LOG_DIRECTORY));
       isLogRedirectionEnabled = true;
@@ -140,12 +148,35 @@ public SessionHandle openSession(String username, String password, Map<String, S
     }
     HiveSession session;
     if (withImpersonation) {
-      HiveSessionImplwithUGI hiveSessionUgi = new HiveSessionImplwithUGI(username, password,
-          sessionConf, delegationToken);
+      HiveSessionImplwithUGI hiveSessionUgi;
+      if (sessionImplWithUGIclassName == null) {
+        hiveSessionUgi = new HiveSessionImplwithUGI(username, password,
+            sessionConf, delegationToken);
+      } else {
+        try {
+          Class<?> clazz = Class.forName(sessionImplWithUGIclassName);
+          Constructor<?> constructor = clazz.getConstructor(String.class, String.class, Map.class, String.class);
+          hiveSessionUgi = (HiveSessionImplwithUGI) constructor.newInstance(new Object[]
+              {username, password, sessionConf, delegationToken});
+        } catch (Exception e) {
+          throw new HiveSQLException("Cannot initilize session class:" + sessionImplWithUGIclassName);
+        }
+      }
       session = HiveSessionProxy.getProxy(hiveSessionUgi, hiveSessionUgi.getSessionUgi());
       hiveSessionUgi.setProxySession(session);
     } else {
-      session = new HiveSessionImpl(username, password, sessionConf);
+      if (sessionImplclassName == null) {
+        session = new HiveSessionImpl(username, password, sessionConf);
+      } else {
+        try {
+          Class<?> clazz = Class.forName(sessionImplclassName);
+          Constructor<?> constructor = clazz.getConstructor(String.class, String.class, Map.class);
+          session = (HiveSession) constructor.newInstance(new Object[]
+              {username, password, sessionConf});
+        } catch (Exception e) {
+          throw new HiveSQLException("Cannot initilize session class:" + sessionImplclassName);
+        }
+      }
     }
     session.setSessionManager(this);
     session.setOperationManager(operationManager);