hive-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ser...@apache.org
Subject [21/50] [abbrv] hive git commit: HIVE-11317 - ACID: Improve transaction Abort logic due to timeout (Eugene Koifman, reviewed by Alan Gates)
Date Tue, 25 Aug 2015 21:28:51 GMT
HIVE-11317 - ACID: Improve transaction Abort logic due to timeout (Eugene Koifman, reviewed
by Alan Gates)


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

Branch: refs/heads/llap
Commit: 97a6cd35a444315885008f11c20c7c28249bd42c
Parents: e8329ee
Author: Eugene Koifman <ekoifman@hortonworks.com>
Authored: Sat Aug 15 10:22:55 2015 -0700
Committer: Eugene Koifman <ekoifman@hortonworks.com>
Committed: Sat Aug 15 10:22:55 2015 -0700

----------------------------------------------------------------------
 .../org/apache/hadoop/hive/conf/HiveConf.java   |   4 +
 .../deployers/config/hive/hive-site.mysql.xml   |  22 +++
 .../hive/hcatalog/streaming/TestStreaming.java  |  54 ++++++-
 .../hadoop/hive/metastore/HiveMetaStore.java    |  18 +++
 .../hive/metastore/HouseKeeperService.java      |  39 +++++
 .../hadoop/hive/metastore/txn/TxnHandler.java   | 153 +++++++++++++------
 .../hive/metastore/txn/TestTxnHandler.java      |   7 +-
 .../java/org/apache/hadoop/hive/ql/Driver.java  |  15 +-
 .../hive/ql/txn/AcidHouseKeeperService.java     | 104 +++++++++++++
 .../hive/ql/txn/compactor/CompactorMR.java      |   6 +-
 .../hadoop/hive/ql/txn/compactor/Initiator.java |   1 +
 .../hadoop/hive/ql/txn/compactor/Worker.java    |   2 +-
 .../apache/hadoop/hive/ql/TestTxnCommands.java  |  21 +++
 .../apache/hadoop/hive/ql/TestTxnCommands2.java |   1 +
 .../hive/ql/lockmgr/TestDbTxnManager.java       |  35 +++--
 15 files changed, 421 insertions(+), 61 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hive/blob/97a6cd35/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
