Repository: hive
Updated Branches:
  refs/heads/master 5a5f8e49f -> 89703e7d0


HIVE-12266 When client exists abnormally, it doesn't release ACID locks (Wei 
Zheng, via Eugene Koifman)


Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/595fa998
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/595fa998
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/595fa998

Branch: refs/heads/master
Commit: 595fa9988fcb3e67b60845b44e1df4cc49ce38b2
Parents: 5a5f8e4
Author: Eugene Koifman <ekoif...@hortonworks.com>
Authored: Tue Nov 3 09:03:54 2015 -0800
Committer: Eugene Koifman <ekoif...@hortonworks.com>
Committed: Tue Nov 3 09:03:54 2015 -0800

----------------------------------------------------------------------
 .../java/org/apache/hadoop/hive/ql/Driver.java  | 43 +++++++++++++++-----
 1 file changed, 32 insertions(+), 11 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hive/blob/595fa998/ql/src/java/org/apache/hadoop/hive/ql/Driver.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/Driver.java 
b/ql/src/java/org/apache/hadoop/hive/ql/Driver.java
index 18052f3..93c7a54 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/Driver.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/Driver.java
@@ -121,12 +121,14 @@ import org.apache.hadoop.hive.shims.Utils;
 import org.apache.hadoop.mapred.ClusterStatus;
 import org.apache.hadoop.mapred.JobClient;
 import org.apache.hadoop.mapred.JobConf;
+import org.apache.hive.common.util.ShutdownHookManager;
 
 public class Driver implements CommandProcessor {
 
   static final private String CLASS_NAME = Driver.class.getName();
   private static final Logger LOG = LoggerFactory.getLogger(CLASS_NAME);
   static final private LogHelper console = new LogHelper(LOG);
+  static final int SHUTDOWN_HOOK_PRIORITY = 0;
 
   private int maxRows = 100;
   ByteStream.Output bos = new ByteStream.Output();
@@ -390,7 +392,20 @@ public class Driver implements CommandProcessor {
 
     try {
       // Initialize the transaction manager.  This must be done before analyze 
is called.
-      SessionState.get().initTxnMgr(conf);
+      final HiveTxnManager txnManager = SessionState.get().initTxnMgr(conf);
+      // In case when user Ctrl-C twice to kill Hive CLI JVM, we want to 
release locks
+      ShutdownHookManager.addShutdownHook(
+          new Runnable() {
+            @Override
+            public void run() {
+              try {
+                releaseLocksAndCommitOrRollback(false, txnManager);
+              } catch (LockException e) {
+                LOG.warn("Exception when releasing locks in ShutdownHook for 
Driver: " +
+                    e.getMessage());
+              }
+            }
+          }, SHUTDOWN_HOOK_PRIORITY);
 
       command = new VariableSubstitution(new HiveVariableSource() {
         @Override
@@ -537,7 +552,7 @@ public class Driver implements CommandProcessor {
    *
    * @param sem semantic analyzer for analyzed query
    * @param plan query plan
-   * @param astStringTree AST tree dump
+   * @param astTree AST tree dump
    * @throws java.io.IOException
    */
   private String getExplainOutput(BaseSemanticAnalyzer sem, QueryPlan plan,
@@ -1049,15 +1064,21 @@ public class Driver implements CommandProcessor {
   /**
    * @param commit if there is an open transaction and if true, commit,
    *               if false rollback.  If there is no open transaction this 
parameter is ignored.
+   * @param txnManager an optional existing transaction manager retrieved 
earlier from the session
    *
    **/
-  private void releaseLocksAndCommitOrRollback(boolean commit)
+  private void releaseLocksAndCommitOrRollback(boolean commit, HiveTxnManager 
txnManager)
       throws LockException {
     PerfLogger perfLogger = SessionState.getPerfLogger();
     perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.RELEASE_LOCKS);
 
-    SessionState ss = SessionState.get();
-    HiveTxnManager txnMgr = ss.getTxnMgr();
+    HiveTxnManager txnMgr;
+    if (txnManager == null) {
+      SessionState ss = SessionState.get();
+      txnMgr = ss.getTxnMgr();
+    } else {
+      txnMgr = txnManager;
+    }
     // If we've opened a transaction we need to commit or rollback rather than 
explicitly
     // releasing the locks.
     if (txnMgr.isTxnOpen()) {
@@ -1206,7 +1227,7 @@ public class Driver implements CommandProcessor {
     }
     if (ret != 0) {
       try {
-        releaseLocksAndCommitOrRollback(false);
+        releaseLocksAndCommitOrRollback(false, null);
       } catch (LockException e) {
         LOG.warn("Exception in releasing locks. "
             + org.apache.hadoop.util.StringUtils.stringifyException(e));
@@ -1287,7 +1308,7 @@ public class Driver implements CommandProcessor {
         if(plan.getAutoCommitValue() && !txnManager.getAutoCommit()) {
           /*here, if there is an open txn, we want to commit it; this behavior 
matches
           * 
https://docs.oracle.com/javase/6/docs/api/java/sql/Connection.html#setAutoCommit(boolean)*/
-          releaseLocksAndCommitOrRollback(true);
+          releaseLocksAndCommitOrRollback(true, null);
           txnManager.setAutoCommit(true);
         }
         else if(!plan.getAutoCommitValue() && txnManager.getAutoCommit()) {
@@ -1315,10 +1336,10 @@ public class Driver implements CommandProcessor {
     //if needRequireLock is false, the release here will do nothing because 
there is no lock
     try {
       if(txnManager.getAutoCommit() || plan.getOperation() == 
HiveOperation.COMMIT) {
-        releaseLocksAndCommitOrRollback(true);
+        releaseLocksAndCommitOrRollback(true, null);
       }
       else if(plan.getOperation() == HiveOperation.ROLLBACK) {
-        releaseLocksAndCommitOrRollback(false);
+        releaseLocksAndCommitOrRollback(false, null);
       }
       else {
         //txn (if there is one started) is not finished
@@ -1349,7 +1370,7 @@ public class Driver implements CommandProcessor {
   private CommandProcessorResponse rollback(CommandProcessorResponse cpr) {
     //console.printError(cpr.toString());
     try {
-      releaseLocksAndCommitOrRollback(false);
+      releaseLocksAndCommitOrRollback(false, null);
     }
     catch (LockException e) {
       LOG.error("rollback() FAILED: " + cpr);//make sure not to loose 
@@ -1897,7 +1918,7 @@ public class Driver implements CommandProcessor {
     destroyed = true;
     if (!hiveLocks.isEmpty()) {
       try {
-        releaseLocksAndCommitOrRollback(false);
+        releaseLocksAndCommitOrRollback(false, null);
       } catch (LockException e) {
         LOG.warn("Exception when releasing locking in destroy: " +
             e.getMessage());

Reply via email to