hive-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From w...@apache.org
Subject [1/2] hive git commit: HIVE-13249 : Hard upper bound on number of open transactions (Wei Zheng, reviewed by Eugene Koifman)
Date Fri, 20 May 2016 17:25:19 GMT
Repository: hive
Updated Branches:
  refs/heads/branch-1 5fe252b93 -> cb3636f3f


http://git-wip-us.apache.org/repos/asf/hive/blob/cb3636f3/ql/src/java/org/apache/hadoop/hive/ql/txn/AcidOpenTxnsCounterService.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/txn/AcidOpenTxnsCounterService.java b/ql/src/java/org/apache/hadoop/hive/ql/txn/AcidOpenTxnsCounterService.java
new file mode 100644
index 0000000..f5eb8a1
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/txn/AcidOpenTxnsCounterService.java
@@ -0,0 +1,69 @@
+/**
+ * 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.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.txn.TxnStore;
+import org.apache.hadoop.hive.metastore.txn.TxnUtils;
+import org.apache.hadoop.hive.ql.txn.compactor.HouseKeeperServiceBase;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+/**
+ * Background running thread, periodically updating number of open transactions.
+ * Runs inside Hive Metastore Service.
+ */
+public class AcidOpenTxnsCounterService extends HouseKeeperServiceBase {
+  private static final Logger LOG = LoggerFactory.getLogger(AcidOpenTxnsCounterService.class);
+  @Override
+  protected long getStartDelayMs() {
+    return 100;  // in miliseconds
+  }
+  @Override
+  protected long getIntervalMs() {
+    return hiveConf.getTimeVar(HiveConf.ConfVars.HIVE_COUNT_OPEN_TXNS_INTERVAL, TimeUnit.MILLISECONDS);
+  }
+  @Override
+  protected Runnable getScheduedAction(HiveConf hiveConf, AtomicInteger isAliveCounter) {
+    return new OpenTxnsCounter(hiveConf, isAliveCounter);
+  }
+  @Override
+  public String getServiceDescription() {
+    return "Count number of open transactions";
+  }
+  private static final class OpenTxnsCounter implements Runnable {
+    private final TxnStore txnHandler;
+    private final AtomicInteger isAliveCounter;
+    private OpenTxnsCounter(HiveConf hiveConf, AtomicInteger isAliveCounter) {
+      txnHandler = TxnUtils.getTxnStore(hiveConf);
+      this.isAliveCounter = isAliveCounter;
+    }
+    @Override
+    public void run() {
+      try {
+        long startTime = System.currentTimeMillis();
+        txnHandler.countOpenTxns();
+        int count = isAliveCounter.incrementAndGet();
+        LOG.info("OpenTxnsCounter ran for " + (System.currentTimeMillis() - startTime)/1000 + "seconds.  isAliveCounter=" + count);
+      }
+      catch(Throwable t) {
+        LOG.error("Serious error in {}", Thread.currentThread().getName(), ": {}" + t.getMessage(), t);
+      }
+    }
+  }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hive/blob/cb3636f3/ql/src/test/org/apache/hadoop/hive/metastore/txn/TestCompactionTxnHandler.java
----------------------------------------------------------------------
diff --git a/ql/src/test/org/apache/hadoop/hive/metastore/txn/TestCompactionTxnHandler.java b/ql/src/test/org/apache/hadoop/hive/metastore/txn/TestCompactionTxnHandler.java
new file mode 100644
index 0000000..23ad54e
--- /dev/null
+++ b/ql/src/test/org/apache/hadoop/hive/metastore/txn/TestCompactionTxnHandler.java
@@ -0,0 +1,447 @@
+/**
+ * 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.txn;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.api.*;
+import org.apache.log4j.Level;
+import org.apache.log4j.LogManager;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+import static junit.framework.Assert.*;
+
+/**
+ * Tests for TxnHandler.
+ */
+public class TestCompactionTxnHandler {
+
+  private HiveConf conf = new HiveConf();
+  private TxnStore txnHandler;
+
+  public TestCompactionTxnHandler() throws Exception {
+    TxnDbUtil.setConfValues(conf);
+    LogManager.getLogger(TxnHandler.class.getName()).setLevel(Level.DEBUG);
+    tearDown();
+  }
+
+  @Test
+  public void testFindNextToCompact() throws Exception {
+    CompactionRequest rqst = new CompactionRequest("foo", "bar", CompactionType.MINOR);
+    rqst.setPartitionname("ds=today");
+    txnHandler.compact(rqst);
+    long now = System.currentTimeMillis();
+    CompactionInfo ci = txnHandler.findNextToCompact("fred");
+    assertNotNull(ci);
+    assertEquals("foo", ci.dbname);
+    assertEquals("bar", ci.tableName);
+    assertEquals("ds=today", ci.partName);
+    assertEquals(CompactionType.MINOR, ci.type);
+    assertNull(ci.runAs);
+    assertNull(txnHandler.findNextToCompact("fred"));
+
+    txnHandler.setRunAs(ci.id, "bob");
+
+    ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
+    List<ShowCompactResponseElement> compacts = rsp.getCompacts();
+    assertEquals(1, compacts.size());
+    ShowCompactResponseElement c = compacts.get(0);
+    assertEquals("foo", c.getDbname());
+    assertEquals("bar", c.getTablename());
+    assertEquals("ds=today", c.getPartitionname());
+    assertEquals(CompactionType.MINOR, c.getType());
+    assertEquals("working", c.getState());
+    assertTrue(c.getStart() - 5000 < now && c.getStart() + 5000 > now);
+    assertEquals("fred", c.getWorkerid());
+    assertEquals("bob", c.getRunAs());
+  }
+
+  @Test
+  public void testFindNextToCompact2() throws Exception {
+    CompactionRequest rqst = new CompactionRequest("foo", "bar", CompactionType.MINOR);
+    rqst.setPartitionname("ds=today");
+    txnHandler.compact(rqst);
+
+    rqst = new CompactionRequest("foo", "bar", CompactionType.MINOR);
+    rqst.setPartitionname("ds=yesterday");
+    txnHandler.compact(rqst);
+
+    long now = System.currentTimeMillis();
+    boolean expectToday = false;
+    CompactionInfo ci = txnHandler.findNextToCompact("fred");
+    assertNotNull(ci);
+    assertEquals("foo", ci.dbname);
+    assertEquals("bar", ci.tableName);
+    if ("ds=today".equals(ci.partName)) expectToday = false;
+    else if ("ds=yesterday".equals(ci.partName)) expectToday = true;
+    else fail("partition name should have been today or yesterday but was " + ci.partName);
+    assertEquals(CompactionType.MINOR, ci.type);
+
+    ci = txnHandler.findNextToCompact("fred");
+    assertNotNull(ci);
+    assertEquals("foo", ci.dbname);
+    assertEquals("bar", ci.tableName);
+    if (expectToday) assertEquals("ds=today", ci.partName);
+    else assertEquals("ds=yesterday", ci.partName);
+    assertEquals(CompactionType.MINOR, ci.type);
+
+    assertNull(txnHandler.findNextToCompact("fred"));
+
+    ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
+    List<ShowCompactResponseElement> compacts = rsp.getCompacts();
+    assertEquals(2, compacts.size());
+    for (ShowCompactResponseElement e : compacts) {
+      assertEquals("working", e.getState());
+      assertTrue(e.getStart() - 5000 < now && e.getStart() + 5000 > now);
+      assertEquals("fred", e.getWorkerid());
+    }
+  }
+
+  @Test
+  public void testFindNextToCompactNothingToCompact() throws Exception {
+    assertNull(txnHandler.findNextToCompact("fred"));
+  }
+
+  @Test
+  public void testMarkCompacted() throws Exception {
+    CompactionRequest rqst = new CompactionRequest("foo", "bar", CompactionType.MINOR);
+    rqst.setPartitionname("ds=today");
+    txnHandler.compact(rqst);
+    CompactionInfo ci = txnHandler.findNextToCompact("fred");
+    assertNotNull(ci);
+
+    txnHandler.markCompacted(ci);
+    assertNull(txnHandler.findNextToCompact("fred"));
+
+
+
+    ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
+    List<ShowCompactResponseElement> compacts = rsp.getCompacts();
+    assertEquals(1, compacts.size());
+    ShowCompactResponseElement c = compacts.get(0);
+    assertEquals("foo", c.getDbname());
+    assertEquals("bar", c.getTablename());
+    assertEquals("ds=today", c.getPartitionname());
+    assertEquals(CompactionType.MINOR, c.getType());
+    assertEquals("ready for cleaning", c.getState());
+    assertNull(c.getWorkerid());
+  }
+
+  @Test
+  public void testFindNextToClean() throws Exception {
+    CompactionRequest rqst = new CompactionRequest("foo", "bar", CompactionType.MINOR);
+    rqst.setPartitionname("ds=today");
+    txnHandler.compact(rqst);
+    assertEquals(0, txnHandler.findReadyToClean().size());
+    CompactionInfo ci = txnHandler.findNextToCompact("fred");
+    assertNotNull(ci);
+
+    assertEquals(0, txnHandler.findReadyToClean().size());
+    txnHandler.markCompacted(ci);
+    assertNull(txnHandler.findNextToCompact("fred"));
+
+    List<CompactionInfo> toClean = txnHandler.findReadyToClean();
+    assertEquals(1, toClean.size());
+    assertNull(txnHandler.findNextToCompact("fred"));
+
+    ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
+    List<ShowCompactResponseElement> compacts = rsp.getCompacts();
+    assertEquals(1, compacts.size());
+    ShowCompactResponseElement c = compacts.get(0);
+    assertEquals("foo", c.getDbname());
+    assertEquals("bar", c.getTablename());
+    assertEquals("ds=today", c.getPartitionname());
+    assertEquals(CompactionType.MINOR, c.getType());
+    assertEquals("ready for cleaning", c.getState());
+    assertNull(c.getWorkerid());
+  }
+
+  @Test
+  public void testMarkCleaned() throws Exception {
+    CompactionRequest rqst = new CompactionRequest("foo", "bar", CompactionType.MINOR);
+    rqst.setPartitionname("ds=today");
+    txnHandler.compact(rqst);
+    assertEquals(0, txnHandler.findReadyToClean().size());
+    CompactionInfo ci = txnHandler.findNextToCompact("fred");
+    assertNotNull(ci);
+
+    assertEquals(0, txnHandler.findReadyToClean().size());
+    txnHandler.markCompacted(ci);
+    assertNull(txnHandler.findNextToCompact("fred"));
+
+    List<CompactionInfo> toClean = txnHandler.findReadyToClean();
+    assertEquals(1, toClean.size());
+    assertNull(txnHandler.findNextToCompact("fred"));
+    txnHandler.markCleaned(ci);
+    assertNull(txnHandler.findNextToCompact("fred"));
+    assertEquals(0, txnHandler.findReadyToClean().size());
+
+    ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
+    assertEquals(1, rsp.getCompactsSize());
+    assertTrue(TxnHandler.SUCCEEDED_RESPONSE.equals(rsp.getCompacts().get(0).getState()));
+  }
+
+  @Test
+  public void testRevokeFromLocalWorkers() throws Exception {
+    CompactionRequest rqst = new CompactionRequest("foo", "bar", CompactionType.MINOR);
+    txnHandler.compact(rqst);
+    rqst = new CompactionRequest("foo", "baz", CompactionType.MINOR);
+    txnHandler.compact(rqst);
+    rqst = new CompactionRequest("foo", "bazzoo", CompactionType.MINOR);
+    txnHandler.compact(rqst);
+    assertNotNull(txnHandler.findNextToCompact("fred-193892"));
+    assertNotNull(txnHandler.findNextToCompact("bob-193892"));
+    assertNotNull(txnHandler.findNextToCompact("fred-193893"));
+    txnHandler.revokeFromLocalWorkers("fred");
+
+    ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
+    List<ShowCompactResponseElement> compacts = rsp.getCompacts();
+    assertEquals(3, compacts.size());
+    boolean sawWorkingBob = false;
+    int initiatedCount = 0;
+    for (ShowCompactResponseElement c : compacts) {
+      if (c.getState().equals("working")) {
+        assertEquals("bob-193892", c.getWorkerid());
+        sawWorkingBob = true;
+      } else if (c.getState().equals("initiated")) {
+        initiatedCount++;
+      } else {
+        fail("Unexpected state");
+      }
+    }
+    assertTrue(sawWorkingBob);
+    assertEquals(2, initiatedCount);
+  }
+
+  @Test
+  public void testRevokeTimedOutWorkers() throws Exception {
+    CompactionRequest rqst = new CompactionRequest("foo", "bar", CompactionType.MINOR);
+    txnHandler.compact(rqst);
+    rqst = new CompactionRequest("foo", "baz", CompactionType.MINOR);
+    txnHandler.compact(rqst);
+
+    assertNotNull(txnHandler.findNextToCompact("fred-193892"));
+    Thread.sleep(200);
+    assertNotNull(txnHandler.findNextToCompact("fred-193892"));
+    txnHandler.revokeTimedoutWorkers(100);
+
+    ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
+    List<ShowCompactResponseElement> compacts = rsp.getCompacts();
+    assertEquals(2, compacts.size());
+    boolean sawWorking = false, sawInitiated = false;
+    for (ShowCompactResponseElement c : compacts) {
+      if (c.getState().equals("working"))  sawWorking = true;
+      else if (c.getState().equals("initiated")) sawInitiated = true;
+      else fail("Unexpected state");
+    }
+    assertTrue(sawWorking);
+    assertTrue(sawInitiated);
+  }
+
+  @Test
+  public void testFindPotentialCompactions() throws Exception {
+    // Test that committing unlocks
+    long txnid = openTxn();
+    LockComponent comp = new LockComponent(LockType.SHARED_WRITE, LockLevel.DB,
+        "mydb");
+    comp.setTablename("mytable");
+    comp.setOperationType(DataOperationType.UPDATE);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    comp = new LockComponent(LockType.SHARED_WRITE, LockLevel.DB,
+        "mydb");
+    comp.setTablename("yourtable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.UPDATE);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    req.setTxnid(txnid);
+    LockResponse res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+    txnHandler.commitTxn(new CommitTxnRequest(txnid));
+    assertEquals(0, txnHandler.numLocksInLockTable());
+
+    Set<CompactionInfo> potentials = txnHandler.findPotentialCompactions(100);
+    assertEquals(2, potentials.size());
+    boolean sawMyTable = false, sawYourTable = false;
+    for (CompactionInfo ci : potentials) {
+      sawMyTable |= (ci.dbname.equals("mydb") && ci.tableName.equals("mytable") &&
+          ci.partName ==  null);
+      sawYourTable |= (ci.dbname.equals("mydb") && ci.tableName.equals("yourtable") &&
+          ci.partName.equals("mypartition"));
+    }
+    assertTrue(sawMyTable);
+    assertTrue(sawYourTable);
+  }
+
+  // TODO test changes to mark cleaned to clean txns and txn_components
+
+  @Test
+  public void testMarkCleanedCleansTxnsAndTxnComponents()
+      throws Exception {
+    long txnid = openTxn();
+    LockComponent comp = new LockComponent(LockType.SHARED_WRITE, LockLevel.DB,
+        "mydb");
+    comp.setTablename("mytable");
+    comp.setOperationType(DataOperationType.INSERT);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    req.setTxnid(txnid);
+    LockResponse res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+    txnHandler.abortTxn(new AbortTxnRequest(txnid));
+
+    txnid = openTxn();
+    comp = new LockComponent(LockType.SHARED_WRITE, LockLevel.DB, "mydb");
+    comp.setTablename("yourtable");
+    comp.setOperationType(DataOperationType.DELETE);
+    components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    req.setTxnid(txnid);
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+    txnHandler.abortTxn(new AbortTxnRequest(txnid));
+
+    txnid = openTxn();
+    comp = new LockComponent(LockType.SHARED_WRITE, LockLevel.DB, "mydb");
+    comp.setTablename("foo");
+    comp.setPartitionname("bar");
+    comp.setOperationType(DataOperationType.UPDATE);
+    components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    req.setTxnid(txnid);
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+
+    comp = new LockComponent(LockType.SHARED_WRITE, LockLevel.DB, "mydb");
+    comp.setTablename("foo");
+    comp.setPartitionname("baz");
+    comp.setOperationType(DataOperationType.UPDATE);
+    components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    req.setTxnid(txnid);
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+    txnHandler.abortTxn(new AbortTxnRequest(txnid));
+
+    CompactionInfo ci = new CompactionInfo();
+
+    // Now clean them and check that they are removed from the count.
+    CompactionRequest rqst = new CompactionRequest("mydb", "mytable", CompactionType.MAJOR);
+    txnHandler.compact(rqst);
+    assertEquals(0, txnHandler.findReadyToClean().size());
+    ci = txnHandler.findNextToCompact("fred");
+    assertNotNull(ci);
+    txnHandler.markCompacted(ci);
+
+    List<CompactionInfo> toClean = txnHandler.findReadyToClean();
+    assertEquals(1, toClean.size());
+    txnHandler.markCleaned(ci);
+
+    // Check that we are cleaning up the empty aborted transactions
+    GetOpenTxnsResponse txnList = txnHandler.getOpenTxns();
+    assertEquals(3, txnList.getOpen_txnsSize());
+    txnHandler.cleanEmptyAbortedTxns();
+    txnList = txnHandler.getOpenTxns();
+    assertEquals(2, txnList.getOpen_txnsSize());
+
+    rqst = new CompactionRequest("mydb", "foo", CompactionType.MAJOR);
+    rqst.setPartitionname("bar");
+    txnHandler.compact(rqst);
+    assertEquals(0, txnHandler.findReadyToClean().size());
+    ci = txnHandler.findNextToCompact("fred");
+    assertNotNull(ci);
+    txnHandler.markCompacted(ci);
+
+    toClean = txnHandler.findReadyToClean();
+    assertEquals(1, toClean.size());
+    txnHandler.markCleaned(ci);
+
+    txnHandler.openTxns(new OpenTxnRequest(1, "me", "localhost"));
+    txnHandler.cleanEmptyAbortedTxns();
+    txnList = txnHandler.getOpenTxns();
+    assertEquals(3, txnList.getOpen_txnsSize());
+  }
+
+  @Test
+  public void addDynamicPartitions() throws Exception {
+    String dbName = "default";
+    String tableName = "adp_table";
+    OpenTxnsResponse openTxns = txnHandler.openTxns(new OpenTxnRequest(1, "me", "localhost"));
+    long txnId = openTxns.getTxn_ids().get(0);
+    // lock a table, as in dynamic partitions
+    LockComponent lc = new LockComponent(LockType.SHARED_WRITE, LockLevel.TABLE, dbName);
+    lc.setTablename(tableName);
+    DataOperationType dop = DataOperationType.UPDATE; 
+    lc.setOperationType(dop);
+    LockRequest lr = new LockRequest(Arrays.asList(lc), "me", "localhost");
+    lr.setTxnid(txnId);
+    LockResponse lock = txnHandler.lock(lr);
+    assertEquals(LockState.ACQUIRED, lock.getState());
+
+    AddDynamicPartitions adp = new AddDynamicPartitions(txnId, dbName, tableName,
+      Arrays.asList("ds=yesterday", "ds=today"));
+    adp.setOperationType(dop);
+    txnHandler.addDynamicPartitions(adp);
+    txnHandler.commitTxn(new CommitTxnRequest(txnId));
+
+    Set<CompactionInfo> potentials = txnHandler.findPotentialCompactions(1000);
+    assertEquals(2, potentials.size());
+    SortedSet<CompactionInfo> sorted = new TreeSet<CompactionInfo>(potentials);
+
+    int i = 0;
+    for (CompactionInfo ci : sorted) {
+      assertEquals(dbName, ci.dbname);
+      assertEquals(tableName, ci.tableName);
+      switch (i++) {
+        case 0: assertEquals("ds=today", ci.partName); break;
+        case 1: assertEquals("ds=yesterday", ci.partName); break;
+      default: throw new RuntimeException("What?");
+      }
+    }
+  }
+
+  @Before
+  public void setUp() throws Exception {
+    TxnDbUtil.prepDb();
+    txnHandler = TxnUtils.getTxnStore(conf);
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    TxnDbUtil.cleanDb();
+  }
+
+  private long openTxn() throws MetaException {
+    List<Long> txns = txnHandler.openTxns(new OpenTxnRequest(1, "me", "localhost")).getTxn_ids();
+    return txns.get(0);
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/hive/blob/cb3636f3/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
new file mode 100644
index 0000000..0d4fc59
--- /dev/null
+++ b/ql/src/test/org/apache/hadoop/hive/metastore/txn/TestTxnHandler.java
@@ -0,0 +1,1521 @@
+/**
+ * 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.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.api.*;
+import org.apache.log4j.Level;
+import org.apache.log4j.LogManager;
+import org.junit.*;
+import org.apache.hadoop.util.StringUtils;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static junit.framework.Assert.assertEquals;
+import static junit.framework.Assert.assertFalse;
+import static junit.framework.Assert.assertNull;
+import static junit.framework.Assert.assertTrue;
+import static junit.framework.Assert.fail;
+
+/**
+ * Tests for TxnHandler.
+ */
+public class TestTxnHandler {
+  static final private String CLASS_NAME = TxnHandler.class.getName();
+  static final private Log LOG = LogFactory.getLog(CLASS_NAME);
+
+  private HiveConf conf = new HiveConf();
+  private TxnStore txnHandler;
+
+  public TestTxnHandler() throws Exception {
+    TxnDbUtil.setConfValues(conf);
+    LogManager.getLogger(TxnHandler.class.getName()).setLevel(Level.DEBUG);
+    tearDown();
+  }
+
+  @Test
+  public void testValidTxnsEmpty() throws Exception {
+    GetOpenTxnsInfoResponse txnsInfo = txnHandler.getOpenTxnsInfo();
+    assertEquals(0L, txnsInfo.getTxn_high_water_mark());
+    assertTrue(txnsInfo.getOpen_txns().isEmpty());
+    GetOpenTxnsResponse txns = txnHandler.getOpenTxns();
+    assertEquals(0L, txns.getTxn_high_water_mark());
+    assertTrue(txns.getOpen_txns().isEmpty());
+  }
+
+  @Test
+  public void testOpenTxn() throws Exception {
+    long first = openTxn();
+    assertEquals(1L, first);
+    long second = openTxn();
+    assertEquals(2L, second);
+    GetOpenTxnsInfoResponse txnsInfo = txnHandler.getOpenTxnsInfo();
+    assertEquals(2L, txnsInfo.getTxn_high_water_mark());
+    assertEquals(2, txnsInfo.getOpen_txns().size());
+    assertEquals(1L, txnsInfo.getOpen_txns().get(0).getId());
+    assertEquals(TxnState.OPEN, txnsInfo.getOpen_txns().get(0).getState());
+    assertEquals(2L, txnsInfo.getOpen_txns().get(1).getId());
+    assertEquals(TxnState.OPEN, txnsInfo.getOpen_txns().get(1).getState());
+    assertEquals("me", txnsInfo.getOpen_txns().get(1).getUser());
+    assertEquals("localhost", txnsInfo.getOpen_txns().get(1).getHostname());
+
+    GetOpenTxnsResponse txns = txnHandler.getOpenTxns();
+    assertEquals(2L, txns.getTxn_high_water_mark());
+    assertEquals(2, txns.getOpen_txns().size());
+    boolean[] saw = new boolean[3];
+    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]);
+  }
+
+  @Test
+  public void testAbortTxn() throws Exception {
+    OpenTxnsResponse openedTxns = txnHandler.openTxns(new OpenTxnRequest(2, "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));
+    GetOpenTxnsInfoResponse txnsInfo = txnHandler.getOpenTxnsInfo();
+    assertEquals(2L, txnsInfo.getTxn_high_water_mark());
+    assertEquals(2, 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());
+
+    GetOpenTxnsResponse txns = txnHandler.getOpenTxns();
+    assertEquals(2L, txns.getTxn_high_water_mark());
+    assertEquals(2, txns.getOpen_txns().size());
+    boolean[] saw = new boolean[3];
+    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]);
+  }
+
+  @Test
+  public void testAbortInvalidTxn() throws Exception {
+    boolean caught = false;
+    try {
+      txnHandler.abortTxn(new AbortTxnRequest(195L));
+    } catch (NoSuchTxnException e) {
+      caught = true;
+    }
+    assertTrue(caught);
+  }
+
+  @Test
+  public void testValidTxnsNoneOpen() throws Exception {
+    txnHandler.openTxns(new OpenTxnRequest(2, "me", "localhost"));
+    txnHandler.commitTxn(new CommitTxnRequest(1));
+    txnHandler.commitTxn(new CommitTxnRequest(2));
+    GetOpenTxnsInfoResponse txnsInfo = txnHandler.getOpenTxnsInfo();
+    assertEquals(2L, txnsInfo.getTxn_high_water_mark());
+    assertEquals(0, txnsInfo.getOpen_txns().size());
+    GetOpenTxnsResponse txns = txnHandler.getOpenTxns();
+    assertEquals(2L, txns.getTxn_high_water_mark());
+    assertEquals(0, txns.getOpen_txns().size());
+  }
+
+  @Test
+  public void testValidTxnsSomeOpen() throws Exception {
+    txnHandler.openTxns(new OpenTxnRequest(3, "me", "localhost"));
+    txnHandler.abortTxn(new AbortTxnRequest(1));
+    txnHandler.commitTxn(new CommitTxnRequest(2));
+    GetOpenTxnsInfoResponse txnsInfo = txnHandler.getOpenTxnsInfo();
+    assertEquals(3L, txnsInfo.getTxn_high_water_mark());
+    assertEquals(2, txnsInfo.getOpen_txns().size());
+    assertEquals(1L, txnsInfo.getOpen_txns().get(0).getId());
+    assertEquals(TxnState.ABORTED, txnsInfo.getOpen_txns().get(0).getState());
+    assertEquals(3L, txnsInfo.getOpen_txns().get(1).getId());
+    assertEquals(TxnState.OPEN, txnsInfo.getOpen_txns().get(1).getState());
+
+    GetOpenTxnsResponse txns = txnHandler.getOpenTxns();
+    assertEquals(3L, txns.getTxn_high_water_mark());
+    assertEquals(2, 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;
+    }
+    assertTrue(saw[1]);
+    assertFalse(saw[2]);
+    assertTrue(saw[3]);
+  }
+
+  @Test
+  public void testLockDifferentDBs() throws Exception {
+    // Test that two different databases don't collide on their locks
+    LockComponent comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    LockResponse res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+
+    comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "yourdb");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+  }
+
+  @Test
+  public void testLockSameDB() throws Exception {
+    // Test that two different databases don't collide on their locks
+    LockComponent comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    LockResponse res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+
+    comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.WAITING);
+  }
+
+  @Test
+  public void testLockDbLocksTable() throws Exception {
+    // Test that locking a database prevents locking of tables in the database
+    LockComponent comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    LockResponse res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+
+    comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    comp.setTablename("mytable");
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.WAITING);
+  }
+
+  @Test
+  public void testLockDbDoesNotLockTableInDifferentDB() throws Exception {
+    // Test that locking a database prevents locking of tables in the database
+    LockComponent comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    LockResponse res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+
+    comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "yourdb");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    comp.setTablename("mytable");
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+  }
+
+  @Test
+  public void testLockDifferentTables() throws Exception {
+    // Test that two different tables don't collide on their locks
+    LockComponent comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    comp.setTablename("mytable");
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    LockResponse res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+
+    comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    comp.setTablename("yourtable");
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+  }
+
+  @Test
+  public void testLockSameTable() throws Exception {
+    // Test that two different tables don't collide on their locks
+    LockComponent comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    LockResponse res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+
+    comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.WAITING);
+  }
+
+  @Test
+  public void testLockTableLocksPartition() throws Exception {
+    // Test that locking a table prevents locking of partitions of the table
+    LockComponent comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    LockResponse res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+
+    comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.WAITING);
+  }
+
+  @Test
+  public void testLockDifferentTableDoesntLockPartition() throws Exception {
+    // Test that locking a table prevents locking of partitions of the table
+    LockComponent comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    LockResponse res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+
+    comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setTablename("yourtable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+  }
+
+  @Test
+  public void testLockDifferentPartitions() throws Exception {
+    // Test that two different partitions don't collide on their locks
+    LockComponent comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    LockResponse res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+
+    comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("yourpartition");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+  }
+
+  @Test
+  public void testLockSamePartition() throws Exception {
+    // Test that two different partitions don't collide on their locks
+    LockComponent comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    LockResponse res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+
+    comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.WAITING);
+  }
+
+  @Test
+  public void testLockSRSR() throws Exception {
+    // Test that two shared read locks can share a partition
+    LockComponent comp = new LockComponent(LockType.SHARED_READ, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.INSERT);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    LockResponse res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+
+    comp = new LockComponent(LockType.SHARED_READ, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.SELECT);
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+  }
+
+  @Test
+  public void testLockESRSR() throws Exception {
+    // Test that exclusive lock blocks shared reads
+    LockComponent comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    LockResponse res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+
+    comp = new LockComponent(LockType.SHARED_READ, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.INSERT);
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.WAITING);
+
+    comp = new LockComponent(LockType.SHARED_READ, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.SELECT);
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.WAITING);
+  }
+
+  @Test
+  public void testLockSRSW() throws Exception {
+    // Test that write can acquire after read
+    LockComponent comp = new LockComponent(LockType.SHARED_READ, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.INSERT);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    LockResponse res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+
+    comp = new LockComponent(LockType.SHARED_WRITE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.DELETE);
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    req.setTxnid(openTxn());
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+  }
+
+  @Test
+  public void testLockESRSW() throws Exception {
+    // Test that exclusive lock blocks read and write
+    LockComponent comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    LockResponse res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+
+    comp = new LockComponent(LockType.SHARED_READ, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.SELECT);
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.WAITING);
+
+    comp = new LockComponent(LockType.SHARED_WRITE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.UPDATE);
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    req.setTxnid(openTxn());
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.WAITING);
+  }
+
+  @Test
+  public void testLockSRE() throws Exception {
+    // Test that read blocks exclusive
+    LockComponent comp = new LockComponent(LockType.SHARED_READ, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.SELECT);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    LockResponse res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+
+    comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.WAITING);
+  }
+
+  @Test
+  public void testLockESRE() throws Exception {
+    // Test that exclusive blocks read and exclusive
+    LockComponent comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    LockResponse res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+
+    comp = new LockComponent(LockType.SHARED_READ, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.SELECT);
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.WAITING);
+
+    comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.WAITING);
+  }
+
+  @Test
+  public void testLockSWSR() throws Exception {
+    // Test that read can acquire after write
+    LockComponent comp = new LockComponent(LockType.SHARED_WRITE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.UPDATE);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    req.setTxnid(openTxn());
+    LockResponse res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+
+    comp = new LockComponent(LockType.SHARED_READ, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.SELECT);
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+  }
+
+  @Test
+  public void testLockSWSWSR() throws Exception {
+    // Test that write blocks write but read can still acquire
+    LockComponent comp = new LockComponent(LockType.SHARED_WRITE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.UPDATE);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    req.setTxnid(openTxn());
+    LockResponse res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+
+    comp = new LockComponent(LockType.SHARED_WRITE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.DELETE);
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    req.setTxnid(openTxn());
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.WAITING);
+
+    comp = new LockComponent(LockType.SHARED_READ, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.INSERT);
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+  }
+
+  @Test
+  public void testWrongLockForOperation() throws Exception {
+    LockComponent comp = new LockComponent(LockType.SHARED_WRITE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    req.setTxnid(openTxn());
+    Exception expectedError = null;
+    try {
+      LockResponse res = txnHandler.lock(req);
+    }
+    catch(Exception e) {
+      expectedError = e;
+    }
+    Assert.assertTrue(expectedError != null && expectedError.getMessage().contains("Unexpected DataOperationType"));
+  }
+  @Test
+  public void testLockSWSWSW() throws Exception {
+    // Test that write blocks two writes
+    LockComponent comp = new LockComponent(LockType.SHARED_WRITE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.DELETE);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    req.setTxnid(openTxn());
+    LockResponse res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+
+    comp = new LockComponent(LockType.SHARED_WRITE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.DELETE);
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    req.setTxnid(openTxn());
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.WAITING);
+
+    comp = new LockComponent(LockType.SHARED_WRITE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.DELETE);
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    req.setTxnid(openTxn());
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.WAITING);
+  }
+
+  @Test
+  public void testLockEESW() throws Exception {
+    // Test that exclusive blocks exclusive and write
+    LockComponent comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    LockResponse res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+
+    comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.WAITING);
+
+    comp = new LockComponent(LockType.SHARED_WRITE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.DELETE);
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    req.setTxnid(openTxn());
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.WAITING);
+  }
+
+  @Test
+  public void testLockEESR() throws Exception {
+    // Test that exclusive blocks exclusive and read
+    LockComponent comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    LockResponse res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+
+    comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.WAITING);
+
+    comp = new LockComponent(LockType.SHARED_READ, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.SELECT);
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.WAITING);
+  }
+
+  @Test
+  public void testCheckLockAcquireAfterWaiting() throws Exception {
+    LockComponent comp = new LockComponent(LockType.SHARED_WRITE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.DELETE);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    long txnId = openTxn();
+    req.setTxnid(txnId);
+    LockResponse res = txnHandler.lock(req);
+    long lockid1 = res.getLockid();
+    assertTrue(res.getState() == LockState.ACQUIRED);
+
+    comp = new LockComponent(LockType.SHARED_WRITE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.UPDATE);
+    components.clear();
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    req.setTxnid(openTxn());
+    res = txnHandler.lock(req);
+    long lockid2 = res.getLockid();
+    assertTrue(res.getState() == LockState.WAITING);
+
+    txnHandler.abortTxn(new AbortTxnRequest(txnId));
+    res = txnHandler.checkLock(new CheckLockRequest(lockid2));
+    assertTrue(res.getState() == LockState.ACQUIRED);
+  }
+
+  @Test
+  public void testCheckLockNoSuchLock() throws Exception {
+    try {
+      txnHandler.checkLock(new CheckLockRequest(23L));
+      fail("Allowed to check lock on non-existent lock");
+    } catch (NoSuchLockException e) {
+    }
+  }
+
+  @Test
+  public void testCheckLockTxnAborted() throws Exception {
+    // Test that when a transaction is aborted, the heartbeat fails
+    long txnid = openTxn();
+    LockComponent comp = new LockComponent(LockType.SHARED_WRITE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.DELETE);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    req.setTxnid(txnid);
+    LockResponse res = txnHandler.lock(req);
+    long lockid = res.getLockid();
+    txnHandler.abortTxn(new AbortTxnRequest(txnid));
+    try {
+      // This will throw NoSuchLockException (even though it's the
+      // transaction we've closed) because that will have deleted the lock.
+      txnHandler.checkLock(new CheckLockRequest(lockid));
+      fail("Allowed to check lock on aborted transaction.");
+    } catch (NoSuchLockException e) {
+    }
+  }
+
+  @Test
+  public void testMultipleLock() throws Exception {
+    // Test more than one lock can be handled in a lock request
+    LockComponent comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    List<LockComponent> components = new ArrayList<LockComponent>(2);
+    components.add(comp);
+
+    comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("anotherpartition");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    LockResponse res = txnHandler.lock(req);
+    long lockid = res.getLockid();
+    assertTrue(res.getState() == LockState.ACQUIRED);
+    res = txnHandler.checkLock(new CheckLockRequest(lockid));
+    assertTrue(res.getState() == LockState.ACQUIRED);
+    txnHandler.unlock(new UnlockRequest(lockid));
+    assertEquals(0, txnHandler.numLocksInLockTable());
+  }
+
+  @Test
+  public void testMultipleLockWait() throws Exception {
+    // Test that two shared read locks can share a partition
+    LockComponent comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    List<LockComponent> components = new ArrayList<LockComponent>(2);
+    components.add(comp);
+
+    comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("anotherpartition");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    LockResponse res = txnHandler.lock(req);
+    long lockid1 = res.getLockid();
+    assertTrue(res.getState() == LockState.ACQUIRED);
+
+
+    comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    res = txnHandler.lock(req);
+    long lockid2 = res.getLockid();
+    assertTrue(res.getState() == LockState.WAITING);
+
+    txnHandler.unlock(new UnlockRequest(lockid1));
+
+    res = txnHandler.checkLock(new CheckLockRequest(lockid2));
+    assertTrue(res.getState() == LockState.ACQUIRED);
+  }
+
+  @Test
+  public void testUnlockOnCommit() throws Exception {
+    // Test that committing unlocks
+    long txnid = openTxn();
+    LockComponent comp = new LockComponent(LockType.SHARED_WRITE, LockLevel.DB,  "mydb");
+    comp.setTablename("mytable");
+    comp.setOperationType(DataOperationType.DELETE);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    req.setTxnid(txnid);
+    LockResponse res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+    txnHandler.commitTxn(new CommitTxnRequest(txnid));
+    assertEquals(0, txnHandler.numLocksInLockTable());
+  }
+
+  @Test
+  public void testUnlockOnAbort() throws Exception {
+    // Test that committing unlocks
+    long txnid = openTxn();
+    LockComponent comp = new LockComponent(LockType.SHARED_WRITE, LockLevel.DB, "mydb");
+    comp.setOperationType(DataOperationType.UPDATE);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    req.setTxnid(txnid);
+    LockResponse res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+    txnHandler.abortTxn(new AbortTxnRequest(txnid));
+    assertEquals(0, txnHandler.numLocksInLockTable());
+  }
+
+  @Test
+  public void testUnlockWithTxn() throws Exception {
+    LOG.debug("Starting testUnlockWithTxn");
+    // Test that attempting to unlock locks associated with a transaction
+    // generates an error
+    long txnid = openTxn();
+    LockComponent comp = new LockComponent(LockType.SHARED_WRITE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.DELETE);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    req.setTxnid(txnid);
+    LockResponse res = txnHandler.lock(req);
+    long lockid = res.getLockid();
+    try {
+      txnHandler.unlock(new UnlockRequest(lockid));
+      fail("Allowed to unlock lock associated with transaction.");
+    } catch (TxnOpenException e) {
+    }
+  }
+
+  @Test
+  public void testHeartbeatTxnAborted() throws Exception {
+    // Test that when a transaction is aborted, the heartbeat fails
+    openTxn();
+    txnHandler.abortTxn(new AbortTxnRequest(1));
+    HeartbeatRequest h = new HeartbeatRequest();
+    h.setTxnid(1);
+    try {
+      txnHandler.heartbeat(h);
+      fail("Told there was a txn, when it should have been aborted.");
+    } catch (TxnAbortedException e) {
+    }
+ }
+
+  @Test
+  public void testHeartbeatNoTxn() throws Exception {
+    // Test that when a transaction is aborted, the heartbeat fails
+    HeartbeatRequest h = new HeartbeatRequest();
+    h.setTxnid(939393L);
+    try {
+      txnHandler.heartbeat(h);
+      fail("Told there was a txn, when there wasn't.");
+    } catch (NoSuchTxnException e) {
+    }
+  }
+
+  @Test
+  public void testHeartbeatLock() throws Exception {
+    conf.setTimeVar(HiveConf.ConfVars.HIVE_TXN_TIMEOUT, 1, TimeUnit.SECONDS);
+    HeartbeatRequest h = new HeartbeatRequest();
+    LockComponent comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setTablename("mytable");
+    comp.setPartitionname("mypartition");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    LockResponse res = txnHandler.lock(req);
+    assertTrue(res.getState() == LockState.ACQUIRED);
+    h.setLockid(res.getLockid());
+    for (int i = 0; i < 30; i++) {
+      try {
+        txnHandler.heartbeat(h);
+      } catch (NoSuchLockException e) {
+        fail("Told there was no lock, when the heartbeat should have kept it.");
+      }
+    }
+  }
+
+  @Test
+  public void heartbeatTxnRange() throws Exception {
+    long txnid = openTxn();
+    assertEquals(1, txnid);
+    txnid = openTxn();
+    txnid = openTxn();
+    HeartbeatTxnRangeResponse rsp =
+        txnHandler.heartbeatTxnRange(new HeartbeatTxnRangeRequest(1, 3));
+    assertEquals(0, rsp.getAborted().size());
+    assertEquals(0, rsp.getNosuch().size());
+  }
+
+  @Test
+  public void heartbeatTxnRangeOneCommitted() throws Exception {
+    long txnid = openTxn();
+    assertEquals(1, txnid);
+    txnHandler.commitTxn(new CommitTxnRequest(1));
+    txnid = openTxn();
+    txnid = openTxn();
+    HeartbeatTxnRangeResponse rsp =
+      txnHandler.heartbeatTxnRange(new HeartbeatTxnRangeRequest(1, 3));
+    assertEquals(1, rsp.getNosuchSize());
+    Long txn = rsp.getNosuch().iterator().next();
+    assertEquals(1L, (long)txn);
+    assertEquals(0, rsp.getAborted().size());
+  }
+
+  @Test
+  public void heartbeatTxnRangeOneAborted() throws Exception {
+    long txnid = openTxn();
+    assertEquals(1, txnid);
+    txnid = openTxn();
+    txnid = openTxn();
+    txnHandler.abortTxn(new AbortTxnRequest(3));
+    HeartbeatTxnRangeResponse rsp =
+      txnHandler.heartbeatTxnRange(new HeartbeatTxnRangeRequest(1, 3));
+    assertEquals(1, rsp.getAbortedSize());
+    Long txn = rsp.getAborted().iterator().next();
+    assertEquals(3L, (long)txn);
+    assertEquals(0, rsp.getNosuch().size());
+  }
+
+  @Test
+  public void testLockTimeout() throws Exception {
+    long timeout = txnHandler.setTimeout(1);
+    try {
+      LockComponent comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+      comp.setTablename("mytable");
+      comp.setPartitionname("mypartition");
+      comp.setOperationType(DataOperationType.NO_TXN);
+      List<LockComponent> components = new ArrayList<LockComponent>(1);
+      components.add(comp);
+      LockRequest req = new LockRequest(components, "me", "localhost");
+      LockResponse res = txnHandler.lock(req);
+      assertTrue(res.getState() == LockState.ACQUIRED);
+      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) {
+    } finally {
+      txnHandler.setTimeout(timeout);
+    }
+  }
+
+  @Test
+  public void testRecoverManyTimeouts() throws Exception {
+    long timeout = txnHandler.setTimeout(1);
+    try {
+      txnHandler.openTxns(new OpenTxnRequest(503, "me", "localhost"));
+      Thread.sleep(10);
+      txnHandler.performTimeOuts();
+      GetOpenTxnsInfoResponse rsp = txnHandler.getOpenTxnsInfo();
+      int numAborted = 0;
+      for (TxnInfo txnInfo : rsp.getOpen_txns()) {
+        assertEquals(TxnState.ABORTED, txnInfo.getState());
+        numAborted++;
+      }
+      assertEquals(503, numAborted);
+    } finally {
+      txnHandler.setTimeout(timeout);
+    }
+
+
+  }
+
+  @Test
+  public void testHeartbeatNoLock() throws Exception {
+    HeartbeatRequest h = new HeartbeatRequest();
+    h.setLockid(29389839L);
+    try {
+      txnHandler.heartbeat(h);
+      fail("Told there was a lock, when there wasn't.");
+    } catch (NoSuchLockException e) {
+    }
+  }
+
+  @Test
+  public void testCompactMajorWithPartition() throws Exception {
+    CompactionRequest rqst = new CompactionRequest("foo", "bar", CompactionType.MAJOR);
+    rqst.setPartitionname("ds=today");
+    txnHandler.compact(rqst);
+    ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
+    List<ShowCompactResponseElement> compacts = rsp.getCompacts();
+    assertEquals(1, compacts.size());
+    ShowCompactResponseElement c = compacts.get(0);
+    assertEquals("foo", c.getDbname());
+    assertEquals("bar", c.getTablename());
+    assertEquals("ds=today", c.getPartitionname());
+    assertEquals(CompactionType.MAJOR, c.getType());
+    assertEquals("initiated", c.getState());
+    assertEquals(0L, c.getStart());
+  }
+
+  @Test
+  public void testCompactMinorNoPartition() throws Exception {
+    CompactionRequest rqst = new CompactionRequest("foo", "bar", CompactionType.MINOR);
+    rqst.setRunas("fred");
+    txnHandler.compact(rqst);
+
+    ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
+    List<ShowCompactResponseElement> compacts = rsp.getCompacts();
+    assertEquals(1, compacts.size());
+    ShowCompactResponseElement c = compacts.get(0);
+    assertEquals("foo", c.getDbname());
+    assertEquals("bar", c.getTablename());
+    assertNull(c.getPartitionname());
+    assertEquals(CompactionType.MINOR, c.getType());
+    assertEquals("initiated", c.getState());
+    assertEquals(0L, c.getStart());
+    assertEquals("fred", c.getRunAs());
+  }
+
+  @Test
+  public void showLocks() throws Exception {
+    long begining = System.currentTimeMillis();
+    LockComponent comp = new LockComponent(LockType.EXCLUSIVE, LockLevel.DB, "mydb");
+    comp.setOperationType(DataOperationType.NO_TXN);
+    List<LockComponent> components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    LockRequest req = new LockRequest(components, "me", "localhost");
+    LockResponse res = txnHandler.lock(req);
+
+    // Open txn
+    long txnid = openTxn();
+    comp = new LockComponent(LockType.SHARED_READ, LockLevel.TABLE, "mydb");
+    comp.setTablename("mytable");
+    comp.setOperationType(DataOperationType.SELECT);
+    components = new ArrayList<LockComponent>(1);
+    components.add(comp);
+    req = new LockRequest(components, "me", "localhost");
+    req.setTxnid(txnid);
+    res = txnHandler.lock(req);
+
+    // Locks not associated with a txn
+    components = new ArrayList<LockComponent>(1);
+    comp = new LockComponent(LockType.SHARED_READ, LockLevel.PARTITION, "yourdb");
+    comp.setTablename("yourtable");
+    comp.setPartitionname("yourpartition");
+    comp.setOperationType(DataOperationType.INSERT);
+    components.add(comp);
+    req = new LockRequest(components, "you", "remotehost");
+    res = txnHandler.lock(req);
+
+    ShowLocksResponse rsp = txnHandler.showLocks(new ShowLocksRequest());
+    List<ShowLocksResponseElement> locks = rsp.getLocks();
+    assertEquals(3, locks.size());
+    boolean[] saw = new boolean[locks.size()];
+    for (int i = 0; i < saw.length; i++) saw[i] = false;
+    for (ShowLocksResponseElement lock : locks) {
+      if (lock.getLockid() == 1) {
+        assertEquals(0, lock.getTxnid());
+        assertEquals("mydb", lock.getDbname());
+        assertNull(lock.getTablename());
+        assertNull(lock.getPartname());
+        assertEquals(LockState.ACQUIRED, lock.getState());
+        assertEquals(LockType.EXCLUSIVE, lock.getType());
+        assertTrue(lock.toString(), 0 != lock.getLastheartbeat());
+        assertTrue("Expected acquired at " + lock.getAcquiredat() + " to be between " + begining
+            + " and " + System.currentTimeMillis(),
+            begining <= lock.getAcquiredat() && System.currentTimeMillis() >= lock.getAcquiredat());
+        assertEquals("me", lock.getUser());
+        assertEquals("localhost", lock.getHostname());
+        saw[0] = true;
+      } else if (lock.getLockid() == 2) {
+        assertEquals(1, lock.getTxnid());
+        assertEquals("mydb", lock.getDbname());
+        assertEquals("mytable", lock.getTablename());
+        assertNull(lock.getPartname());
+        assertEquals(LockState.WAITING, lock.getState());
+        assertEquals(LockType.SHARED_READ, lock.getType());
+        assertTrue(lock.toString(), 0 == lock.getLastheartbeat() &&
+          lock.getTxnid() != 0);
+        assertEquals(0, lock.getAcquiredat());
+        assertEquals("me", lock.getUser());
+        assertEquals("localhost", lock.getHostname());
+        saw[1] = true;
+      } else if (lock.getLockid() == 3) {
+        assertEquals(0, lock.getTxnid());
+        assertEquals("yourdb", lock.getDbname());
+        assertEquals("yourtable", lock.getTablename());
+        assertEquals("yourpartition", lock.getPartname());
+        assertEquals(LockState.ACQUIRED, lock.getState());
+        assertEquals(LockType.SHARED_READ, lock.getType());
+        assertTrue(lock.toString(), begining <= lock.getLastheartbeat() &&
+            System.currentTimeMillis() >= lock.getLastheartbeat());
+        assertTrue(begining <= lock.getAcquiredat() &&
+            System.currentTimeMillis() >= lock.getAcquiredat());
+        assertEquals("you", lock.getUser());
+        assertEquals("remotehost", lock.getHostname());
+        saw[2] = true;
+      } else {
+        fail("Unknown lock id");
+      }
+    }
+    for (int i = 0; i < saw.length; i++) assertTrue("Didn't see lock id " + i, saw[i]);
+  }
+
+  @Test
+  @Ignore("Wedges Derby")
+  public void deadlockDetected() throws Exception {
+    LOG.debug("Starting deadlock test");
+    if (txnHandler instanceof TxnHandler) {
+      final TxnHandler tHndlr = (TxnHandler)txnHandler;
+      Connection conn = tHndlr.getDbConn(Connection.TRANSACTION_SERIALIZABLE);
+      Statement stmt = conn.createStatement();
+      long now = tHndlr.getDbTime(conn);
+      stmt.executeUpdate("insert into TXNS (txn_id, txn_state, txn_started, txn_last_heartbeat, " +
+          "txn_user, txn_host) values (1, 'o', " + now + ", " + now + ", 'shagy', " +
+          "'scooby.com')");
+      stmt.executeUpdate("insert into HIVE_LOCKS (hl_lock_ext_id, hl_lock_int_id, hl_txnid, " +
+          "hl_db, hl_table, hl_partition, hl_lock_state, hl_lock_type, hl_last_heartbeat, " +
+          "hl_user, hl_host) values (1, 1, 1, 'mydb', 'mytable', 'mypartition', '" +
+          tHndlr.LOCK_WAITING + "', '" + tHndlr.LOCK_EXCLUSIVE + "', " + now + ", 'fred', " +
+          "'scooby.com')");
+      conn.commit();
+      tHndlr.closeDbConn(conn);
+
+      final AtomicBoolean sawDeadlock = new AtomicBoolean();
+
+      final Connection conn1 = tHndlr.getDbConn(Connection.TRANSACTION_SERIALIZABLE);
+      final Connection conn2 = tHndlr.getDbConn(Connection.TRANSACTION_SERIALIZABLE);
+      try {
+
+        for (int i = 0; i < 5; i++) {
+          Thread t1 = new Thread() {
+            @Override
+            public void run() {
+              try {
+                try {
+                  updateTxns(conn1);
+                  updateLocks(conn1);
+                  Thread.sleep(1000);
+                  conn1.commit();
+                  LOG.debug("no exception, no deadlock");
+                } catch (SQLException e) {
+                  try {
+                    tHndlr.checkRetryable(conn1, e, "thread t1");
+                    LOG.debug("Got an exception, but not a deadlock, SQLState is " +
+                        e.getSQLState() + " class of exception is " + e.getClass().getName() +
+                        " msg is <" + e.getMessage() + ">");
+                  } catch (TxnHandler.RetryException de) {
+                    LOG.debug("Forced a deadlock, SQLState is " + e.getSQLState() + " class of " +
+                        "exception is " + e.getClass().getName() + " msg is <" + e
+                        .getMessage() + ">");
+                    sawDeadlock.set(true);
+                  }
+                }
+                conn1.rollback();
+              } catch (Exception e) {
+                throw new RuntimeException(e);
+              }
+            }
+          };
+
+          Thread t2 = new Thread() {
+            @Override
+            public void run() {
+              try {
+                try {
+                  updateLocks(conn2);
+                  updateTxns(conn2);
+                  Thread.sleep(1000);
+                  conn2.commit();
+                  LOG.debug("no exception, no deadlock");
+                } catch (SQLException e) {
+                  try {
+                    tHndlr.checkRetryable(conn2, e, "thread t2");
+                    LOG.debug("Got an exception, but not a deadlock, SQLState is " +
+                        e.getSQLState() + " class of exception is " + e.getClass().getName() +
+                        " msg is <" + e.getMessage() + ">");
+                  } catch (TxnHandler.RetryException de) {
+                    LOG.debug("Forced a deadlock, SQLState is " + e.getSQLState() + " class of " +
+                        "exception is " + e.getClass().getName() + " msg is <" + e
+                        .getMessage() + ">");
+                    sawDeadlock.set(true);
+                  }
+                }
+                conn2.rollback();
+              } catch (Exception e) {
+                throw new RuntimeException(e);
+              }
+            }
+          };
+
+          t1.start();
+          t2.start();
+          t1.join();
+          t2.join();
+          if (sawDeadlock.get()) break;
+        }
+        assertTrue(sawDeadlock.get());
+      } finally {
+        conn1.rollback();
+        tHndlr.closeDbConn(conn1);
+        conn2.rollback();
+        tHndlr.closeDbConn(conn2);
+      }
+    }
+  }
+
+  /**
+   * This cannnot be run against Derby (thus in UT) but it can run againt MySQL.
+   * 1. add to metastore/pom.xml
+   *     <dependency>
+   *      <groupId>mysql</groupId>
+   *      <artifactId>mysql-connector-java</artifactId>
+   *      <version>5.1.30</version>
+   *     </dependency>
+   * 2. Hack in the c'tor of this class
+   *     conf.setVar(HiveConf.ConfVars.METASTORECONNECTURLKEY, "jdbc:mysql://localhost/metastore");
+   *      conf.setVar(HiveConf.ConfVars.METASTORE_CONNECTION_USER_NAME, "hive");
+   *      conf.setVar(HiveConf.ConfVars.METASTOREPWD, "hive");
+   *      conf.setVar(HiveConf.ConfVars.METASTORE_CONNECTION_DRIVER, "com.mysql.jdbc.Driver");
+   * 3. Remove TxnDbUtil.prepDb(); in TxnHandler.checkQFileTestHack()
+   *      
+   */
+  @Ignore("multiple threads wedge Derby")
+  @Test
+  public void testMutexAPI() throws Exception {
+    final TxnStore.MutexAPI api =  txnHandler.getMutexAPI();
+    final AtomicInteger stepTracker = new AtomicInteger(0);
+    /**
+     * counter = 0;
+     * Thread1 counter=1, lock, wait 3s, check counter(should be 2), counter=3, unlock 
+     * Thread2 counter=2, lock (and block), inc counter, should be 4
+     */
+    Thread t1 = new Thread("MutexTest1") {
+      public void run() {
+        try {
+          stepTracker.incrementAndGet();//now 1
+          TxnStore.MutexAPI.LockHandle handle = api.acquireLock(TxnHandler.MUTEX_KEY.HouseKeeper.name());
+          Thread.sleep(4000);
+          //stepTracker should now be 2 which indicates t2 has started
+          Assert.assertEquals("Thread2 should have started by now but not done work", 2, stepTracker.get());
+          stepTracker.incrementAndGet();//now 3
+          handle.releaseLocks();
+        }
+        catch(Exception ex) {
+          throw new RuntimeException(ex.getMessage(), ex);
+        }
+      }
+    };
+    t1.setDaemon(true);
+    ErrorHandle ueh1 = new ErrorHandle();
+    t1.setUncaughtExceptionHandler(ueh1);
+    Thread t2 = new Thread("MutexTest2") {
+      public void run() {
+        try {
+          stepTracker.incrementAndGet();//now 2
+          //this should block until t1 unlocks
+          TxnStore.MutexAPI.LockHandle handle = api.acquireLock(TxnHandler.MUTEX_KEY.HouseKeeper.name());
+          stepTracker.incrementAndGet();//now 4
+          Assert.assertEquals(4, stepTracker.get());
+          handle.releaseLocks();
+          stepTracker.incrementAndGet();//now 5
+        }
+        catch(Exception ex) {
+          throw new RuntimeException(ex.getMessage(), ex);
+        }
+      }
+    };
+    t2.setDaemon(true);
+    ErrorHandle ueh2 = new ErrorHandle();
+    t2.setUncaughtExceptionHandler(ueh2);
+    t1.start();
+    try {
+      Thread.sleep(1000);
+    }
+    catch(InterruptedException ex) {
+      LOG.info("Sleep was interrupted");
+    }
+    t2.start();
+    t1.join(6000);//so that test doesn't block
+    t2.join(6000);
+
+    if(ueh1.error != null) {
+      Assert.assertTrue("Unexpected error from t1: " + StringUtils.stringifyException(ueh1.error), false);
+    }
+    if (ueh2.error != null) {
+      Assert.assertTrue("Unexpected error from t2: " + StringUtils.stringifyException(ueh2.error), false);
+    }
+    Assert.assertEquals("5 means both threads have completed", 5, stepTracker.get());
+  }
+  private final static class ErrorHandle implements Thread.UncaughtExceptionHandler {
+    Throwable error = null;
+    @Override
+    public void uncaughtException(Thread t, Throwable e) {
+      LOG.error("Uncaught exception from " + t.getName() + ": " + e.getMessage());
+      error = e;
+    }
+  }
+
+  @Test
+  public void testRetryableRegex() throws Exception {
+    SQLException sqlException = new SQLException("ORA-08177: can't serialize access for this transaction", "72000");
+    // Note that we have 3 regex'es below
+    conf.setVar(HiveConf.ConfVars.HIVE_TXN_RETRYABLE_SQLEX_REGEX, "^Deadlock detected, roll back,.*08177.*,.*08178.*");
+    boolean result = TxnHandler.isRetryable(conf, sqlException);
+    Assert.assertTrue("regex should be retryable", result);
+
+    sqlException = new SQLException("This error message, has comma in it");
+    conf.setVar(HiveConf.ConfVars.HIVE_TXN_RETRYABLE_SQLEX_REGEX, ".*comma.*");
+    result = TxnHandler.isRetryable(conf, sqlException);
+    Assert.assertTrue("regex should be retryable", result);
+  }
+
+  private void updateTxns(Connection conn) throws SQLException {
+    Statement stmt = conn.createStatement();
+    stmt.executeUpdate("update TXNS set txn_last_heartbeat = txn_last_heartbeat + 1");
+  }
+
+  private void updateLocks(Connection conn) throws SQLException {
+    Statement stmt = conn.createStatement();
+    stmt.executeUpdate("update HIVE_LOCKS set hl_last_heartbeat = hl_last_heartbeat + 1");
+  }
+
+  @Test
+  public void testBuildQueryWithINClause() throws Exception {
+    List<String> queries = new ArrayList<String>();
+
+    StringBuilder prefix = new StringBuilder();
+    StringBuilder suffix = new StringBuilder();
+
+    // Note, this is a "real" query that depends on one of the metastore tables
+    prefix.append("select count(*) from TXNS where ");
+    suffix.append(" and TXN_STATE = 'o'");
+
+    // Case 1 - Max in list members: 10; Max query string length: 1KB
+    //          The first query happens to have 2 full batches.
+    conf.setIntVar(HiveConf.ConfVars.METASTORE_DIRECT_SQL_MAX_QUERY_LENGTH, 1);
+    conf.setIntVar(HiveConf.ConfVars.METASTORE_DIRECT_SQL_MAX_ELEMENTS_IN_CLAUSE, 10);
+    List<Long> inList = new ArrayList<Long>();
+    for (long i = 1; i <= 200; i++) {
+      inList.add(i);
+    }
+    TxnUtils.buildQueryWithINClause(conf, queries, prefix, suffix, inList, "TXN_ID", true, false);
+    Assert.assertEquals(1, queries.size());
+    runAgainstDerby(queries);
+
+    // Case 2 - Max in list members: 10; Max query string length: 1KB
+    //          The first query has 2 full batches, and the second query only has 1 batch which only contains 1 member
+    queries.clear();
+    inList.add((long)201);
+    TxnUtils.buildQueryWithINClause(conf, queries, prefix, suffix, inList, "TXN_ID", true, false);
+    Assert.assertEquals(2, queries.size());
+    runAgainstDerby(queries);
+
+    // Case 3 - Max in list members: 1000; Max query string length: 5KB
+    conf.setIntVar(HiveConf.ConfVars.METASTORE_DIRECT_SQL_MAX_QUERY_LENGTH, 10);
+    conf.setIntVar(HiveConf.ConfVars.METASTORE_DIRECT_SQL_MAX_ELEMENTS_IN_CLAUSE, 1000);
+    queries.clear();
+    for (long i = 202; i <= 4321; i++) {
+      inList.add(i);
+    }
+    TxnUtils.buildQueryWithINClause(conf, queries, prefix, suffix, inList, "TXN_ID", true, false);
+    Assert.assertEquals(3, queries.size());
+    runAgainstDerby(queries);
+
+    // Case 4 - NOT IN list
+    queries.clear();
+    TxnUtils.buildQueryWithINClause(conf, queries, prefix, suffix, inList, "TXN_ID", true, true);
+    Assert.assertEquals(3, queries.size());
+    runAgainstDerby(queries);
+
+    // Case 5 - No parenthesis
+    queries.clear();
+    suffix.setLength(0);
+    suffix.append("");
+    TxnUtils.buildQueryWithINClause(conf, queries, prefix, suffix, inList, "TXN_ID", false, false);
+    Assert.assertEquals(3, queries.size());
+    runAgainstDerby(queries);
+  }
+
+  /** Verify queries can run against Derby DB.
+   *  As long as Derby doesn't complain, we assume the query is syntactically/semantically correct.
+   */
+  private void runAgainstDerby(List<String> queries) throws Exception {
+    Connection conn = null;
+    Statement stmt = null;
+    ResultSet rs = null;
+
+    try {
+      conn = TxnDbUtil.getConnection();
+      stmt = conn.createStatement();
+      for (String query : queries) {
+        rs = stmt.executeQuery(query);
+        Assert.assertTrue("The query is not valid", rs.next());
+      }
+    } finally {
+      TxnDbUtil.closeResources(conn, stmt, rs);
+    }
+  }
+
+  @Before
+  public void setUp() throws Exception {
+    TxnDbUtil.prepDb();
+    txnHandler = TxnUtils.getTxnStore(conf);
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    TxnDbUtil.cleanDb();
+  }
+
+  private long openTxn() throws MetaException {
+    List<Long> txns = txnHandler.openTxns(new OpenTxnRequest(1, "me", "localhost")).getTxn_ids();
+    return txns.get(0);
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/hive/blob/cb3636f3/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 8a8101b..6a944bc 100644
--- a/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java
+++ b/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java
@@ -26,9 +26,12 @@ import org.apache.hadoop.hive.common.FileUtils;
 import org.apache.hadoop.hive.common.ValidTxnList;
 import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.metastore.HouseKeeperService;
+import org.apache.hadoop.hive.metastore.api.CommitTxnRequest;
 import org.apache.hadoop.hive.metastore.api.CompactionRequest;
 import org.apache.hadoop.hive.metastore.api.CompactionType;
 import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.api.OpenTxnRequest;
+import org.apache.hadoop.hive.metastore.api.OpenTxnsResponse;
 import org.apache.hadoop.hive.metastore.api.ShowCompactRequest;
 import org.apache.hadoop.hive.metastore.api.ShowCompactResponse;
 import org.apache.hadoop.hive.metastore.api.ShowCompactResponseElement;
@@ -39,6 +42,7 @@ import org.apache.hadoop.hive.metastore.txn.TxnUtils;
 import org.apache.hadoop.hive.ql.processors.CommandProcessorResponse;
 import org.apache.hadoop.hive.ql.session.SessionState;
 import org.apache.hadoop.hive.ql.txn.AcidCompactionHistoryService;
+import org.apache.hadoop.hive.ql.txn.AcidOpenTxnsCounterService;
 import org.apache.hadoop.hive.ql.txn.compactor.Cleaner;
 import org.apache.hadoop.hive.ql.txn.compactor.Initiator;
 import org.apache.hadoop.hive.ql.txn.compactor.Worker;
@@ -703,7 +707,7 @@ public class TestTxnCommands2 {
     while(houseKeeperService.getIsAliveCounter() <= lastCount) {
       if(iterCount++ >= maxIter) {
         //prevent test hangs
-        throw new IllegalStateException("HouseKeeper didn't run after " + iterCount + " waits");
+        throw new IllegalStateException("HouseKeeper didn't run after " + (iterCount - 1) + " waits");
       }
       try {
         Thread.sleep(100);//make sure it has run at least once
@@ -793,6 +797,41 @@ public class TestTxnCommands2 {
     Assert.assertTrue(exception.getMessage().contains("HIVETESTMODEFAILHEARTBEATER=true"));
   }
 
+  @Test
+  public void testOpenTxnsCounter() throws Exception {
+    hiveConf.setIntVar(HiveConf.ConfVars.HIVE_MAX_OPEN_TXNS, 3);
+    hiveConf.setTimeVar(HiveConf.ConfVars.HIVE_COUNT_OPEN_TXNS_INTERVAL, 10, TimeUnit.MILLISECONDS);
+    TxnStore txnHandler = TxnUtils.getTxnStore(hiveConf);
+    OpenTxnsResponse openTxnsResponse = txnHandler.openTxns(new OpenTxnRequest(3, "me", "localhost"));
+
+    AcidOpenTxnsCounterService openTxnsCounterService = new AcidOpenTxnsCounterService();
+    runHouseKeeperService(openTxnsCounterService, hiveConf);  // will update current number of open txns to 3
+
+    MetaException exception = null;
+    // This should fail once it finds out the threshold has been reached
+    try {
+      txnHandler.openTxns(new OpenTxnRequest(1, "you", "localhost"));
+    } catch (MetaException e) {
+      exception = e;
+    }
+    Assert.assertNotNull("Opening new transaction shouldn't be allowed", exception);
+    Assert.assertTrue(exception.getMessage().equals("Maximum allowed number of open transactions has been reached. See hive.max.open.txns."));
+
+    // After committing the initial txns, and updating current number of open txns back to 0,
+    // new transactions should be allowed to open
+    for (long txnid : openTxnsResponse.getTxn_ids()) {
+      txnHandler.commitTxn(new CommitTxnRequest(txnid));
+    }
+    runHouseKeeperService(openTxnsCounterService, hiveConf);  // will update current number of open txns back to 0
+    exception = null;
+    try {
+      txnHandler.openTxns(new OpenTxnRequest(1, "him", "localhost"));
+    } catch (MetaException e) {
+      exception = e;
+    }
+    Assert.assertNull(exception);
+  }
+
   /**
    * takes raw data and turns it into a string as if from Driver.getResults()
    * sorts rows in dictionary order


Mime
View raw message