hive-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ekoif...@apache.org
Subject hive git commit: HIVE-13014 RetryingMetaStoreClient is retrying acid related calls too aggressievley(Eugene Koifman, reviewed by Alan Gates)
Date Tue, 24 Jan 2017 21:01:13 GMT
Repository: hive
Updated Branches:
  refs/heads/master cc3fd84ee -> 1d7f4b75d


HIVE-13014 RetryingMetaStoreClient is retrying acid related calls too aggressievley(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/1d7f4b75
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/1d7f4b75
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/1d7f4b75

Branch: refs/heads/master
Commit: 1d7f4b75de43db1201f9d494d7a2100f923a7ad6
Parents: cc3fd84
Author: Eugene Koifman <ekoifman@hortonworks.com>
Authored: Tue Jan 24 13:00:43 2017 -0800
Committer: Eugene Koifman <ekoifman@hortonworks.com>
Committed: Tue Jan 24 13:00:43 2017 -0800

----------------------------------------------------------------------
 .../common/classification/RetrySemantics.java   |  38 ++++
 .../hadoop/hive/metastore/IMetaStoreClient.java |   2 +
 .../hive/metastore/RetryingMetaStoreClient.java |  13 +-
 .../metastore/txn/CompactionTxnHandler.java     |  32 +++-
 .../hadoop/hive/metastore/txn/TxnHandler.java   | 180 ++++++++++++++++---
 .../hadoop/hive/metastore/txn/TxnStore.java     |  38 +++-
 .../hive/metastore/txn/TestTxnHandler.java      |  42 +++--
 .../apache/hadoop/hive/ql/TestTxnCommands2.java |   2 +-
 .../hive/ql/lockmgr/TestDbTxnManager.java       |  21 ++-
 9 files changed, 320 insertions(+), 48 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hive/blob/1d7f4b75/common/src/java/org/apache/hadoop/hive/common/classification/RetrySemantics.java
----------------------------------------------------------------------
diff --git a/common/src/java/org/apache/hadoop/hive/common/classification/RetrySemantics.java b/common/src/java/org/apache/hadoop/hive/common/classification/RetrySemantics.java
new file mode 100644
index 0000000..abad45e
--- /dev/null
+++ b/common/src/java/org/apache/hadoop/hive/common/classification/RetrySemantics.java
@@ -0,0 +1,38 @@
+package org.apache.hadoop.hive.common.classification;
+
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * These annotations are meant to indicate how to handle retry logic.
+ * Initially meant for Metastore API when made across a network, i.e. asynchronously where
+ * the response may not reach the caller and thus it cannot know if the operation was actually
+ * performed on the server.
+ * @see RetryingMetastoreClient
+ */
+@InterfaceStability.Evolving
+@InterfaceAudience.LimitedPrivate("Hive developer")
+public class RetrySemantics {
+  @Retention(RetentionPolicy.RUNTIME)
+  @Target(ElementType.METHOD)
+  public @interface Idempotent {
+    String[] value() default "";
+    int maxRetryCount() default Integer.MAX_VALUE;
+    int delayMs() default 100;
+  }
+  @Retention(RetentionPolicy.RUNTIME)
+  @Target(ElementType.METHOD)
+  public @interface ReadOnly {/*trivially retry-able*/}
+  @Retention(RetentionPolicy.RUNTIME)
+  @Target(ElementType.METHOD)
+  public @interface CannotRetry {}
+  @Retention(RetentionPolicy.RUNTIME)
+  @Target(ElementType.METHOD)
+  public @interface SafeToRetry {
+    /*may not be Idempotent but is safe to retry*/
+    String[] value() default "";
+  }
+}

http://git-wip-us.apache.org/repos/asf/hive/blob/1d7f4b75/metastore/src/java/org/apache/hadoop/hive/metastore/IMetaStoreClient.java
----------------------------------------------------------------------
diff --git a/metastore/src/java/org/apache/hadoop/hive/metastore/IMetaStoreClient.java b/metastore/src/java/org/apache/hadoop/hive/metastore/IMetaStoreClient.java
index fb61db1..84ec332 100644
--- a/metastore/src/java/org/apache/hadoop/hive/metastore/IMetaStoreClient.java
+++ b/metastore/src/java/org/apache/hadoop/hive/metastore/IMetaStoreClient.java
@@ -30,6 +30,7 @@ import org.apache.hadoop.hive.common.ValidTxnList;
 import org.apache.hadoop.hive.common.classification.InterfaceAudience;
 import org.apache.hadoop.hive.common.classification.InterfaceAudience.Public;
 import org.apache.hadoop.hive.common.classification.InterfaceStability.Evolving;
+import org.apache.hadoop.hive.common.classification.RetrySemantics;
 import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.metastore.TableType;
 import org.apache.hadoop.hive.metastore.annotation.NoReconnect;
@@ -1347,6 +1348,7 @@ public interface IMetaStoreClient {
    * aborted.  This can result from the transaction timing out.
    * @throws TException
    */
+  @RetrySemantics.CannotRetry
   LockResponse lock(LockRequest request)
       throws NoSuchTxnException, TxnAbortedException, TException;
 

http://git-wip-us.apache.org/repos/asf/hive/blob/1d7f4b75/metastore/src/java/org/apache/hadoop/hive/metastore/RetryingMetaStoreClient.java
----------------------------------------------------------------------
diff --git a/metastore/src/java/org/apache/hadoop/hive/metastore/RetryingMetaStoreClient.java b/metastore/src/java/org/apache/hadoop/hive/metastore/RetryingMetaStoreClient.java
index 96d8248..a6545a9 100644
--- a/metastore/src/java/org/apache/hadoop/hive/metastore/RetryingMetaStoreClient.java
+++ b/metastore/src/java/org/apache/hadoop/hive/metastore/RetryingMetaStoreClient.java
@@ -19,6 +19,7 @@
 package org.apache.hadoop.hive.metastore;
 
 import java.io.IOException;
+import java.lang.annotation.Annotation;
 import java.lang.reflect.InvocationHandler;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
@@ -28,6 +29,7 @@ import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.hadoop.hive.common.classification.RetrySemantics;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.hadoop.hive.common.classification.InterfaceAudience.Public;
@@ -142,6 +144,15 @@ public class RetryingMetaStoreClient implements InvocationHandler {
     TException caughtException = null;
 
     boolean allowReconnect = ! method.isAnnotationPresent(NoReconnect.class);
+    boolean allowRetry = true;
+    Annotation[] directives = method.getDeclaredAnnotations();
+    if(directives != null) {
+      for(Annotation a : directives) {
+        if(a instanceof RetrySemantics.CannotRetry) {
+          allowRetry = false;
+        }
+      }
+    }
 
     while (true) {
       try {
@@ -200,7 +211,7 @@ public class RetryingMetaStoreClient implements InvocationHandler {
       }
 
 
-      if (retriesMade >= retryLimit || base.isLocalMetaStore()) {
+      if (retriesMade >= retryLimit || base.isLocalMetaStore() || !allowRetry) {
         throw caughtException;
       }
       retriesMade++;

http://git-wip-us.apache.org/repos/asf/hive/blob/1d7f4b75/metastore/src/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
----------------------------------------------------------------------
diff --git a/metastore/src/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java b/metastore/src/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
index 545244b..60839fa 100644
--- a/metastore/src/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
+++ b/metastore/src/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.hive.metastore.txn;
 
+import org.apache.hadoop.hive.common.classification.RetrySemantics;
 import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.metastore.api.CompactionType;
 import org.apache.hadoop.hive.metastore.api.MetaException;
@@ -55,6 +56,8 @@ class CompactionTxnHandler extends TxnHandler {
    * @return list of CompactionInfo structs.  These will not have id, type,
    * or runAs set since these are only potential compactions not actual ones.
    */
+  @Override
+  @RetrySemantics.ReadOnly
   public Set<CompactionInfo> findPotentialCompactions(int maxAborted) throws MetaException {
     Connection dbConn = null;
     Set<CompactionInfo> response = new HashSet<CompactionInfo>();
@@ -117,6 +120,8 @@ class CompactionTxnHandler extends TxnHandler {
    * @param cq_id id of this entry in the queue
    * @param user user to run the jobs as
    */
+  @Override
+  @RetrySemantics.Idempotent
   public void setRunAs(long cq_id, String user) throws MetaException {
     try {
       Connection dbConn = null;
@@ -154,6 +159,8 @@ class CompactionTxnHandler extends TxnHandler {
    * @param workerId id of the worker calling this, will be recorded in the db
    * @return an info element for this compaction request, or null if there is no work to do now.
    */
+  @Override
+  @RetrySemantics.SafeToRetry
   public CompactionInfo findNextToCompact(String workerId) throws MetaException {
     try {
       Connection dbConn = null;
@@ -225,6 +232,8 @@ class CompactionTxnHandler extends TxnHandler {
    * and put it in the ready to clean state.
    * @param info info on the compaction entry to mark as compacted.
    */
+  @Override
+  @RetrySemantics.SafeToRetry
   public void markCompacted(CompactionInfo info) throws MetaException {
     try {
       Connection dbConn = null;
@@ -264,6 +273,8 @@ class CompactionTxnHandler extends TxnHandler {
    * be cleaned.
    * @return information on the entry in the queue.
    */
+  @Override
+  @RetrySemantics.ReadOnly
   public List<CompactionInfo> findReadyToClean() throws MetaException {
     Connection dbConn = null;
     List<CompactionInfo> rc = new ArrayList<CompactionInfo>();
@@ -317,6 +328,8 @@ class CompactionTxnHandler extends TxnHandler {
    * 
    * @param info info on the compaction entry to remove
    */
+  @Override
+  @RetrySemantics.CannotRetry
   public void markCleaned(CompactionInfo info) throws MetaException {
     try {
       Connection dbConn = null;
@@ -429,6 +442,8 @@ class CompactionTxnHandler extends TxnHandler {
    * txns exist can be that now work was done in this txn (e.g. Streaming opened TransactionBatch and
    * abandoned it w/o doing any work) or due to {@link #markCleaned(CompactionInfo)} being called.
    */
+  @Override
+  @RetrySemantics.SafeToRetry
   public void cleanEmptyAbortedTxns() throws MetaException {
     try {
       Connection dbConn = null;
@@ -493,6 +508,8 @@ class CompactionTxnHandler extends TxnHandler {
    * @param hostname Name of this host.  It is assumed this prefixes the thread's worker id,
    *                 so that like hostname% will match the worker id.
    */
+  @Override
+  @RetrySemantics.Idempotent
   public void revokeFromLocalWorkers(String hostname) throws MetaException {
     try {
       Connection dbConn = null;
@@ -535,6 +552,8 @@ class CompactionTxnHandler extends TxnHandler {
    * @param timeout number of milliseconds since start time that should elapse before a worker is
    *                declared dead.
    */
+  @Override
+  @RetrySemantics.Idempotent
   public void revokeTimedoutWorkers(long timeout) throws MetaException {
     try {
       Connection dbConn = null;
@@ -575,6 +594,8 @@ class CompactionTxnHandler extends TxnHandler {
    * table level stats are examined.
    * @throws MetaException
    */
+  @Override
+  @RetrySemantics.ReadOnly
   public List<String> findColumnsWithStats(CompactionInfo ci) throws MetaException {
     Connection dbConn = null;
     Statement stmt = null;
@@ -631,6 +652,8 @@ class CompactionTxnHandler extends TxnHandler {
    * Record the highest txn id that the {@code ci} compaction job will pay attention to.
    * This is the highest resolved txn id, i.e. such that there are no open txns with lower ids.
    */
+  @Override
+  @RetrySemantics.Idempotent
   public void setCompactionHighestTxnId(CompactionInfo ci, long highestTxnId) throws MetaException {
     Connection dbConn = null;
     Statement stmt = null;
@@ -696,6 +719,8 @@ class CompactionTxnHandler extends TxnHandler {
    * it's not recent.
    * @throws MetaException
    */
+  @Override
+  @RetrySemantics.SafeToRetry
   public void purgeCompactionHistory() throws MetaException {
     Connection dbConn = null;
     Statement stmt = null;
@@ -762,7 +787,7 @@ class CompactionTxnHandler extends TxnHandler {
    * this ensures that the number of failed compaction entries retained is > than number of failed
    * compaction threshold which prevents new compactions from being scheduled.
    */
-  public int getFailedCompactionRetention() {
+  private int getFailedCompactionRetention() {
     int failedThreshold = conf.getIntVar(HiveConf.ConfVars.COMPACTOR_INITIATOR_FAILED_THRESHOLD);
     int failedRetention = conf.getIntVar(HiveConf.ConfVars.COMPACTOR_HISTORY_RETENTION_FAILED);
     if(failedRetention < failedThreshold) {
@@ -783,6 +808,8 @@ class CompactionTxnHandler extends TxnHandler {
    * That would be a meta operations, i.e. first find all partitions for this table (which have 
    * txn info) and schedule each compaction separately.  This avoids complications in this logic.
    */
+  @Override
+  @RetrySemantics.ReadOnly
   public boolean checkFailedCompactions(CompactionInfo ci) throws MetaException {
     Connection dbConn = null;
     Statement stmt = null;
@@ -829,6 +856,8 @@ class CompactionTxnHandler extends TxnHandler {
    * If there is no entry in compaction_queue, it means Initiator failed to even schedule a compaction,
    * which we record as ATTEMPTED_STATE entry in history.
    */
+  @Override
+  @RetrySemantics.CannotRetry
   public void markFailed(CompactionInfo ci) throws MetaException {//todo: this should not throw
     //todo: this should take "comment" as parameter to set in CC_META_INFO to provide some context for the failure
     try {
@@ -894,6 +923,7 @@ class CompactionTxnHandler extends TxnHandler {
     }
   }
   @Override
+  @RetrySemantics.Idempotent
   public void setHadoopJobId(String hadoopJobId, long id) {
     try {
       Connection dbConn = null;

http://git-wip-us.apache.org/repos/asf/hive/blob/1d7f4b75/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 75a58c6..805db34 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
@@ -30,10 +30,10 @@ import org.apache.commons.lang.NotImplementedException;
 import org.apache.hadoop.hive.common.ServerUtils;
 import org.apache.hadoop.hive.common.classification.InterfaceAudience;
 import org.apache.hadoop.hive.common.classification.InterfaceStability;
+import org.apache.hadoop.hive.common.classification.RetrySemantics;
 import org.apache.hadoop.hive.metastore.DatabaseProduct;
 import org.apache.hadoop.hive.metastore.HouseKeeperService;
 import org.apache.hadoop.hive.metastore.Warehouse;
-import org.apache.thrift.TException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.commons.dbcp.PoolingDataSource;
@@ -101,6 +101,17 @@ import java.util.regex.Pattern;
  * It's imperative that any operation on a txn (e.g. commit), ensure (atomically) that this txn is
  * still valid and active.  In the code this is usually achieved at the same time the txn record
  * is locked for some operation.
+ * 
+ * Note on retry logic:
+ * Metastore has retry logic in both {@link org.apache.hadoop.hive.metastore.RetryingMetaStoreClient}
+ * and {@link org.apache.hadoop.hive.metastore.RetryingHMSHandler}.  The retry logic there is very
+ * generic and is not aware whether the operations are idempotent or not.  (This is separate from
+ * retry logic here in TxnHander which can/does retry DB errors intelligently).  The worst case is
+ * when an op here issues a successful commit against the RDBMS but the calling stack doesn't
+ * receive the ack and retries.  (If an op fails before commit, it's trivially idempotent)
+ * Thus the ops here need to be made idempotent as much as possible or
+ * the metstore call stack should have logic not to retry.  There are {@link RetrySemantics}
+ * annotations to document the behavior.
  */
 @InterfaceAudience.Private
 @InterfaceStability.Evolving
@@ -120,6 +131,8 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
   // Transaction states
   static final protected char TXN_ABORTED = 'a';
   static final protected char TXN_OPEN = 'o';
+  //todo: make these like OperationType and remove above char constatns
+  enum TxnStatus {OPEN, ABORTED, COMMITTED, UNKNOWN}
 
   // Lock states
   static final protected char LOCK_ACQUIRED = 'a';
@@ -263,7 +276,8 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
       dumpConfig = false;
     }
   }
-
+  @Override
+  @RetrySemantics.ReadOnly
   public GetOpenTxnsInfoResponse getOpenTxnsInfo() throws MetaException {
     try {
       // We need to figure out the current transaction number and the list of
@@ -339,7 +353,8 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
       return getOpenTxnsInfo();
     }
   }
-
+  @Override
+  @RetrySemantics.ReadOnly
   public GetOpenTxnsResponse getOpenTxns() throws MetaException {
     try {
       // We need to figure out the current transaction number and the list of
@@ -414,6 +429,12 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
     }
   }
 
+  /**
+   * Retry-by-caller note:
+   * Worst case, it will leave an open txn which will timeout.
+   */
+  @Override
+  @RetrySemantics.Idempotent
   public OpenTxnsResponse openTxns(OpenTxnRequest rqst) throws MetaException {
     if (openTxnsCounter == null) {
       synchronized (TxnHandler.class) {
@@ -515,7 +536,8 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
       return openTxns(rqst);
     }
   }
-
+  @Override
+  @RetrySemantics.Idempotent
   public void abortTxn(AbortTxnRequest rqst) throws NoSuchTxnException, MetaException, TxnAbortedException {
     long txnid = rqst.getTxnid();
     try {
@@ -525,10 +547,14 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
         lockInternal();
         dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED);
         if (abortTxns(dbConn, Collections.singletonList(txnid), true) != 1) {
-          LOG.debug("Going to rollback");
-          dbConn.rollback();
           stmt = dbConn.createStatement();
-          ensureValidTxn(dbConn, txnid, stmt);
+          TxnStatus status = findTxnState(txnid,stmt);
+          if(status == TxnStatus.ABORTED) {
+            LOG.info("abortTxn(" + JavaUtils.txnIdToString(txnid) +
+              ") requested by it is already " + TxnStatus.ABORTED);
+            return;
+          }
+          raiseTxnUnexpectedState(status, txnid);
         }
 
         LOG.debug("Going to commit");
@@ -547,7 +573,8 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
       abortTxn(rqst);
     }
   }
-
+  @Override
+  @RetrySemantics.Idempotent
   public void abortTxns(AbortTxnsRequest rqst) throws NoSuchTxnException, MetaException {
     List<Long> txnids = rqst.getTxn_ids();
     try {
@@ -556,7 +583,7 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
         dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED);
         int numAborted = abortTxns(dbConn, txnids, false);
         if (numAborted != txnids.size()) {
-          LOG.warn("Abort Transactions command only abort " + numAborted + " out of " +
+          LOG.warn("Abort Transactions command only aborted " + numAborted + " out of " +
               txnids.size() + " transactions. It's possible that the other " +
               (txnids.size() - numAborted) +
               " transactions have been aborted or committed, or the transaction ids are invalid.");
@@ -602,6 +629,8 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
    * txnid:2, which is possible if commitTxn() and openTxnx() is not mutexed)
    * 'x' would be updated to the same value by both, i.e. lost update. 
    */
+  @Override
+  @RetrySemantics.Idempotent("No-op if already committed")
   public void commitTxn(CommitTxnRequest rqst)
     throws NoSuchTxnException, TxnAbortedException,  MetaException {
     long txnid = rqst.getTxnid();
@@ -622,9 +651,19 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
          */
         lockHandle = lockTransactionRecord(stmt, txnid, TXN_OPEN);
         if (lockHandle == null) {
-          //this also ensures that txn is still there and in expected state (hasn't been timed out)
-          ensureValidTxn(dbConn, txnid, stmt);
+          //if here, txn was not found (in expected state)
+          TxnStatus actualTxnStatus = findTxnState(txnid, stmt);
+          if(actualTxnStatus == TxnStatus.COMMITTED) {
+            /**
+             * This makes the operation idempotent
+             * (assume that this is most likely due to retry logic)
+             */
+            LOG.info("Nth commitTxn(" + JavaUtils.txnIdToString(txnid) + ") msg");
+            return;
+          }
+          raiseTxnUnexpectedState(actualTxnStatus, txnid);
           shouldNeverHappen(txnid);
+          //dbConn is rolled back in finally{}
         }
         String conflictSQLSuffix = "from TXN_COMPONENTS where tc_txnid=" + txnid + " and tc_operation_type IN(" +
           quoteChar(OpertaionType.UPDATE.sqlConst) + "," + quoteChar(OpertaionType.DELETE.sqlConst) + ")";
@@ -645,9 +684,17 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
           }
           long commitId = commitIdRs.getLong(1);
           Savepoint undoWriteSetForCurrentTxn = dbConn.setSavepoint();
+          /**
+           * "select distinct" is used below because
+           * 1. once we get to multi-statement txns, we only care to record that something was updated once
+           * 2. if {@link #addDynamicPartitions(AddDynamicPartitions)} is retried by caller it my create
+           *  duplicate entries in TXN_COMPONENTS
+           * but we want to add a PK on WRITE_SET which won't have unique rows w/o this distinct
+           * even if it includes all of it's columns
+           */
           int numCompsWritten = stmt.executeUpdate(
             "insert into WRITE_SET (ws_database, ws_table, ws_partition, ws_txnid, ws_commit_id, ws_operation_type)" +
-            " select tc_database, tc_table, tc_partition, tc_txnid, " + commitId + ", tc_operation_type " + conflictSQLSuffix);
+            " select distinct tc_database, tc_table, tc_partition, tc_txnid, " + commitId + ", tc_operation_type " + conflictSQLSuffix);
           /**
            * see if there are any overlapping txns wrote the same element, i.e. have a conflict
            * Since entire commit operation is mutexed wrt other start/commit ops,
@@ -750,6 +797,7 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
     }
   }
   @Override
+  @RetrySemantics.SafeToRetry
   public void performWriteSetGC() {
     Connection dbConn = null;
     Statement stmt = null;
@@ -798,7 +846,17 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
    * connection (but separate transactions).  This avoid some flakiness in BONECP where if you
    * perform an operation on 1 connection and immediately get another fron the pool, the 2nd one
    * doesn't see results of the first.
+   * 
+   * Retry-by-caller note: If the call to lock is from a transaction, then in the worst case
+   * there will be a duplicate set of locks but both sets will belong to the same txn so they 
+   * will not conflict with each other.  For locks w/o txn context (i.e. read-only query), this
+   * may lead to deadlock (at least a long wait).  (e.g. 1st call creates locks in {@code LOCK_WAITING}
+   * mode and response gets lost.  Then {@link org.apache.hadoop.hive.metastore.RetryingMetaStoreClient}
+   * retries, and enqueues another set of locks in LOCK_WAITING.  The 2nd LockResponse is delivered
+   * to the DbLockManager, which will keep dong {@link #checkLock(CheckLockRequest)} until the 1st
+   * set of locks times out.
    */
+  @RetrySemantics.CannotRetry
   public LockResponse lock(LockRequest rqst) throws NoSuchTxnException, TxnAbortedException, MetaException {
     ConnectionLockIdPair connAndLockId = enqueueLockWithRetry(rqst);
     try {
@@ -1063,7 +1121,12 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
    * {@link #checkLock(java.sql.Connection, long)}  must run at SERIALIZABLE (make sure some lock we are checking
    * against doesn't move from W to A in another txn) but this method can heartbeat in
    * separate txn at READ_COMMITTED.
+   * 
+   * Retry-by-caller note:
+   * Retryable because {@link #checkLock(Connection, long)} is
    */
+  @Override
+  @RetrySemantics.SafeToRetry
   public LockResponse checkLock(CheckLockRequest rqst)
     throws NoSuchTxnException, NoSuchLockException, TxnAbortedException, MetaException {
     try {
@@ -1112,6 +1175,7 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
    * heartbeat/performTimeout which are update/delete of HIVE_LOCKS thus will be locked as needed by db.
    * since this only removes from HIVE_LOCKS at worst some lock acquire is delayed
    */
+  @RetrySemantics.Idempotent
   public void unlock(UnlockRequest rqst)
     throws NoSuchLockException, TxnOpenException, MetaException {
     try {
@@ -1146,11 +1210,15 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
           if(info == null) {
             //didn't find any lock with extLockId but at ReadCommitted there is a possibility that
             //it existed when above delete ran but it didn't have the expected state.
-            LOG.error("No lock in " + LOCK_WAITING + " mode found for unlock(" + rqst + ")");
-            throw new NoSuchLockException("No such lock " + JavaUtils.lockIdToString(extLockId));
+            LOG.info("No lock in " + LOCK_WAITING + " mode found for unlock(" +
+              JavaUtils.lockIdToString(rqst.getLockid()) + ")");
+            //bail here to make the operation idempotent
+            return;
           }
           if(info.txnId != 0) {
             String msg = "Unlocking locks associated with transaction not permitted.  " + info;
+            //if a lock is associated with a txn we can only "unlock" if if it's in WAITING state
+            // which really means that the caller wants to give up waiting for the lock
             LOG.error(msg);
             throw new TxnOpenException(msg);
           }
@@ -1189,6 +1257,7 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
       this.e = e;
     }
   }
+  @RetrySemantics.ReadOnly
   public ShowLocksResponse showLocks(ShowLocksRequest rqst) throws MetaException {
     try {
       Connection dbConn = null;
@@ -1297,6 +1366,8 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
    * {@code ids} should only have txnid or lockid but not both, ideally.
    * Currently DBTxnManager.heartbeat() enforces this.
    */
+  @Override
+  @RetrySemantics.SafeToRetry
   public void heartbeat(HeartbeatRequest ids)
     throws NoSuchTxnException,  NoSuchLockException, TxnAbortedException, MetaException {
     try {
@@ -1318,7 +1389,8 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
       heartbeat(ids);
     }
   }
-
+  @Override
+  @RetrySemantics.SafeToRetry
   public HeartbeatTxnRangeResponse heartbeatTxnRange(HeartbeatTxnRangeRequest rqst)
     throws MetaException {
     try {
@@ -1379,6 +1451,7 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
     return id;
   }
   @Override
+  @RetrySemantics.Idempotent
   public CompactionResponse compact(CompactionRequest rqst) throws MetaException {
     // Put a compaction request in the queue.
     try {
@@ -1503,6 +1576,7 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
         return Character.toString(s);
     }
   }
+  @RetrySemantics.ReadOnly
   public ShowCompactResponse showCompact(ShowCompactRequest rqst) throws MetaException {
     ShowCompactResponse response = new ShowCompactResponse(new ArrayList<ShowCompactResponseElement>());
     Connection dbConn = null;
@@ -1574,6 +1648,13 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
       + JavaUtils.lockIdToString(extLockId) + " " + intLockId);
   }
 
+  /**
+   * Retry-by-caller note:
+   * This may be retried after dbConn.commit.  At worst, it will create duplicate entries in
+   * TXN_COMPONENTS which won't affect anything.  See more comments in {@link #commitTxn(CommitTxnRequest)}
+   */
+  @Override
+  @RetrySemantics.SafeToRetry
   public void addDynamicPartitions(AddDynamicPartitions rqst)
       throws NoSuchTxnException,  TxnAbortedException, MetaException {
     Connection dbConn = null;
@@ -1629,7 +1710,11 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
   /**
    * Clean up corresponding records in metastore tables when corresponding object is dropped,
    * specifically: TXN_COMPONENTS, COMPLETED_TXN_COMPONENTS, COMPACTION_QUEUE, COMPLETED_COMPACTIONS
+   * Retry-by-caller note: this is only idempotent assuming it's only called by dropTable/Db/etc
+   * operations.
    */
+  @Override
+  @RetrySemantics.Idempotent
   public void cleanupRecords(HiveObjectType type, Database db, Table table,
                              Iterator<Partition> partitionIterator) throws MetaException {
     try {
@@ -2322,6 +2407,7 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
    * If they happen to be for the same txnid, there will be a WW conflict (in MS DB), if different txnid,
    * checkLock() will in the worst case keep locks in Waiting state a little longer.
    */
+  @RetrySemantics.SafeToRetry("See @SafeToRetry")
   private LockResponse checkLock(Connection dbConn, long extLockId)
     throws NoSuchLockException, NoSuchTxnException, TxnAbortedException, MetaException, SQLException {
     TxnStore.MutexAPI.LockHandle handle =  null;
@@ -2331,7 +2417,7 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
     /**
      * todo: Longer term we should pass this from client somehow - this would be an optimization;  once
      * that is in place make sure to build and test "writeSet" below using OperationType not LockType
-     * With SP we assume that the query modifies exactly the partitions it locked.  (not entirely
+     * With Static Partitions we assume that the query modifies exactly the partitions it locked.  (not entirely
      * realistic since Update/Delete may have some predicate that filters out all records out of
      * some partition(s), but plausible).  For DP, we acquire locks very wide (all known partitions),
      * but for most queries only a fraction will actually be updated.  #addDynamicPartitions() tells
@@ -2517,6 +2603,7 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
         // If we've found it and it's already been marked acquired,
         // then just look at the other locks.
         if (locks[index].state == LockState.ACQUIRED) {
+          /**this is what makes this method @SafeToRetry*/
           continue;
         }
 
@@ -2720,6 +2807,55 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
     }
   }
 
+  /**
+   * Returns the state of the transaction iff it's able to determine it.  Some cases where it cannot:
+   * 1. txnid was Aborted/Committed and then GC'd (compacted)
+   * 2. txnid was committed but it didn't modify anything (nothing in COMPLETED_TXN_COMPONENTS)
+   */
+  private TxnStatus findTxnState(long txnid, Statement stmt) throws SQLException, MetaException {
+    String s = "select txn_state from TXNS where txn_id = " + txnid;
+    LOG.debug("Going to execute query <" + s + ">");
+    ResultSet rs = stmt.executeQuery(s);
+    if (!rs.next()) {
+      s = sqlGenerator.addLimitClause(1, "1 from COMPLETED_TXN_COMPONENTS where CTC_TXNID = " + txnid);
+      LOG.debug("Going to execute query <" + s + ">");
+      ResultSet rs2 = stmt.executeQuery(s);
+      if(rs2.next()) {
+        return TxnStatus.COMMITTED;
+      }
+      //could also check WRITE_SET but that seems overkill
+      return TxnStatus.UNKNOWN;
+    }
+    char txnState = rs.getString(1).charAt(0);
+    if (txnState == TXN_ABORTED) {
+      return TxnStatus.ABORTED;
+    }
+    assert txnState == TXN_OPEN : "we found it in TXNS but it's not ABORTED, so must be OPEN";
+    return TxnStatus.OPEN;
+  }
+
+  /**
+   * Used to raise an informative error when the caller expected a txn in a particular TxnStatus
+   * but found it in some other status
+   */
+  private static void raiseTxnUnexpectedState(TxnStatus actualStatus, long txnid) 
+    throws NoSuchTxnException, TxnAbortedException {
+    switch (actualStatus) {
+      case ABORTED:
+        throw new TxnAbortedException("Transaction " + JavaUtils.txnIdToString(txnid) + " already aborted");
+      case COMMITTED:
+        throw new NoSuchTxnException("Transaction " + JavaUtils.txnIdToString(txnid) + " is already committed.");
+      case UNKNOWN:
+        throw new NoSuchTxnException("No such transaction " + JavaUtils.txnIdToString(txnid));
+      case OPEN:
+        throw new NoSuchTxnException(JavaUtils.txnIdToString(txnid) + " is " + TxnStatus.OPEN);
+      default:
+        throw new IllegalArgumentException("Unknown TxnStatus " + actualStatus);
+    }
+  }
+  /**
+   * Returns the state of the transaction with {@code txnid} or throws if {@code raiseError} is true.
+   */
   private static void ensureValidTxn(Connection dbConn, long txnid, Statement stmt)
       throws SQLException, NoSuchTxnException, TxnAbortedException {
     // We need to check whether this transaction is valid and open
@@ -2734,7 +2870,7 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
       //possible for for multi-stmt txns
       boolean alreadyCommitted = rs2.next() && rs2.getInt(1) > 0;
       LOG.debug("Going to rollback");
-      dbConn.rollback();
+      rollbackDBConn(dbConn);
       if(alreadyCommitted) {
         //makes the message more informative - helps to find bugs in client code
         throw new NoSuchTxnException("Transaction " + JavaUtils.txnIdToString(txnid) + " is already committed.");
@@ -2743,7 +2879,7 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
     }
     if (rs.getString(1).charAt(0) == TXN_ABORTED) {
       LOG.debug("Going to rollback");
-      dbConn.rollback();
+      rollbackDBConn(dbConn);
       throw new TxnAbortedException("Transaction " + JavaUtils.txnIdToString(txnid) +
         " already aborted");//todo: add time of abort, which is not currently tracked.  Requires schema change
     }
@@ -2869,6 +3005,7 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
    * 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.
    */
+  @RetrySemantics.Idempotent
   public void performTimeOuts() {
     Connection dbConn = null;
     Statement stmt = null;
@@ -2939,7 +3076,8 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
       close(rs, stmt, dbConn);
     }
   }
-
+  @Override
+  @RetrySemantics.ReadOnly
   public void countOpenTxns() throws MetaException {
     Connection dbConn = null;
     Statement stmt = null;
@@ -3244,6 +3382,7 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
     }
   }
   @Override
+  @RetrySemantics.Idempotent
   public MutexAPI getMutexAPI() {
     return this;
   }
@@ -3360,6 +3499,7 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
   }
   /**
    * Helper class that generates SQL queries with syntax specific to target DB
+   * todo: why throw MetaException?
    */
   @VisibleForTesting
   static final class SQLGenerator {
@@ -3459,7 +3599,7 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
     }
     /**
      * 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.
+     * to the first 5 rows.  The mechanism to do that differs in different DBs.
      * Make {@code noSelectsqlQuery} to be "a,b from T" and this method will return the
      * appropriately modified row limiting query.
      *

http://git-wip-us.apache.org/repos/asf/hive/blob/1d7f4b75/metastore/src/java/org/apache/hadoop/hive/metastore/txn/TxnStore.java
----------------------------------------------------------------------
diff --git a/metastore/src/java/org/apache/hadoop/hive/metastore/txn/TxnStore.java b/metastore/src/java/org/apache/hadoop/hive/metastore/txn/TxnStore.java
index 879ae55..041d55b 100644
--- a/metastore/src/java/org/apache/hadoop/hive/metastore/txn/TxnStore.java
+++ b/metastore/src/java/org/apache/hadoop/hive/metastore/txn/TxnStore.java
@@ -20,6 +20,7 @@ package org.apache.hadoop.hive.metastore.txn;
 import com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.hive.common.classification.InterfaceAudience;
 import org.apache.hadoop.hive.common.classification.InterfaceStability;
+import org.apache.hadoop.hive.common.classification.RetrySemantics;
 import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.metastore.api.*;
 
@@ -68,6 +69,7 @@ public interface TxnStore {
    * @return information about open transactions
    * @throws MetaException
    */
+  @RetrySemantics.ReadOnly
   public GetOpenTxnsInfoResponse getOpenTxnsInfo() throws MetaException;
 
   /**
@@ -75,12 +77,14 @@ public interface TxnStore {
    * @return list of open transactions, as well as a high water mark.
    * @throws MetaException
    */
+  @RetrySemantics.ReadOnly
   public GetOpenTxnsResponse getOpenTxns() throws MetaException;
 
   /**
    * Get the count for open transactions.
    * @throws MetaException
    */
+  @RetrySemantics.ReadOnly
   public void countOpenTxns() throws MetaException;
 
   /**
@@ -89,6 +93,7 @@ public interface TxnStore {
    * @return information on opened transactions
    * @throws MetaException
    */
+  @RetrySemantics.Idempotent
   public OpenTxnsResponse openTxns(OpenTxnRequest rqst) throws MetaException;
 
   /**
@@ -97,6 +102,7 @@ public interface TxnStore {
    * @throws NoSuchTxnException
    * @throws MetaException
    */
+  @RetrySemantics.Idempotent
   public void abortTxn(AbortTxnRequest rqst) throws NoSuchTxnException, MetaException, TxnAbortedException;
 
   /**
@@ -105,6 +111,7 @@ public interface TxnStore {
    * @throws NoSuchTxnException
    * @throws MetaException
    */
+  @RetrySemantics.Idempotent
   public void abortTxns(AbortTxnsRequest rqst) throws NoSuchTxnException, MetaException;
 
   /**
@@ -114,6 +121,7 @@ public interface TxnStore {
    * @throws TxnAbortedException
    * @throws MetaException
    */
+  @RetrySemantics.Idempotent
   public void commitTxn(CommitTxnRequest rqst)
     throws NoSuchTxnException, TxnAbortedException,  MetaException;
 
@@ -126,6 +134,7 @@ public interface TxnStore {
    * @throws TxnAbortedException
    * @throws MetaException
    */
+  @RetrySemantics.CannotRetry
   public LockResponse lock(LockRequest rqst)
     throws NoSuchTxnException, TxnAbortedException, MetaException;
 
@@ -139,6 +148,7 @@ public interface TxnStore {
    * @throws TxnAbortedException
    * @throws MetaException
    */
+  @RetrySemantics.SafeToRetry
   public LockResponse checkLock(CheckLockRequest rqst)
     throws NoSuchTxnException, NoSuchLockException, TxnAbortedException, MetaException;
 
@@ -151,6 +161,7 @@ public interface TxnStore {
    * @throws TxnOpenException
    * @throws MetaException
    */
+  @RetrySemantics.Idempotent
   public void unlock(UnlockRequest rqst)
     throws NoSuchLockException, TxnOpenException, MetaException;
 
@@ -160,6 +171,7 @@ public interface TxnStore {
    * @return lock information.
    * @throws MetaException
    */
+  @RetrySemantics.ReadOnly
   public ShowLocksResponse showLocks(ShowLocksRequest rqst) throws MetaException;
 
   /**
@@ -170,6 +182,7 @@ public interface TxnStore {
    * @throws TxnAbortedException
    * @throws MetaException
    */
+  @RetrySemantics.SafeToRetry
   public void heartbeat(HeartbeatRequest ids)
     throws NoSuchTxnException,  NoSuchLockException, TxnAbortedException, MetaException;
 
@@ -179,6 +192,7 @@ public interface TxnStore {
    * @return info on txns that were heartbeated
    * @throws MetaException
    */
+  @RetrySemantics.SafeToRetry
   public HeartbeatTxnRangeResponse heartbeatTxnRange(HeartbeatTxnRangeRequest rqst)
     throws MetaException;
 
@@ -189,6 +203,7 @@ public interface TxnStore {
    * @return id of the compaction that has been started or existing id if this resource is already scheduled
    * @throws MetaException
    */
+  @RetrySemantics.Idempotent
   public CompactionResponse compact(CompactionRequest rqst) throws MetaException;
 
   /**
@@ -197,6 +212,7 @@ public interface TxnStore {
    * @return compaction information
    * @throws MetaException
    */
+  @RetrySemantics.ReadOnly
   public ShowCompactResponse showCompact(ShowCompactRequest rqst) throws MetaException;
 
   /**
@@ -206,6 +222,7 @@ public interface TxnStore {
    * @throws TxnAbortedException
    * @throws MetaException
    */
+  @RetrySemantics.SafeToRetry
   public void addDynamicPartitions(AddDynamicPartitions rqst)
       throws NoSuchTxnException,  TxnAbortedException, MetaException;
 
@@ -217,12 +234,14 @@ public interface TxnStore {
    * @param partitionIterator partition iterator
    * @throws MetaException
    */
+  @RetrySemantics.Idempotent
   public void cleanupRecords(HiveObjectType type, Database db, Table table,
                              Iterator<Partition> partitionIterator) throws MetaException;
 
   /**
    * Timeout transactions and/or locks.  This should only be called by the compactor.
    */
+  @RetrySemantics.Idempotent
   public void performTimeOuts();
 
   /**
@@ -234,6 +253,7 @@ public interface TxnStore {
    * @return list of CompactionInfo structs.  These will not have id, type,
    * or runAs set since these are only potential compactions not actual ones.
    */
+  @RetrySemantics.ReadOnly
   public Set<CompactionInfo> findPotentialCompactions(int maxAborted) throws MetaException;
 
   /**
@@ -242,6 +262,7 @@ public interface TxnStore {
    * @param cq_id id of this entry in the queue
    * @param user user to run the jobs as
    */
+  @RetrySemantics.Idempotent
   public void setRunAs(long cq_id, String user) throws MetaException;
 
   /**
@@ -250,6 +271,7 @@ public interface TxnStore {
    * @param workerId id of the worker calling this, will be recorded in the db
    * @return an info element for this compaction request, or null if there is no work to do now.
    */
+  @RetrySemantics.ReadOnly
   public CompactionInfo findNextToCompact(String workerId) throws MetaException;
 
   /**
@@ -257,6 +279,7 @@ public interface TxnStore {
    * and put it in the ready to clean state.
    * @param info info on the compaction entry to mark as compacted.
    */
+  @RetrySemantics.SafeToRetry
   public void markCompacted(CompactionInfo info) throws MetaException;
 
   /**
@@ -264,6 +287,7 @@ public interface TxnStore {
    * be cleaned.
    * @return information on the entry in the queue.
    */
+  @RetrySemantics.ReadOnly
   public List<CompactionInfo> findReadyToClean() throws MetaException;
 
   /**
@@ -272,6 +296,7 @@ public interface TxnStore {
    * 
    * @param info info on the compaction entry to remove
    */
+  @RetrySemantics.CannotRetry
   public void markCleaned(CompactionInfo info) throws MetaException;
 
   /**
@@ -281,6 +306,7 @@ public interface TxnStore {
    * @param info information on the compaction that failed.
    * @throws MetaException
    */
+  @RetrySemantics.CannotRetry
   public void markFailed(CompactionInfo info) throws MetaException;
 
   /**
@@ -288,6 +314,7 @@ public interface TxnStore {
    * txns exist can be that now work was done in this txn (e.g. Streaming opened TransactionBatch and
    * abandoned it w/o doing any work) or due to {@link #markCleaned(CompactionInfo)} being called.
    */
+  @RetrySemantics.SafeToRetry
   public void cleanEmptyAbortedTxns() throws MetaException;
 
   /**
@@ -299,6 +326,7 @@ public interface TxnStore {
    * @param hostname Name of this host.  It is assumed this prefixes the thread's worker id,
    *                 so that like hostname% will match the worker id.
    */
+  @RetrySemantics.Idempotent
   public void revokeFromLocalWorkers(String hostname) throws MetaException;
 
   /**
@@ -310,6 +338,7 @@ public interface TxnStore {
    * @param timeout number of milliseconds since start time that should elapse before a worker is
    *                declared dead.
    */
+  @RetrySemantics.Idempotent
   public void revokeTimedoutWorkers(long timeout) throws MetaException;
 
   /**
@@ -318,11 +347,13 @@ public interface TxnStore {
    * table level stats are examined.
    * @throws MetaException
    */
+  @RetrySemantics.ReadOnly
   public List<String> findColumnsWithStats(CompactionInfo ci) throws MetaException;
 
   /**
    * Record the highest txn id that the {@code ci} compaction job will pay attention to.
    */
+  @RetrySemantics.Idempotent
   public void setCompactionHighestTxnId(CompactionInfo ci, long highestTxnId) throws MetaException;
 
   /**
@@ -333,12 +364,14 @@ public interface TxnStore {
    * it's not recent.
    * @throws MetaException
    */
+  @RetrySemantics.SafeToRetry
   public void purgeCompactionHistory() throws MetaException;
 
   /**
    * WriteSet tracking is used to ensure proper transaction isolation.  This method deletes the 
    * transaction metadata once it becomes unnecessary.  
    */
+  @RetrySemantics.SafeToRetry
   public void performWriteSetGC();
 
   /**
@@ -349,6 +382,7 @@ public interface TxnStore {
    * @return true if it is ok to compact, false if there have been too many failures.
    * @throws MetaException
    */
+  @RetrySemantics.ReadOnly
   public boolean checkFailedCompactions(CompactionInfo ci) throws MetaException;
 
   @VisibleForTesting
@@ -357,6 +391,7 @@ public interface TxnStore {
   @VisibleForTesting
   long setTimeout(long milliseconds);
 
+  @RetrySemantics.Idempotent
   public MutexAPI getMutexAPI();
 
   /**
@@ -382,7 +417,7 @@ public interface TxnStore {
     public void acquireLock(String key, LockHandle handle) throws MetaException;
     public static interface LockHandle {
       /**
-       * Releases all locks associcated with this handle.
+       * Releases all locks associated with this handle.
        */
       public void releaseLocks();
     }
@@ -393,5 +428,6 @@ public interface TxnStore {
    * it calls this to update the metadata.
    * @param id {@link CompactionInfo#id}
    */
+  @RetrySemantics.Idempotent
   public void setHadoopJobId(String hadoopJobId, long id);
 }

http://git-wip-us.apache.org/repos/asf/hive/blob/1d7f4b75/ql/src/test/org/apache/hadoop/hive/metastore/txn/TestTxnHandler.java
----------------------------------------------------------------------
diff --git a/ql/src/test/org/apache/hadoop/hive/metastore/txn/TestTxnHandler.java b/ql/src/test/org/apache/hadoop/hive/metastore/txn/TestTxnHandler.java
index 11cedb9..adfe98a 100644
--- a/ql/src/test/org/apache/hadoop/hive/metastore/txn/TestTxnHandler.java
+++ b/ql/src/test/org/apache/hadoop/hive/metastore/txn/TestTxnHandler.java
@@ -20,6 +20,7 @@ package org.apache.hadoop.hive.metastore.txn;
 import org.apache.hadoop.hive.common.JavaUtils;
 import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.metastore.api.AbortTxnRequest;
+import org.apache.hadoop.hive.metastore.api.AddDynamicPartitions;
 import org.apache.hadoop.hive.metastore.api.CheckLockRequest;
 import org.apache.hadoop.hive.metastore.api.CommitTxnRequest;
 import org.apache.hadoop.hive.metastore.api.CompactionRequest;
@@ -139,58 +140,69 @@ public class TestTxnHandler {
 
   @Test
   public void testAbortTxn() throws Exception {
-    OpenTxnsResponse openedTxns = txnHandler.openTxns(new OpenTxnRequest(2, "me", "localhost"));
+    OpenTxnsResponse openedTxns = txnHandler.openTxns(new OpenTxnRequest(3, "me", "localhost"));
     List<Long> txnList = openedTxns.getTxn_ids();
     long first = txnList.get(0);
     assertEquals(1L, first);
     long second = txnList.get(1);
     assertEquals(2L, second);
     txnHandler.abortTxn(new AbortTxnRequest(1));
+    List<String> parts = new ArrayList<String>();
+    parts.add("p=1");
+    AddDynamicPartitions adp = new AddDynamicPartitions(3, "default", "T", parts);
+    adp.setOperationType(DataOperationType.INSERT);
+    txnHandler.addDynamicPartitions(adp);
     GetOpenTxnsInfoResponse txnsInfo = txnHandler.getOpenTxnsInfo();
-    assertEquals(2L, txnsInfo.getTxn_high_water_mark());
-    assertEquals(2, txnsInfo.getOpen_txns().size());
+    assertEquals(3, txnsInfo.getTxn_high_water_mark());
+    assertEquals(3, txnsInfo.getOpen_txns().size());
     assertEquals(1L, txnsInfo.getOpen_txns().get(0).getId());
     assertEquals(TxnState.ABORTED, txnsInfo.getOpen_txns().get(0).getState());
     assertEquals(2L, txnsInfo.getOpen_txns().get(1).getId());
     assertEquals(TxnState.OPEN, txnsInfo.getOpen_txns().get(1).getState());
+    assertEquals(3, txnsInfo.getOpen_txns().get(2).getId());
+    assertEquals(TxnState.OPEN, txnsInfo.getOpen_txns().get(2).getState());
 
     GetOpenTxnsResponse txns = txnHandler.getOpenTxns();
-    assertEquals(2L, txns.getTxn_high_water_mark());
-    assertEquals(2, txns.getOpen_txns().size());
-    boolean[] saw = new boolean[3];
+    assertEquals(3, txns.getTxn_high_water_mark());
+    assertEquals(3, txns.getOpen_txns().size());
+    boolean[] saw = new boolean[4];
     for (int i = 0; i < saw.length; i++) saw[i] = false;
     for (Long tid : txns.getOpen_txns()) {
       saw[tid.intValue()] = true;
     }
     for (int i = 1; i < saw.length; i++) assertTrue(saw[i]);
     txnHandler.commitTxn(new CommitTxnRequest(2));
+    //this succeeds as abortTxn is idempotent
+    txnHandler.abortTxn(new AbortTxnRequest(1));
     boolean gotException = false;
     try {
-      txnHandler.abortTxn(new AbortTxnRequest(1));
+      txnHandler.abortTxn(new AbortTxnRequest(2));
     }
-    catch(TxnAbortedException ex) {
+    catch(NoSuchTxnException ex) {
       gotException = true;
-      Assert.assertEquals("Transaction " + JavaUtils.txnIdToString(1) + " already aborted", ex.getMessage());
+      //if this wasn't an empty txn, we'd get a better msg
+      Assert.assertEquals("No such transaction " + JavaUtils.txnIdToString(2), ex.getMessage());
     }
     Assert.assertTrue(gotException);
     gotException = false;
+    txnHandler.commitTxn(new CommitTxnRequest(3));
     try {
-      txnHandler.abortTxn(new AbortTxnRequest(2));
+      txnHandler.abortTxn(new AbortTxnRequest(3));
     }
     catch(NoSuchTxnException ex) {
       gotException = true;
-      //if this wasn't an empty txn, we'd get a better msg
-      //Assert.assertEquals("Transaction " + JavaUtils.txnIdToString(2) + " already committed.", ex.getMessage());
-      Assert.assertEquals("No such transaction " + JavaUtils.txnIdToString(2), ex.getMessage());
+      //txn 3 is not empty txn, so we get a better msg
+      Assert.assertEquals("Transaction " + JavaUtils.txnIdToString(3) + " is already committed.", ex.getMessage());
     }
     Assert.assertTrue(gotException);
+    
     gotException = false;
     try {
-      txnHandler.abortTxn(new AbortTxnRequest(3));
+      txnHandler.abortTxn(new AbortTxnRequest(4));
     }
     catch(NoSuchTxnException ex) {
       gotException = true;
-      Assert.assertEquals("No such transaction " + JavaUtils.txnIdToString(3), ex.getMessage());
+      Assert.assertEquals("No such transaction " + JavaUtils.txnIdToString(4), ex.getMessage());
     }
     Assert.assertTrue(gotException);
   }

http://git-wip-us.apache.org/repos/asf/hive/blob/1d7f4b75/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 5932d7e..af1f962 100644
--- a/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java
+++ b/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java
@@ -280,7 +280,7 @@ public class TestTxnCommands2 {
     runStatementOnDriver("insert into " + Table.ACIDTBL + "(a,b,c) " + makeValuesClause(moreTableData));
     List<String> rs0 = runStatementOnDriver("select a,b,c from " + Table.ACIDTBL + " where a > 0 order by a,b,c");
   }
-  @Ignore("not needed but useful for testing")
+//  @Ignore("not needed but useful for testing")
   @Test
   public void testNonAcidInsert() throws Exception {
     runStatementOnDriver("insert into " + Table.NONACIDORCTBL + "(a,b) values(1,2)");

http://git-wip-us.apache.org/repos/asf/hive/blob/1d7f4b75/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 8f26099..6c53538 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
@@ -20,11 +20,14 @@ package org.apache.hadoop.hive.ql.lockmgr;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.GetOpenTxnsInfoResponse;
 import org.apache.hadoop.hive.metastore.api.ShowLocksResponse;
 import org.apache.hadoop.hive.metastore.api.ShowLocksResponseElement;
+import org.apache.hadoop.hive.metastore.api.TxnState;
 import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
 import org.apache.hadoop.hive.metastore.txn.TxnDbUtil;
 import org.apache.hadoop.hive.metastore.txn.TxnStore;
+import org.apache.hadoop.hive.metastore.txn.TxnUtils;
 import org.apache.hadoop.hive.ql.Context;
 import org.apache.hadoop.hive.ql.ErrorMsg;
 import org.apache.hadoop.hive.ql.QueryPlan;
@@ -47,6 +50,8 @@ import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
+import static junit.framework.Assert.assertEquals;
+
 /**
  * Unit tests for {@link DbTxnManager}.
  * See additional tests in {@link org.apache.hadoop.hive.ql.lockmgr.TestDbTxnManager2}
@@ -229,15 +234,13 @@ public class TestDbTxnManager {
     exception = null;
     ((DbTxnManager) txnMgr).openTxn(ctx, "AlexanderIII", HiveConf.getTimeVar(conf, HiveConf.ConfVars.HIVE_TXN_TIMEOUT, TimeUnit.MILLISECONDS) * 2);
     Thread.sleep(HiveConf.getTimeVar(conf, HiveConf.ConfVars.HIVE_TXN_TIMEOUT, TimeUnit.MILLISECONDS));
-    runReaper();
-    try {
-      txnMgr.rollbackTxn();
-    }
-    catch (LockException ex) {
-      exception = ex;
-    }
-    Assert.assertNotNull("Expected exception2", exception);
-    Assert.assertEquals("Wrong Exception2", ErrorMsg.TXN_ABORTED, exception.getCanonicalErrorMsg());
+    runReaper();//this will abort the txn
+    TxnStore txnHandler = TxnUtils.getTxnStore(conf);
+    GetOpenTxnsInfoResponse txnsInfo = txnHandler.getOpenTxnsInfo();
+    assertEquals(2, txnsInfo.getTxn_high_water_mark());
+    assertEquals(2, txnsInfo.getOpen_txns().size());
+    Assert.assertEquals(TxnState.ABORTED, txnsInfo.getOpen_txns().get(1).getState());
+    txnMgr.rollbackTxn();//this is idempotent
   }
 
   @Test


Mime
View raw message