----------------------------------------------------------------------
diff --git a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
index 730f5be..9a6781b 100644
--- a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
+++ b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
@@ -1507,6 +1507,10 @@ public class HiveConf extends Configuration {
 
     HIVE_COMPACTOR_CLEANER_RUN_INTERVAL("hive.compactor.cleaner.run.interval", "5000ms",
         new TimeValidator(TimeUnit.MILLISECONDS), "Time between runs of the cleaner thread"),
+    HIVE_TIMEDOUT_TXN_REAPER_START("hive.timedout.txn.reaper.start", "100s",
+      new TimeValidator(TimeUnit.MILLISECONDS), "Time delay of 1st reaper run after metastore
start"),
+    HIVE_TIMEDOUT_TXN_REAPER_INTERVAL("hive.timedout.txn.reaper.interval", "180s",
+      new TimeValidator(TimeUnit.MILLISECONDS), "Time interval describing how often the reaper
runs"),
 
     // For HBase storage handler
     HIVE_HBASE_WAL_ENABLED("hive.hbase.wal.enabled", true,

http://git-wip-us.apache.org/repos/asf/hive/blob/97a6cd35/hcatalog/src/test/e2e/templeton/deployers/config/hive/hive-site.mysql.xml
----------------------------------------------------------------------
diff --git a/hcatalog/src/test/e2e/templeton/deployers/config/hive/hive-site.mysql.xml b/hcatalog/src/test/e2e/templeton/deployers/config/hive/hive-site.mysql.xml
index 70ccc31..b6f1ab7 100644
--- a/hcatalog/src/test/e2e/templeton/deployers/config/hive/hive-site.mysql.xml
+++ b/hcatalog/src/test/e2e/templeton/deployers/config/hive/hive-site.mysql.xml
@@ -62,6 +62,28 @@
         <name>hive.exec.dynamic.partition.mode</name>
         <value>nonstrict</value>
     </property>
+    <property>
+        <name>hive.compactor.initiator.on</name>
+        <value>false</value>
+    </property>
+    <property>
+        <name>hive.compactor.worker.threads</name>
+        <value>2</value>
+    </property>
+    <property>
+        <name>hive.timedout.txn.reaper.start</name>
+        <value>2s</value>
+    </property>
+<!--    <property>
+        <name>hive.txn.timeout</name>
+        <value>60s</value>
+    </property>
+    -->
+    <property>
+        <name>hive.timedout.txn.reaper.interval</name>
+        <value>30s</value>
+    </property>
+
     <!--end ACID related properties-->
 <!--
     <property>

http://git-wip-us.apache.org/repos/asf/hive/blob/97a6cd35/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/TestStreaming.java
----------------------------------------------------------------------
diff --git a/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/TestStreaming.java
b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/TestStreaming.java
index c0af533..c28d4aa 100644
--- a/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/TestStreaming.java
+++ b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/TestStreaming.java
@@ -18,7 +18,6 @@
 
 package org.apache.hive.hcatalog.streaming;
 
-import junit.framework.Assert;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.RawLocalFileSystem;
@@ -36,6 +35,7 @@ import org.apache.hadoop.hive.metastore.api.Partition;
 import org.apache.hadoop.hive.metastore.api.SerDeInfo;
 import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
 import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.metastore.api.TxnAbortedException;
 import org.apache.hadoop.hive.metastore.txn.TxnDbUtil;
 import org.apache.hadoop.hive.ql.io.AcidUtils;
 import org.apache.hadoop.hive.ql.io.HiveInputFormat;
@@ -43,6 +43,7 @@ import org.apache.hadoop.hive.ql.io.orc.OrcInputFormat;
 import org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat;
 import org.apache.hadoop.hive.ql.io.orc.OrcSerde;
 import org.apache.hadoop.hive.ql.io.orc.OrcStruct;
+import org.apache.hadoop.hive.ql.txn.AcidHouseKeeperService;
 import org.apache.hadoop.hive.serde.serdeConstants;
 import org.apache.hadoop.io.NullWritable;
 import org.apache.hadoop.mapred.InputFormat;
@@ -51,6 +52,7 @@ import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.Reporter;
 import org.apache.thrift.TException;
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
@@ -67,6 +69,7 @@ import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 
 
 public class TestStreaming {
@@ -301,6 +304,55 @@ public class TestStreaming {
     connection.close();
   }
 
+  /**
+   * check that transactions that have not heartbeated and timedout get properly aborted
+   * @throws Exception
+   */
+  @Test
+  public void testTimeOutReaper() throws Exception {
+    HiveEndPoint endPt = new HiveEndPoint(metaStoreURI, dbName2, tblName2, null);
+    DelimitedInputWriter writer = new DelimitedInputWriter(fieldNames2,",", endPt);
+    StreamingConnection connection = endPt.newConnection(false, null);
+
+    TransactionBatch txnBatch =  connection.fetchTransactionBatch(5, writer);
+    txnBatch.beginNextTransaction();
+    conf.setTimeVar(HiveConf.ConfVars.HIVE_TIMEDOUT_TXN_REAPER_START, 0, TimeUnit.SECONDS);
+    //ensure txn timesout
+    conf.setTimeVar(HiveConf.ConfVars.HIVE_TXN_TIMEOUT, 1, TimeUnit.MILLISECONDS);
+    AcidHouseKeeperService houseKeeperService = new AcidHouseKeeperService();
+    houseKeeperService.start(conf);
+    while(houseKeeperService.getIsAliveCounter() <= Integer.MIN_VALUE) {
+      Thread.sleep(100);//make sure it has run at least once
+    }
+    houseKeeperService.stop();
+    try {
+      //should fail because the TransactionBatch timed out
+      txnBatch.commit();
+    }
+    catch(TransactionError e) {
+      Assert.assertTrue("Expected aborted transaction", e.getCause() instanceof TxnAbortedException);
+    }
+    txnBatch.close();
+    txnBatch =  connection.fetchTransactionBatch(10, writer);
+    txnBatch.beginNextTransaction();
+    txnBatch.commit();
+    txnBatch.beginNextTransaction();
+    int lastCount = houseKeeperService.getIsAliveCounter();
+    houseKeeperService.start(conf);
+    while(houseKeeperService.getIsAliveCounter() <= lastCount) {
+      Thread.sleep(100);//make sure it has run at least once
+    }
+    houseKeeperService.stop();
+    try {
+      //should fail because the TransactionBatch timed out
+      txnBatch.commit();
+    }
+    catch(TransactionError e) {
+      Assert.assertTrue("Expected aborted transaction", e.getCause() instanceof TxnAbortedException);
+    }
+    txnBatch.close();
+    connection.close();
+  }
   @Test
   public void testTransactionBatchEmptyAbort() throws Exception {
     // 1) to partitioned table

http://git-wip-us.apache.org/repos/asf/hive/blob/97a6cd35/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java
----------------------------------------------------------------------
diff --git a/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java b/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java
index 65117c4..ae500bf 100644
--- a/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java
+++ b/metastore/src/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java
@@ -6225,6 +6225,7 @@ public class HiveMetaStore extends ThriftHiveMetastore {
           startCompactorInitiator(conf);
           startCompactorWorkers(conf);
           startCompactorCleaner(conf);
+          startHouseKeeperService(conf);
         } catch (Throwable e) {
           LOG.error("Failure when starting the compactor, compactions may not happen, " +
               StringUtils.stringifyException(e));
@@ -6284,4 +6285,21 @@ public class HiveMetaStore extends ThriftHiveMetastore {
     thread.init(new AtomicBoolean(), new AtomicBoolean());
     thread.start();
   }
+  private static void startHouseKeeperService(HiveConf conf) throws Exception {
+    if(!HiveConf.getBoolVar(conf, HiveConf.ConfVars.HIVE_COMPACTOR_INITIATOR_ON)) {
+      return;
+    }
+    Class c = Class.forName("org.apache.hadoop.hive.ql.txn.AcidHouseKeeperService");
+    //todo: when metastore adds orderly-shutdown logic, houseKeeper.stop()
+    //should be called form it
+    HouseKeeperService houseKeeper = (HouseKeeperService)c.newInstance();
+    try {
+      houseKeeper.start(conf);
+    }
+    catch (Exception ex) {
+      LOG.fatal("Failed to start " + houseKeeper.getClass() +
+        ".  The system will not handle " + houseKeeper.getServiceDescription()  +
+        ".  Root Cause: ", ex);
+    }
+  }
 }

http://git-wip-us.apache.org/repos/asf/hive/blob/97a6cd35/metastore/src/java/org/apache/hadoop/hive/metastore/HouseKeeperService.java
----------------------------------------------------------------------
diff --git a/metastore/src/java/org/apache/hadoop/hive/metastore/HouseKeeperService.java b/metastore/src/java/org/apache/hadoop/hive/metastore/HouseKeeperService.java
new file mode 100644
index 0000000..eb4ea93
--- /dev/null
+++ b/metastore/src/java/org/apache/hadoop/hive/metastore/HouseKeeperService.java
@@ -0,0 +1,39 @@
+/**
+ * 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.metastore;
+
+import org.apache.hadoop.hive.common.classification.InterfaceAudience;
+import org.apache.hadoop.hive.common.classification.InterfaceStability;
+import org.apache.hadoop.hive.conf.HiveConf;
+
+/**
+ * Runs arbitrary background logic inside the metastore service.  
+ */
+@InterfaceAudience.LimitedPrivate({"Hive"})
+@InterfaceStability.Evolving
+public interface HouseKeeperService {
+  public void start(HiveConf hiveConf) throws Exception;
+  /**
+   * Should perform orderly shutdown
+   */
+  public void stop();
+  /**
+   * Returns short description of services this module provides.
+   */
+  public String getServiceDescription();
+}

http://git-wip-us.apache.org/repos/asf/hive/blob/97a6cd35/metastore/src/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java
----------------------------------------------------------------------
diff --git a/metastore/src/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java b/metastore/src/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java
index 88e007c..795b2d9 100644
--- a/metastore/src/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java
+++ b/metastore/src/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java
@@ -46,6 +46,11 @@ import java.util.concurrent.TimeUnit;
 /**
  * A handler to answer transaction related calls that come into the metastore
  * server.
+ *
+ * Note on log messages:  Please include txnid:X and lockid info
+ * {@link org.apache.hadoop.hive.common.JavaUtils#lockIdToString(long)} in all messages.
+ * The txnid:X and lockid:Y matches how Thrift object toString() methods are generated,
+ * so keeping the format consistent makes grep'ing the logs much easier.
  */
 public class TxnHandler {
   // Compactor states
@@ -212,7 +217,6 @@ public class TxnHandler {
       Statement stmt = null;
       try {
         dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED);
-        timeOutTxns(dbConn);
         stmt = dbConn.createStatement();
         String s = "select ntxn_next - 1 from NEXT_TXN_ID";
         LOG.debug("Going to execute query <" + s + ">");
@@ -463,8 +467,6 @@ public class TxnHandler {
       try {
         dbConn = getDbConn(Connection.TRANSACTION_SERIALIZABLE);
         long extLockId = rqst.getLockid();
-        // Clean up timed out locks
-        timeOutLocks(dbConn);
 
         // Heartbeat on the lockid first, to assure that our lock is still valid.
         // Then look up the lock info (hopefully in the cache).  If these locks
@@ -1361,8 +1363,6 @@ public class TxnHandler {
     // and prevent other operations (such as committing transactions, showing locks,
     // etc.) that should not interfere with this one.
     synchronized (lockLock) {
-      // Clean up timed out locks before we attempt to acquire any.
-      timeOutLocks(dbConn);
       Statement stmt = null;
       try {
         stmt = dbConn.createStatement();
@@ -1732,15 +1732,22 @@ public class TxnHandler {
       LOG.debug("Going to execute query <" + s + ">");
       ResultSet rs = stmt.executeQuery(s);
       if (!rs.next()) {
+        s = "select count(*) from COMPLETED_TXN_COMPONENTS where CTC_TXNID = " + txnid;
+        ResultSet rs2 = stmt.executeQuery(s);
+        boolean alreadyCommitted = rs2.next() && rs2.getInt(1) > 0;
         LOG.debug("Going to rollback");
         dbConn.rollback();
+        if(alreadyCommitted) {
+          //makes the message more informative - helps to find bugs in client code
+          throw new NoSuchTxnException("Transaction " + JavaUtils.txnIdToString(txnid) +
" is already committed.");
+        }
         throw new NoSuchTxnException("No such transaction " + JavaUtils.txnIdToString(txnid));
       }
       if (rs.getString(1).charAt(0) == TXN_ABORTED) {
         LOG.debug("Going to rollback");
         dbConn.rollback();
         throw new TxnAbortedException("Transaction " + JavaUtils.txnIdToString(txnid) +
-          " already aborted");
+          " already aborted");//todo: add time of abort, which is not currently tracked
       }
       s = "update TXNS set txn_last_heartbeat = " + now +
         " where txn_id = " + txnid;
@@ -1802,61 +1809,121 @@ public class TxnHandler {
     }
   }
 
-  // Clean time out locks from the database.  This does a commit,
+  // Clean time out locks from the database not associated with a transactions, i.e. locks
+  // for read-only autoCommit=true statements.  This does a commit,
   // and thus should be done before any calls to heartbeat that will leave
   // open transactions.
-  private void timeOutLocks(Connection dbConn) throws SQLException, MetaException {
-    long now = getDbTime(dbConn);
+  private void timeOutLocks(Connection dbConn) {
     Statement stmt = null;
     try {
+      long now = getDbTime(dbConn);
       stmt = dbConn.createStatement();
       // Remove any timed out locks from the table.
       String s = "delete from HIVE_LOCKS where hl_last_heartbeat < " +
-        (now - timeout);
+        (now - timeout) + " and (hl_txnid = 0 or hl_txnid is NULL)";//when txnid is >
0, the lock is
+      //associated with a txn and is handled by performTimeOuts()
+      //want to avoid expiring locks for a txn w/o expiring the txn itself
       LOG.debug("Going to execute update <" + s + ">");
-      stmt.executeUpdate(s);
+      int deletedLocks = stmt.executeUpdate(s);
+      if(deletedLocks > 0) {
+        LOG.info("Deleted " + deletedLocks + " locks from HIVE_LOCKS due to timeout");
+      }
       LOG.debug("Going to commit");
       dbConn.commit();
+    }
+    catch(SQLException ex) {
+      LOG.error("Failed to purge timedout locks due to: " + getMessage(ex), ex);
+    }
+    catch(Exception ex) {
+      LOG.error("Failed to purge timedout locks due to: " + ex.getMessage(), ex);
     } finally {
       closeStmt(stmt);
     }
   }
 
-  // Abort timed out transactions.  This does a commit,
-  // and thus should be done before any calls to heartbeat that will leave
-  // open transactions on the underlying database.
-  private void timeOutTxns(Connection dbConn) throws SQLException, MetaException, RetryException
{
-    long now = getDbTime(dbConn);
+  /**
+   * Suppose you have a query "select a,b from T" and you want to limit the result set
+   * to the first 5 rows.  The mechanism to do that differs in different DB.
+   * Make {@code noSelectsqlQuery} to be "a,b from T" and this method will return the
+   * appropriately modified row limiting query.
+   */
+  private String addLimitClause(Connection dbConn, int numRows, String noSelectsqlQuery)
throws MetaException {
+    DatabaseProduct prod = determineDatabaseProduct(dbConn);
+    switch (prod) {
+      case DERBY:
+        //http://db.apache.org/derby/docs/10.7/ref/rrefsqljoffsetfetch.html
+        return "select " + noSelectsqlQuery + " fetch first " + numRows + " rows only";
+      case MYSQL:
+        //http://www.postgresql.org/docs/7.3/static/queries-limit.html
+      case POSTGRES:
+        //https://dev.mysql.com/doc/refman/5.0/en/select.html
+        return "select " + noSelectsqlQuery + " limit " + numRows;
+      case ORACLE:
+        //newer versions (12c and later) support OFFSET/FETCH
+        return "select * from (select " + noSelectsqlQuery + ") where rownum <= " + numRows;
+      case SQLSERVER:
+        //newer versions (2012 and later) support OFFSET/FETCH
+        //https://msdn.microsoft.com/en-us/library/ms189463.aspx
+        return "select TOP(" + numRows + ") " + noSelectsqlQuery;
+      default:
+        String msg = "Unrecognized database product name <" + prod + ">";
+        LOG.error(msg);
+        throw new MetaException(msg);
+    }
+  }
+  /**
+   * This will find transactions that have timed out and abort them.
+   * Will also delete locks which are not associated with a transaction and have timed out
+   * Tries to keep transactions (against metastore db) small to reduce lock contention.
+   */
+  public void performTimeOuts() {
+    Connection dbConn = null;
     Statement stmt = null;
+    ResultSet rs = null;
     try {
-      stmt = dbConn.createStatement();
-      // Abort any timed out locks from the table.
-      String s = "select txn_id from TXNS where txn_state = '" + TXN_OPEN +
-        "' and txn_last_heartbeat <  " + (now - timeout);
-      LOG.debug("Going to execute query <" + s + ">");
-      ResultSet rs = stmt.executeQuery(s);
-      List<Long> deadTxns = new ArrayList<Long>();
-      // Limit the number of timed out transactions we do in one pass to keep from generating
a
-      // huge delete statement
-      do {
-        deadTxns.clear();
-        for (int i = 0; i <  TIMED_OUT_TXN_ABORT_BATCH_SIZE && rs.next(); i++)
{
-          deadTxns.add(rs.getLong(1));
+      dbConn = getDbConn(Connection.TRANSACTION_SERIALIZABLE);
+      long now = getDbTime(dbConn);
+      timeOutLocks(dbConn);
+      while(true) {
+        stmt = dbConn.createStatement();
+        String s = " txn_id from TXNS where txn_state = '" + TXN_OPEN +
+          "' and txn_last_heartbeat <  " + (now - timeout);
+        s = addLimitClause(dbConn, 2500, s);
+        LOG.debug("Going to execute query <" + s + ">");
+        rs = stmt.executeQuery(s);
+        if(!rs.next()) {
+          return;//no more timedout txns
         }
-        // We don't care whether all of the transactions get deleted or not,
-        // if some didn't it most likely means someone else deleted them in the interum
-        if (deadTxns.size() > 0) abortTxns(dbConn, deadTxns);
-      } while (deadTxns.size() > 0);
-      LOG.debug("Going to commit");
-      dbConn.commit();
-    } catch (SQLException e) {
-      LOG.debug("Going to rollback");
-      rollbackDBConn(dbConn);
-      checkRetryable(dbConn, e, "abortTxn");
-      throw new MetaException("Unable to update transaction database "
-        + StringUtils.stringifyException(e));
-    } finally {
-      closeStmt(stmt);
+        List<List<Long>> timedOutTxns = new ArrayList<>();
+        List<Long> currentBatch = new ArrayList<>(TIMED_OUT_TXN_ABORT_BATCH_SIZE);
+        timedOutTxns.add(currentBatch);
+        do {
+          currentBatch.add(rs.getLong(1));
+          if(currentBatch.size() == TIMED_OUT_TXN_ABORT_BATCH_SIZE) {
+            currentBatch = new ArrayList<>(TIMED_OUT_TXN_ABORT_BATCH_SIZE);
+            timedOutTxns.add(currentBatch);
+          }
+        } while(rs.next());
+        close(rs, stmt, null);
+        dbConn.commit();
+        for(List<Long> batchToAbort : timedOutTxns) {
+          abortTxns(dbConn, batchToAbort);
+          dbConn.commit();
+          //todo: add TXNS.COMMENT filed and set it to 'aborted by system due to timeout'
+          LOG.info("Aborted the following transactions due to timeout: " + timedOutTxns.toString());
+        }
+        int numTxnsAborted = (timedOutTxns.size() - 1) * TIMED_OUT_TXN_ABORT_BATCH_SIZE +
+          timedOutTxns.get(timedOutTxns.size() - 1).size();
+        LOG.info("Aborted " + numTxnsAborted + " transactions due to timeout");
+      }
+    } catch (SQLException ex) {
+      LOG.warn("Aborting timedout transactions failed due to " + getMessage(ex), ex);
+    }
+    catch(MetaException e) {
+      LOG.warn("Aborting timedout transactions failed due to " + e.getMessage(), e);
+    }
+    finally {
+      close(rs, stmt, dbConn);
     }
   }
 

http://git-wip-us.apache.org/repos/asf/hive/blob/97a6cd35/metastore/src/test/org/apache/hadoop/hive/metastore/txn/TestTxnHandler.java
----------------------------------------------------------------------
diff --git a/metastore/src/test/org/apache/hadoop/hive/metastore/txn/TestTxnHandler.java b/metastore/src/test/org/apache/hadoop/hive/metastore/txn/TestTxnHandler.java
index 6dc0bd3..59114fe 100644
--- a/metastore/src/test/org/apache/hadoop/hive/metastore/txn/TestTxnHandler.java
+++ b/metastore/src/test/org/apache/hadoop/hive/metastore/txn/TestTxnHandler.java
@@ -986,7 +986,8 @@ public class TestTxnHandler {
       LockRequest req = new LockRequest(components, "me", "localhost");
       LockResponse res = txnHandler.lock(req);
       assertTrue(res.getState() == LockState.ACQUIRED);
-      Thread.currentThread().sleep(10);
+      Thread.sleep(10);
+      txnHandler.performTimeOuts();
       txnHandler.checkLock(new CheckLockRequest(res.getLockid()));
       fail("Told there was a lock, when it should have timed out.");
     } catch (NoSuchLockException e) {
@@ -1000,8 +1001,8 @@ public class TestTxnHandler {
     long timeout = txnHandler.setTimeout(1);
     try {
       txnHandler.openTxns(new OpenTxnRequest(503, "me", "localhost"));
-      Thread.currentThread().sleep(10);
-      txnHandler.getOpenTxns();
+      Thread.sleep(10);
+      txnHandler.performTimeOuts();
       GetOpenTxnsInfoResponse rsp = txnHandler.getOpenTxnsInfo();
       int numAborted = 0;
       for (TxnInfo txnInfo : rsp.getOpen_txns()) {

http://git-wip-us.apache.org/repos/asf/hive/blob/97a6cd35/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 c0c1b2e..4030075 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/Driver.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/Driver.java
@@ -977,11 +977,19 @@ public class Driver implements CommandProcessor {
         return 10;
       }
 
-      boolean existingTxn = txnMgr.isTxnOpen();
+      boolean initiatingTransaction = false;
+      boolean readOnlyQueryInAutoCommit = false;
       if((txnMgr.getAutoCommit() && haveAcidWrite()) || plan.getOperation() == HiveOperation.START_TRANSACTION
||
         (!txnMgr.getAutoCommit() && startTxnImplicitly)) {
+        if(txnMgr.isTxnOpen()) {
+          throw new RuntimeException("Already have an open transaction txnid:" + txnMgr.getCurrentTxnId());
+        }
         // We are writing to tables in an ACID compliant way, so we need to open a transaction
         txnMgr.openTxn(userFromUGI);
+        initiatingTransaction = true;
+      }
+      else {
+        readOnlyQueryInAutoCommit = txnMgr.getAutoCommit() && plan.getOperation()
== HiveOperation.QUERY && !haveAcidWrite();
       }
       // Set the transaction id in all of the acid file sinks
       if (haveAcidWrite()) {
@@ -997,9 +1005,9 @@ public class Driver implements CommandProcessor {
       For multi-stmt txns this is not sufficient and will be managed via WriteSet tracking
       in the lock manager.*/
       txnMgr.acquireLocks(plan, ctx, userFromUGI);
-      if(!existingTxn) {
+      if(initiatingTransaction || readOnlyQueryInAutoCommit) {
         //For multi-stmt txns we should record the snapshot when txn starts but
-        // don't update it after that until txn completes.  Thus the check for {@code existingTxn}
+        // don't update it after that until txn completes.  Thus the check for {@code initiatingTransaction}
         //For autoCommit=true, Read-only statements, txn is implicit, i.e. lock in the snapshot
         //for each statement.
         recordValidTxns();
@@ -1294,6 +1302,7 @@ public class Driver implements CommandProcessor {
   }
 
   private CommandProcessorResponse rollback(CommandProcessorResponse cpr) {
+    //console.printError(cpr.toString());
     try {
       releaseLocksAndCommitOrRollback(ctx.getHiveLocks(), false);
     }

http://git-wip-us.apache.org/repos/asf/hive/blob/97a6cd35/ql/src/java/org/apache/hadoop/hive/ql/txn/AcidHouseKeeperService.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/txn/AcidHouseKeeperService.java b/ql/src/java/org/apache/hadoop/hive/ql/txn/AcidHouseKeeperService.java
new file mode 100644
index 0000000..d22ca8d
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/txn/AcidHouseKeeperService.java
@@ -0,0 +1,104 @@
+/**
+ * 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.txn;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.HouseKeeperService;
+import org.apache.hadoop.hive.metastore.txn.TxnHandler;
+import org.apache.hadoop.hive.ql.lockmgr.HiveTxnManager;
+import org.apache.hadoop.hive.ql.lockmgr.TxnManagerFactory;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Performs background tasks for Transaction management in Hive.
+ * Runs inside Hive Metastore Service.
+ */
+public class AcidHouseKeeperService implements HouseKeeperService {
+  private static final Log LOG = LogFactory.getLog(AcidHouseKeeperService.class);
+  private ScheduledExecutorService pool = null;
+  private AtomicInteger isAliveCounter = new AtomicInteger(Integer.MIN_VALUE);
+  @Override
+  public void start(HiveConf hiveConf) throws Exception {
+    HiveTxnManager mgr = TxnManagerFactory.getTxnManagerFactory().getTxnManager(hiveConf);
+    if(!mgr.supportsAcid()) {
+      LOG.info(AcidHouseKeeperService.class.getName() + " not started since " +
+        mgr.getClass().getName()  + " does not support Acid.");
+      return;//there are no transactions in this case
+    }
+    pool = Executors.newScheduledThreadPool(1, new ThreadFactory() {
+      private final AtomicInteger threadCounter = new AtomicInteger();
+      @Override
+      public Thread newThread(Runnable r) {
+        return new Thread(r, "DeadTxnReaper-" + threadCounter.getAndIncrement());
+      }
+    });
+    TimeUnit tu = TimeUnit.MILLISECONDS;
+    pool.scheduleAtFixedRate(new TimedoutTxnReaper(hiveConf, this),
+      hiveConf.getTimeVar(HiveConf.ConfVars.HIVE_TIMEDOUT_TXN_REAPER_START, tu),
+      hiveConf.getTimeVar(HiveConf.ConfVars.HIVE_TIMEDOUT_TXN_REAPER_INTERVAL, tu),
+      TimeUnit.MILLISECONDS);
+    LOG.info("Started " + this.getClass().getName() + " with delay/interval = " +
+      hiveConf.getTimeVar(HiveConf.ConfVars.HIVE_TIMEDOUT_TXN_REAPER_START, tu) + "/" +
+      hiveConf.getTimeVar(HiveConf.ConfVars.HIVE_TIMEDOUT_TXN_REAPER_INTERVAL, tu) + " "
+ tu);
+  }
+  @Override
+  public void stop() {
+    if(pool != null && !pool.isShutdown()) {
+      pool.shutdown();
+    }
+    pool = null;
+  }
+  @Override
+  public String getServiceDescription() {
+    return "Abort expired transactions";
+  }
+  private static final class TimedoutTxnReaper implements Runnable {
+    private final TxnHandler txnHandler;
+    private final AcidHouseKeeperService owner;
+    private TimedoutTxnReaper(HiveConf hiveConf, AcidHouseKeeperService owner) {
+      txnHandler = new TxnHandler(hiveConf);
+      this.owner = owner;
+    }
+    @Override
+    public void run() {
+      try {
+        txnHandler.performTimeOuts();
+        owner.isAliveCounter.incrementAndGet();
+        LOG.info("timeout reaper ran");
+      }
+      catch(Throwable t) {
+        LOG.fatal("Serious error in " + Thread.currentThread().getName() + ": " + t.getMessage(),
t);
+      }
+    }
+  }
+
+  /**
+   * This is used for testing only.  Each time the housekeeper runs, counter is incremented
by 1.
+   * Starts with {@link java.lang.Integer#MIN_VALUE}
+   */
+  public int getIsAliveCounter() {
+    return isAliveCounter.get();
+  }
+}

http://git-wip-us.apache.org/repos/asf/hive/blob/97a6cd35/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorMR.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorMR.java b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorMR.java
index 6c77ba4..8e431b2 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorMR.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorMR.java
@@ -28,6 +28,7 @@ import org.apache.hadoop.hive.common.JavaUtils;
 import org.apache.hadoop.hive.common.ValidTxnList;
 import org.apache.hadoop.hive.common.ValidReadTxnList;
 import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.api.CompactionType;
 import org.apache.hadoop.hive.metastore.api.FieldSchema;
 import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
 import org.apache.hadoop.hive.metastore.api.Table;
@@ -52,6 +53,7 @@ import org.apache.hadoop.mapred.OutputCollector;
 import org.apache.hadoop.mapred.OutputCommitter;
 import org.apache.hadoop.mapred.RecordReader;
 import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapred.RunningJob;
 import org.apache.hadoop.mapred.TaskAttemptContext;
 import org.apache.hadoop.mapred.lib.NullOutputFormat;
 import org.apache.hadoop.util.StringUtils;
@@ -183,7 +185,9 @@ public class CompactorMR {
     LOG.debug("Setting minimum transaction to " + minTxn);
     LOG.debug("Setting maximume transaction to " + maxTxn);
 
-    JobClient.runJob(job).waitForCompletion();
+    RunningJob rj = JobClient.runJob(job);
+    LOG.info("Submitted " + (isMajor ? CompactionType.MAJOR : CompactionType.MINOR) + " compaction
job '" + jobName + "' with jobID=" + rj.getID());
+    rj.waitForCompletion();
     su.gatherStats();
   }
 

http://git-wip-us.apache.org/repos/asf/hive/blob/97a6cd35/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Initiator.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Initiator.java b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Initiator.java
index 32a9ef8..73715c6 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Initiator.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Initiator.java
@@ -52,6 +52,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * A class to initiate compactions.  This will run in a separate thread.
+ * It's critical that there exactly 1 of these in a given warehouse.
  */
 public class Initiator extends CompactorThread {
   static final private String CLASS_NAME = Initiator.class.getName();

http://git-wip-us.apache.org/repos/asf/hive/blob/97a6cd35/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Worker.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Worker.java b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Worker.java
index e164661..0548117 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Worker.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Worker.java
@@ -264,7 +264,7 @@ public class Worker extends CompactorThread {
         sb.append(colName).append(",");
       }
       sb.setLength(sb.length() - 1);//remove trailing ,
-      LOG.debug("running '" + sb.toString() + "'");
+      LOG.info("running '" + sb.toString() + "'");
       Driver d = new Driver(conf, userName);
       SessionState localSession = null;
       if(SessionState.get() == null) {

http://git-wip-us.apache.org/repos/asf/hive/blob/97a6cd35/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands.java
----------------------------------------------------------------------
diff --git a/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands.java b/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands.java
index c73621f..e13e6eb 100644
--- a/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands.java
+++ b/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands.java
@@ -9,6 +9,7 @@ import org.apache.hadoop.hive.ql.io.AcidUtils;
 import org.apache.hadoop.hive.ql.io.orc.FileDump;
 import org.apache.hadoop.hive.ql.processors.CommandProcessorResponse;
 import org.apache.hadoop.hive.ql.session.SessionState;
+import org.apache.hadoop.hive.ql.txn.AcidHouseKeeperService;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -24,6 +25,7 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Comparator;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 
 /**
  * The LockManager is not ready, but for no-concurrency straight-line path we can 
@@ -388,6 +390,25 @@ public class TestTxnCommands {
     int[][] updatedData = {{1,7},{3,7},{5,2},{5,3}};
     Assert.assertEquals("Bulk update failed", stringifyValues(updatedData), rs);
   }
+  @Test
+  public void testTimeOutReaper() throws Exception {
+    runStatementOnDriver("set autocommit false");
+    runStatementOnDriver("start transaction");
+    runStatementOnDriver("delete from " + Table.ACIDTBL + " where a = 5");
+    //make sure currently running txn is considered aborted by housekeeper
+    hiveConf.setTimeVar(HiveConf.ConfVars.HIVE_TIMEDOUT_TXN_REAPER_START, 0, TimeUnit.SECONDS);
+    hiveConf.setTimeVar(HiveConf.ConfVars.HIVE_TXN_TIMEOUT, 1, TimeUnit.MILLISECONDS);
+    AcidHouseKeeperService houseKeeperService = new AcidHouseKeeperService();
+    //this will abort the txn
+    houseKeeperService.start(hiveConf);
+    while(houseKeeperService.getIsAliveCounter() <= Integer.MIN_VALUE) {
+      Thread.sleep(100);//make sure it has run at least once
+    }
+    houseKeeperService.stop();
+    //this should fail because txn aborted due to timeout
+    CommandProcessorResponse cpr = runStatementOnDriverNegative("delete from " + Table.ACIDTBL
+ " where a = 5");
+    Assert.assertTrue("Actual: " + cpr.getErrorMessage(), cpr.getErrorMessage().contains("Transaction
manager has aborted the transaction txnid:1"));
+  }
 
   /**
    * takes raw data and turns it into a string as if from Driver.getResults()

http://git-wip-us.apache.org/repos/asf/hive/blob/97a6cd35/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java
----------------------------------------------------------------------
diff --git a/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java b/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java
index 57e4fb9..58c2fca 100644
--- a/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java
+++ b/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java
@@ -131,6 +131,7 @@ public class TestTxnCommands2 {
     hiveConf.setBoolVar(HiveConf.ConfVars.HIVEOPTINDEXFILTER, enablePPD);//enables ORC PPD
     int[][] tableData = {{1,2},{3,4}};
     runStatementOnDriver("insert into " + Table.ACIDTBL + "(a,b) " + makeValuesClause(tableData));
+    List<String> rs2 = runStatementOnDriver("select a,b from " + Table.ACIDTBL + "
where a > 1 order by a,b");
     runStatementOnDriver("alter table "+ Table.ACIDTBL + " compact 'MAJOR'");
     Worker t = new Worker();
     t.setThreadId((int) t.getId());

http://git-wip-us.apache.org/repos/asf/hive/blob/97a6cd35/ql/src/test/org/apache/hadoop/hive/ql/lockmgr/TestDbTxnManager.java
----------------------------------------------------------------------
diff --git a/ql/src/test/org/apache/hadoop/hive/ql/lockmgr/TestDbTxnManager.java b/ql/src/test/org/apache/hadoop/hive/ql/lockmgr/TestDbTxnManager.java
index f57350d..db119e1 100644
--- a/ql/src/test/org/apache/hadoop/hive/ql/lockmgr/TestDbTxnManager.java
+++ b/ql/src/test/org/apache/hadoop/hive/ql/lockmgr/TestDbTxnManager.java
@@ -29,10 +29,10 @@ import org.apache.hadoop.hive.ql.hooks.WriteEntity;
 import org.apache.hadoop.hive.ql.metadata.DummyPartition;import org.apache.hadoop.hive.ql.metadata.Partition;
 import org.apache.hadoop.hive.ql.metadata.Table;
 import org.apache.hadoop.hive.ql.session.SessionState;
+import org.apache.hadoop.hive.ql.txn.AcidHouseKeeperService;
 import org.apache.log4j.Level;
 import org.apache.log4j.LogManager;
 import static org.hamcrest.CoreMatchers.is;
-import static org.hamcrest.CoreMatchers.not;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -49,6 +49,7 @@ public class TestDbTxnManager {
 
   private HiveConf conf = new HiveConf();
   private HiveTxnManager txnMgr;
+  private AcidHouseKeeperService houseKeeperService = null;
   private Context ctx;
   private int nextInput;
   private int nextOutput;
@@ -56,7 +57,6 @@ public class TestDbTxnManager {
   HashSet<WriteEntity> writeEntities;
 
   public TestDbTxnManager() throws Exception {
-    conf.setTimeVar(HiveConf.ConfVars.HIVE_TXN_TIMEOUT, 500, TimeUnit.MILLISECONDS);
     TxnDbUtil.setConfValues(conf);
     SessionState.start(conf);
     ctx = new Context(conf);
@@ -179,14 +179,30 @@ public class TestDbTxnManager {
     locks = txnMgr.getLockManager().getLocks(false, false);
     Assert.assertEquals(0, locks.size());
   }
+
+  /**
+   * aborts timed out transactions
+   */
+  private void runReaper() throws Exception {
+    int lastCount = houseKeeperService.getIsAliveCounter();
+    houseKeeperService.start(conf);
+    while(houseKeeperService.getIsAliveCounter() <= lastCount) {
+      try {
+        Thread.sleep(100);//make sure it has run at least once
+      }
+      catch(InterruptedException ex) {
+        //...
+      }
+    }
+    houseKeeperService.stop();
+  }
   @Test
   public void testExceptions() throws Exception {
     WriteEntity we = addPartitionOutput(newTable(true), WriteEntity.WriteType.INSERT);
     QueryPlan qp = new MockQueryPlan(this);
     txnMgr.acquireLocks(qp, ctx, "PeterI");
     txnMgr.openTxn("NicholasII");
-    Thread.sleep(1000);//let txn timeout
-    txnMgr.getValidTxns();
+    runReaper();
     LockException exception = null;
     try {
       txnMgr.commitTxn();
@@ -198,8 +214,7 @@ public class TestDbTxnManager {
     Assert.assertEquals("Wrong Exception1", ErrorMsg.TXN_ABORTED, exception.getCanonicalErrorMsg());
     exception = null;
     txnMgr.openTxn("AlexanderIII");
-    Thread.sleep(1000);
-    txnMgr.getValidTxns();
+    runReaper();
     try {
       txnMgr.rollbackTxn();
     }
@@ -213,8 +228,7 @@ public class TestDbTxnManager {
     txnMgr.acquireLocks(qp, ctx, "PeterI");
     List<HiveLock> locks = ctx.getHiveLocks();
     Assert.assertThat("Unexpected lock count", locks.size(), is(1));
-    Thread.sleep(1000);
-    txnMgr.getValidTxns();
+    runReaper();
     try {
       txnMgr.heartbeat();
     }
@@ -341,7 +355,6 @@ public class TestDbTxnManager {
     Assert.assertTrue(sawException);
   }
 
-
   @Before
   public void setUp() throws Exception {
     TxnDbUtil.prepDb();
@@ -351,10 +364,14 @@ public class TestDbTxnManager {
     nextOutput = 1;
     readEntities = new HashSet<ReadEntity>();
     writeEntities = new HashSet<WriteEntity>();
+    conf.setTimeVar(HiveConf.ConfVars.HIVE_TIMEDOUT_TXN_REAPER_START, 0, TimeUnit.SECONDS);
+    conf.setTimeVar(HiveConf.ConfVars.HIVE_TXN_TIMEOUT, 1, TimeUnit.MILLISECONDS);
+    houseKeeperService = new AcidHouseKeeperService();
   }
 
   @After
   public void tearDown() throws Exception {
+    if(houseKeeperService != null) houseKeeperService.stop();
     if (txnMgr != null) txnMgr.closeTxnManager();
     TxnDbUtil.cleanDb();
   }


Mime
View raw message