@Test
  public void testReplAllocWriteId() throws Exception {
    int numTxn = 2;
    String[] output = TestTxnDbUtil.queryToString(conf, "SELECT MAX(\"TXN_ID\") + 1 FROM \"TXNS\"").split("\n");
    long startTxnId = Long.parseLong(output[1].trim());
    List<Long> srcTxnIdList = LongStream.rangeClosed(startTxnId, numTxn+startTxnId-1)
            .boxed().collect(Collectors.toList());
    List<Long> targetTxnList = replOpenTxnForTest(startTxnId, numTxn, "destdb.*");
    assert(targetTxnList.size() == numTxn);

    List<TxnToWriteId> srcTxnToWriteId;
    List<TxnToWriteId> targetTxnToWriteId;
    srcTxnToWriteId = new ArrayList<>();

    for (int idx = 0; idx < numTxn; idx++) {
      srcTxnToWriteId.add(new TxnToWriteId(startTxnId+idx, idx+1));
    }
    AllocateTableWriteIdsRequest allocMsg = new AllocateTableWriteIdsRequest("destdb", "tbl1");
    allocMsg.setReplPolicy("destdb.*");
    allocMsg.setSrcTxnToWriteIdList(srcTxnToWriteId);
    targetTxnToWriteId = txnHandler.allocateTableWriteIds(allocMsg).getTxnToWriteIds();
    for (int idx = 0; idx < targetTxnList.size(); idx++) {
      assertEquals(targetTxnToWriteId.get(idx).getWriteId(), srcTxnToWriteId.get(idx).getWriteId());
      assertEquals(Long.valueOf(targetTxnToWriteId.get(idx).getTxnId()), targetTxnList.get(idx));
    }

    // idempotent case for destdb db
    targetTxnToWriteId = txnHandler.allocateTableWriteIds(allocMsg).getTxnToWriteIds();
    for (int idx = 0; idx < targetTxnList.size(); idx++) {
      assertEquals(targetTxnToWriteId.get(idx).getWriteId(), srcTxnToWriteId.get(idx).getWriteId());
      assertEquals(Long.valueOf(targetTxnToWriteId.get(idx).getTxnId()), targetTxnList.get(idx));
    }

    //invalid case
    boolean failed = false;
    srcTxnToWriteId = new ArrayList<>();
    srcTxnToWriteId.add(new TxnToWriteId(startTxnId, 2*numTxn+1));
    allocMsg = new AllocateTableWriteIdsRequest("destdb", "tbl2");
    allocMsg.setReplPolicy("destdb.*");
    allocMsg.setSrcTxnToWriteIdList(srcTxnToWriteId);

    // This is an idempotent case when repl flow forcefully allocate write id if it doesn't match
    // the next write id.
    try {
      txnHandler.allocateTableWriteIds(allocMsg).getTxnToWriteIds();
    } catch (IllegalStateException e) {
      failed = true;
    }
    assertFalse(failed);

    replAbortTxnForTest(srcTxnIdList, "destdb.*");

    // Test for aborted transactions. Idempotent case where allocate write id when txn is already
    // aborted should do nothing.
    failed = false;
    try {
      txnHandler.allocateTableWriteIds(allocMsg).getTxnToWriteIds();
    } catch (RuntimeException e) {
      failed = true;
    }
    assertFalse(failed);
  }
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
ql/src/test/org/apache/hadoop/hive/metastore/txn/TestTxnHandler.java