@Test
  public void testCompleteFailoverWithReverseBootstrap() throws Throwable {
    HiveConf primaryConf = primary.getConf();
    TxnStore txnHandler = TxnUtils.getTxnStore(primary.getConf());
    List<String> failoverConfigs = Arrays.asList("'" + HiveConf.ConfVars.HIVE_REPL_FAILOVER_START + "'='true'");
    WarehouseInstance.Tuple dumpData = primary.run("use " + primaryDbName)
            .run("create table t1 (id int) clustered by(id) into 3 buckets stored as orc " +
                    "tblproperties (\"transactional\"=\"true\")")
            .run("create table t2 (rank int) partitioned by (name string) tblproperties(\"transactional\"=\"true\", " +
                    "\"transactional_properties\"=\"insert_only\")")
            .dump(primaryDbName, failoverConfigs);

    //This dump is not failover ready as target db can be used for replication only after first incremental load.
    FileSystem fs = new Path(dumpData.dumpLocation).getFileSystem(conf);
    Path dumpPath = new Path(dumpData.dumpLocation, ReplUtils.REPL_HIVE_BASE_DIR);
    assertFalse(fs.exists(new Path(dumpPath, ReplAck.FAILOVER_READY_MARKER.toString())));
    assertFalse(MetaStoreUtils.isDbBeingPlannedFailedOver(primary.getDatabase(primaryDbName)));

    replica.load(replicatedDbName, primaryDbName, failoverConfigs)
            .run("use " + replicatedDbName)
            .run("show tables")
            .verifyResults(new String[]{"t1", "t2"})
            .run("repl status " + replicatedDbName)
            .verifyResult(dumpData.lastReplicationId);

    Database db = replica.getDatabase(replicatedDbName);
    assertTrue(MetaStoreUtils.isTargetOfReplication(db));
    assertFalse(MetaStoreUtils.isDbBeingPlannedFailedOver(db));

    primary.run("use " + primaryDbName)
            .run("insert into t1 values(1)")
            .run("insert into t2 partition(name='Bob') values(11)")
            .run("insert into t2 partition(name='Carl') values(10)");

    /**Open transactions can be of two types:
     Case 1) Txns that have not acquired HIVE LOCKS or they belong to different db: These txns would be captured in
     _failovermetadata file inside dump directory.
     Case 2) Txns that have acquired HIVE LOCKS and belong to db under replication: These txns would be aborted by hive
     as part of dump operation.
     */
    // Open 3 txns for Database which is not under replication
    int numTxnsForSecDb = 3;
    List<Long> txnsForSecDb = openTxns(numTxnsForSecDb, txnHandler, primaryConf);

    // Allocate write ids for both tables of secondary db for 3 txns
    // t1=5 and t2=5
    Map<String, Long> tablesInSecDb = new HashMap<>();
    tablesInSecDb.put("t1", (long) numTxnsForSecDb);
    tablesInSecDb.put("t2", (long) numTxnsForSecDb);
    List<Long> lockIdsForSecDb = allocateWriteIdsForTablesAndAcquireLocks(primaryDbName + "_extra",
            tablesInSecDb, txnHandler, txnsForSecDb, primaryConf);

    //Open 2 txns for Primary Db
    int numTxnsForPrimaryDb = 2;
    List<Long> txnsForPrimaryDb = openTxns(numTxnsForPrimaryDb, txnHandler, primaryConf);

    // Allocate write ids for both tables of primary db for 2 txns
    // t1=5 and t2=5
    Map<String, Long> tablesInPrimaryDb = new HashMap<>();
    tablesInPrimaryDb.put("t1", (long) numTxnsForPrimaryDb + 1);
    tablesInPrimaryDb.put("t2", (long) numTxnsForPrimaryDb + 2);
    List<Long> lockIdsForPrimaryDb = allocateWriteIdsForTablesAndAcquireLocks(primaryDbName,
            tablesInPrimaryDb, txnHandler, txnsForPrimaryDb, primaryConf);

    //Open 1 txn with no hive locks acquired
    List<Long> txnsWithNoLocks = openTxns(1, txnHandler, primaryConf);

    dumpData = primary.dump(primaryDbName, failoverConfigs);

    dumpPath = new Path(dumpData.dumpLocation, ReplUtils.REPL_HIVE_BASE_DIR);
    assertTrue(fs.exists(new Path(dumpPath, DUMP_ACKNOWLEDGEMENT.toString())));
    assertTrue(fs.exists(new Path(dumpPath, FailoverMetaData.FAILOVER_METADATA)));
    assertTrue(fs.exists(new Path(dumpPath, ReplAck.FAILOVER_READY_MARKER.toString())));
    assertTrue(MetaStoreUtils.isDbBeingPlannedFailedOverAtEndpoint(primary.getDatabase(primaryDbName),
            MetaStoreUtils.FailoverEndpoint.SOURCE));
    FailoverMetaData failoverMD = new FailoverMetaData(dumpPath, conf);

    List<Long> openTxns = failoverMD.getOpenTxns();
    List<Long> txnsAborted = failoverMD.getAbortedTxns();
    assertTrue(txnsAborted.size() == 2);
    assertTrue(txnsAborted.containsAll(txnsForPrimaryDb));
    assertTrue(openTxns.size() == 4);
    assertTrue(openTxns.containsAll(txnsForSecDb));
    assertTrue(openTxns.containsAll(txnsWithNoLocks));
    assertTrue(failoverMD.getTxnsWithoutLock().equals(txnsWithNoLocks));


    //TxnsForPrimaryDb and txnsWithNoLocks would have been aborted by dump operation.
    verifyAllOpenTxnsAborted(txnsForPrimaryDb, primaryConf);
    verifyAllOpenTxnsNotAborted(txnsForSecDb, primaryConf);
    verifyAllOpenTxnsNotAborted(txnsWithNoLocks, primaryConf);
    //Abort the txns
    txnHandler.abortTxns(new AbortTxnsRequest(txnsForSecDb));
    txnHandler.abortTxns(new AbortTxnsRequest(txnsWithNoLocks));
    verifyAllOpenTxnsAborted(txnsForSecDb, primaryConf);
    verifyAllOpenTxnsAborted(txnsWithNoLocks, primaryConf);
    releaseLocks(txnHandler, lockIdsForSecDb);

    replica.load(replicatedDbName, primaryDbName, failoverConfigs)
            .run("use " + replicatedDbName)
            .run("show tables")
            .verifyResults(new String[]{"t1", "t2"})
            .run("repl status " + replicatedDbName)
            .verifyResult(dumpData.lastReplicationId)
            .run("select id from t1")
            .verifyResults(new String[]{"1"})
            .run("select rank from t2 order by rank")
            .verifyResults(new String[]{"10", "11"});

    db = replica.getDatabase(replicatedDbName);
    assertTrue(MetaStoreUtils.isTargetOfReplication(db));
    assertTrue(MetaStoreUtils.isDbBeingPlannedFailedOverAtEndpoint(db, MetaStoreUtils.FailoverEndpoint.TARGET));
    assertTrue(fs.exists(new Path(dumpPath, ReplAck.LOAD_ACKNOWLEDGEMENT.toString())));

    Path dbRootDir = new Path(dumpData.dumpLocation).getParent();
    long prevDumpDirModifTime = getLatestDumpDirModifTime(dbRootDir);
    primary.run("REPL DUMP " + primaryDbName + " with ('" + HiveConf.ConfVars.HIVE_REPL_FAILOVER_START + "' = 'true')");
    Assert.assertEquals(dumpData.dumpLocation, ReplUtils.getLatestDumpPath(dbRootDir, conf).toString());
    Assert.assertEquals(prevDumpDirModifTime, getLatestDumpDirModifTime(dbRootDir));
    dumpPath = new Path(dumpData.dumpLocation, ReplUtils.REPL_HIVE_BASE_DIR);
    assertTrue(fs.exists(new Path(dumpPath, ReplAck.LOAD_ACKNOWLEDGEMENT.toString())));

    assertTrue(MetaStoreUtils.isDbBeingPlannedFailedOverAtEndpoint(primary.getDatabase(primaryDbName),
            MetaStoreUtils.FailoverEndpoint.SOURCE));

    assertFalse(ReplChangeManager.isSourceOfReplication(replica.getDatabase(replicatedDbName)));

    WarehouseInstance.Tuple reverseDumpData = replica.run("create table t3 (id int)")
                                                      .run("insert into t2 partition(name='Bob') values(20)")
                                                      .run("insert into t3 values (2)")
                                                      .dump(replicatedDbName);

    assertNotEquals(reverseDumpData.dumpLocation, dumpData.dumpLocation);
    assertTrue(fs.exists(new Path(dumpPath, ReplAck.FAILOVER_READY_MARKER.toString())));
    dumpPath = new Path(reverseDumpData.dumpLocation, ReplUtils.REPL_HIVE_BASE_DIR);
    assertFalse(fs.exists(new Path(dumpPath, ReplAck.FAILOVER_READY_MARKER.toString())));
    assertTrue(new DumpMetaData(dumpPath, conf).getDumpType() == DumpType.PRE_OPTIMIZED_BOOTSTRAP);
    assertTrue(fs.exists(new Path(dumpPath, DUMP_ACKNOWLEDGEMENT.toString())));
    db = replica.getDatabase(replicatedDbName);
    assertTrue(MetaStoreUtils.isDbBeingPlannedFailedOverAtEndpoint(db, MetaStoreUtils.FailoverEndpoint.TARGET));
    assertTrue(MetaStoreUtils.isTargetOfReplication(db));
    //do a second reverse dump.
    primary.load(primaryDbName, replicatedDbName);
    reverseDumpData = replica.dump(replicatedDbName);
    dumpPath = new Path(reverseDumpData.dumpLocation, ReplUtils.REPL_HIVE_BASE_DIR);

    primary.load(primaryDbName, replicatedDbName)
            .run("use " + primaryDbName)
            .run("show tables")
            .verifyResults(new String[]{"t1", "t2", "t3"})
            .run("repl status " + primaryDbName)
            .verifyResult(reverseDumpData.lastReplicationId)
            .run("select id from t1")
            .verifyResults(new String[]{"1"})
            .run("select rank from t2 order by rank")
            .verifyResults(new String[]{"10", "11", "20"})
            .run("select id from t3")
            .verifyResults(new String[]{"2"});

    Database primaryDb = primary.getDatabase(primaryDbName);
    assertFalse(primaryDb == null);
    assertFalse(ReplUtils.isFirstIncPending(primaryDb.getParameters()));
    assertTrue(MetaStoreUtils.isTargetOfReplication(primaryDb));
    assertFalse(MetaStoreUtils.isDbBeingPlannedFailedOver(primaryDb));
    assertTrue(fs.exists(new Path(dumpPath, LOAD_ACKNOWLEDGEMENT.toString())));
    assertFalse(ReplChangeManager.isSourceOfReplication(primaryDb));
    assertTrue(ReplChangeManager.isSourceOfReplication(replica.getDatabase(replicatedDbName)));

    reverseDumpData = replica.run("insert into t3 values (3)")
            .run("insert into t2 partition(name='Bob') values(30)")
            .dump(replicatedDbName);
    assertFalse(MetaStoreUtils.isDbBeingPlannedFailedOver(replica.getDatabase(replicatedDbName)));

    primary.load(primaryDbName, replicatedDbName)
            .run("select rank from t2 order by rank")
            .verifyResults(new String[]{"10", "11", "20", "30"})
            .run("select id from t3")
            .verifyResults(new String[]{"2", "3"})
            .run("repl status " + primaryDbName)
            .verifyResult(reverseDumpData.lastReplicationId);
    assertFalse(ReplUtils.isFirstIncPending(primary.getDatabase(primaryDbName).getParameters()));
  }
TestReplicationScenariosAcidTables - test replication for ACID tables.
itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenariosAcidTables.java