You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by ek...@apache.org on 2015/08/15 19:23:08 UTC
hive git commit: HIVE-11317 - ACID: Improve transaction Abort logic
due to timeout (Eugene Koifman, reviewed by Alan Gates)
Repository: hive
Updated Branches:
refs/heads/master e8329ee00 -> 97a6cd35a
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/master
Commit: 97a6cd35a444315885008f11c20c7c28249bd42c
Parents: e8329ee
Author: Eugene Koifman <ek...@hortonworks.com>
Authored: Sat Aug 15 10:22:55 2015 -0700
Committer: Eugene Koifman <ek...@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();
}