private void testDropPartition(boolean blocking) throws Exception {
    dropTable(new String[] {"tab_acid"});
    FileSystem fs = FileSystem.get(conf);
    
    HiveConf.setIntVar(conf, HiveConf.ConfVars.HIVE_LOCKS_PARTITION_THRESHOLD, 1);
    driver = Mockito.spy(driver);

    HiveConf.setBoolVar(driver2.getConf(), HiveConf.ConfVars.HIVE_ACID_DROP_PARTITION_USE_BASE, !blocking);
    driver2 = Mockito.spy(driver2);

    driver.run("create table if not exists tab_acid (a int, b int) partitioned by (p string) " +
      "stored as orc TBLPROPERTIES ('transactional'='true')");
    driver.run("insert into tab_acid partition(p) (a,b,p) values(1,2,'foo'),(3,4,'bar')");

    driver.compileAndRespond("select * from tab_acid");
    List<String> res = new ArrayList<>();

    driver.lockAndRespond();
    List<ShowLocksResponseElement> locks = getLocks();
    Assert.assertEquals("Unexpected lock count", 1, locks.size());

    TestTxnDbUtil.checkLock(LockType.SHARED_READ,
      LockState.ACQUIRED, "default", "tab_acid", null, locks);

    DbTxnManager txnMgr2 = (DbTxnManager) TxnManagerFactory.getTxnManagerFactory().getTxnManager(conf);
    swapTxnManager(txnMgr2);
    driver2.compileAndRespond("alter table tab_acid drop partition (p='foo')");

    if (blocking) {
      txnMgr2.acquireLocks(driver2.getPlan(), ctx, null, false);
      locks = getLocks();

      ShowLocksResponseElement checkLock = TestTxnDbUtil.checkLock(LockType.EXCLUSIVE,
        LockState.WAITING, "default", "tab_acid", "p=foo", locks);

      swapTxnManager(txnMgr);
      Mockito.doNothing().when(driver).lockAndRespond();
      driver.run();

      driver.getFetchTask().fetch(res);
      swapTxnManager(txnMgr2);

      ReflectionUtil.setField(txnMgr2, "numStatements", 0);
      txnMgr2.getMS().unlock(checkLock.getLockid());
    }
    driver2.lockAndRespond();
    locks = getLocks();
    Assert.assertEquals("Unexpected lock count", blocking ? 1 : 2, locks.size());

    TestTxnDbUtil.checkLock(blocking ? LockType.EXCLUSIVE : LockType.EXCL_WRITE,
      LockState.ACQUIRED, "default", "tab_acid", "p=foo", locks);

    Mockito.doNothing().when(driver2).lockAndRespond();
    driver2.run();

    if (!blocking) {
      swapTxnManager(txnMgr);
      Mockito.doNothing().when(driver).lockAndRespond();
      driver.run();
    }
    Mockito.reset(driver, driver2);
    
    FileStatus[] stat = fs.listStatus(new Path(getWarehouseDir(), "tab_acid" + (blocking ? "" : "/p=foo")),
      (blocking ? path -> path.getName().equals("p=foo") : AcidUtils.baseFileFilter));
    if ((blocking ? 0 : 1) != stat.length) {
      Assert.fail("Partition data was " + (blocking ? "not " : "") + "removed from FS");
    }
    driver.getFetchTask().fetch(res);
    Assert.assertEquals("Expecting 2 rows and found " + res.size(), 2, res.size());
    
    driver.run("select * from tab_acid where p='foo'");
    res = new ArrayList<>();
    driver.getFetchTask().fetch(res);
    Assert.assertEquals("Expecting 0 rows and found " + res.size(), 0, res.size());

    //re-create partition with the same name
    driver.run("insert into tab_acid partition(p) (a,b,p) values(1,2,'foo')");
    
    driver.run("select * from tab_acid where p='foo'");
    res = new ArrayList<>();
    driver.getFetchTask().fetch(res);
    Assert.assertEquals("Expecting 1 rows and found " + res.size(), 1, res.size());
  }
T7 is a table with 2 partitions
1. run select from T7
2. run drop partition from T7
concurrently with 1 starting first so that 2 blocks
3. start another concurrent select on T7 - it should block behind waiting X (from drop) - LM should be fair
4. finish #1 so that drop unblocks
5. rollback the drop to release its X lock
6. # should unblock
ql/src/test/org/apache/hadoop/hive/ql/lockmgr/TestDbTxnManager2.java