@Test
  public void testFailoverFailureBeforeReverseReplication() throws Throwable {
    List<String> failoverConfigs = Arrays.asList("'" + HiveConf.ConfVars.HIVE_REPL_FAILOVER_START + "'='true'",
            "'" + HiveConf.ConfVars.REPL_RETAIN_PREV_DUMP_DIR + "'='true'",
            "'" + HiveConf.ConfVars.REPL_RETAIN_PREV_DUMP_DIR_COUNT + "'='1'");
    List<String> retainPrevDumpDir = Arrays.asList("'" + HiveConf.ConfVars.REPL_RETAIN_PREV_DUMP_DIR + "'='true'",
            "'" + HiveConf.ConfVars.REPL_RETAIN_PREV_DUMP_DIR_COUNT + "'='1'",
            "'" + HiveConf.ConfVars.REPL_RUN_DATA_COPY_TASKS_ON_TARGET + "'='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, FailoverMetaData.FAILOVER_METADATA)));
    assertFalse(fs.exists(new Path(dumpPath, ReplAck.FAILOVER_READY_MARKER.toString())));
    assertFalse(MetaStoreUtils.isDbBeingPlannedFailedOver(primary.getDatabase(primaryDbName)));

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

    assertTrue(MetaStoreUtils.isTargetOfReplication(replica.getDatabase(replicatedDbName)));
    dumpData = primary.run("use " + primaryDbName)
            .run("insert into t1 values(1)")
            .run("insert into t2 partition(name='Bob') values(11)")
            .dump(primaryDbName, failoverConfigs);

    dumpPath = new Path(dumpData.dumpLocation, ReplUtils.REPL_HIVE_BASE_DIR);
    Path dumpAckFile = new Path(dumpPath, DUMP_ACKNOWLEDGEMENT.toString());
    Path failoverMdFile = new Path(dumpPath, FailoverMetaData.FAILOVER_METADATA);
    assertTrue(fs.exists(dumpAckFile));
    assertTrue(fs.exists(failoverMdFile));
    assertTrue(fs.exists(new Path(dumpPath, ReplAck.FAILOVER_READY_MARKER.toString())));
    assertTrue(MetaStoreUtils.isDbBeingPlannedFailedOverAtEndpoint(primary.getDatabase(primaryDbName),
            MetaStoreUtils.FailoverEndpoint.SOURCE));
    FailoverMetaData previousFmd = new FailoverMetaData(dumpPath, conf);
    Long failoverEventId = previousFmd.getFailoverEventId();
    assertTrue(failoverEventId >= Long.parseLong(dumpData.lastReplicationId));
    Long failoverMdModifTime = fs.getFileStatus(failoverMdFile).getModificationTime();

    fs.delete(dumpAckFile, false);

    dumpData = primary.run("use " + primaryDbName)
            .run("insert into t2 partition(name='Carl') values(10)")
            .run("create table t3 (id int)")
            .run("insert into t3 values (2)")
            .dump(primaryDbName, failoverConfigs);

    dumpPath = new Path(dumpData.dumpLocation, ReplUtils.REPL_HIVE_BASE_DIR);
    failoverMdFile = new Path(dumpPath, FailoverMetaData.FAILOVER_METADATA);
    assertTrue(fs.exists(new Path(dumpPath, DUMP_ACKNOWLEDGEMENT.toString())));
    assertTrue(fs.exists(failoverMdFile));
    Assert.assertEquals(failoverMdModifTime, (Long)fs.getFileStatus(failoverMdFile).getModificationTime());
    assertTrue(fs.exists(new Path(dumpPath, ReplAck.FAILOVER_READY_MARKER.toString())));
    assertTrue(MetaStoreUtils.isDbBeingPlannedFailedOverAtEndpoint(primary.getDatabase(primaryDbName),
            MetaStoreUtils.FailoverEndpoint.SOURCE));
    assertTrue(failoverEventId >= Long.parseLong(dumpData.lastReplicationId));
    FailoverMetaData currentFmd = new FailoverMetaData(dumpPath, conf);
    assertTrue(currentFmd.equals(previousFmd));

    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[]{"11"});

    Database 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())));

    WarehouseInstance.Tuple reverseDumpData = replica.run("use " + replicatedDbName)
            .run("insert into t2 partition(name='Carl') values(12)")
            .run("create table t3 (id int)")
            .run("insert into t3 values (10)")
            .dump(replicatedDbName, retainPrevDumpDir);
    assertNotEquals(reverseDumpData.dumpLocation, dumpData.dumpLocation);
    assertTrue(fs.exists(dumpPath));
    assertTrue(fs.exists(new Path(dumpPath, ReplAck.FAILOVER_READY_MARKER.toString())));
    dumpPath = new Path(reverseDumpData.dumpLocation, ReplUtils.REPL_HIVE_BASE_DIR);
    dumpAckFile = new Path(dumpPath, DUMP_ACKNOWLEDGEMENT.toString());
    assertTrue(fs.exists(dumpAckFile));
    assertFalse(fs.exists(new Path(dumpPath, ReplAck.FAILOVER_READY_MARKER.toString())));
    assertTrue(new DumpMetaData(dumpPath, conf).getDumpType() == DumpType.PRE_OPTIMIZED_BOOTSTRAP);
    db = replica.getDatabase(replicatedDbName);
    assertTrue(MetaStoreUtils.isDbBeingPlannedFailedOverAtEndpoint(db, MetaStoreUtils.FailoverEndpoint.TARGET));
    assertTrue(MetaStoreUtils.isTargetOfReplication(db));
    primary.load(primaryDbName, replicatedDbName, retainPrevDumpDir);
    //do a second reverse dump.
    reverseDumpData = replica.dump(replicatedDbName, retainPrevDumpDir);
    dumpPath = new Path(reverseDumpData.dumpLocation, ReplUtils.REPL_HIVE_BASE_DIR);

    primary.load(primaryDbName, replicatedDbName, retainPrevDumpDir)
            .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[]{"11", "12"})
            .run("select id from t3")
            .verifyResults(new String[]{"10"});

    assertTrue(fs.exists(new Path(dumpPath, LOAD_ACKNOWLEDGEMENT.toString())));

    reverseDumpData = replica.run("use " + replicatedDbName)
            .run("insert into t3 values (15)")
            .dump(replicatedDbName);
    dumpPath = new Path(reverseDumpData.dumpLocation, ReplUtils.REPL_HIVE_BASE_DIR);
    fs.exists(new Path(dumpPath, DUMP_ACKNOWLEDGEMENT.toString()));
    assertFalse(MetaStoreUtils.isDbBeingPlannedFailedOver(replica.getDatabase(replicatedDbName)));

    primary.load(primaryDbName, replicatedDbName)
            .run("use " + primaryDbName)
            .run("select id from t3")
            .verifyResults(new String[]{"10", "15"});;
    assertTrue(fs.exists(new Path(dumpPath, LOAD_ACKNOWLEDGEMENT.toString())));
  }
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.
itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenariosAcidTables.java