@Test
  public void testFailoverFailureInReverseReplication() 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);
    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));

    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='Bob') values(20)")
            .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);
    Path 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));
    fs.delete(dumpAckFile, false);
    assertFalse(fs.exists(dumpAckFile));
    WarehouseInstance.Tuple preFailoverDumpData = dumpData;
    dumpData = replica.dump(replicatedDbName, retainPrevDumpDir);
    dumpPath = new Path(dumpData.dumpLocation, ReplUtils.REPL_HIVE_BASE_DIR);
    dumpAckFile = new Path(dumpPath, DUMP_ACKNOWLEDGEMENT.toString());
    assertNotEquals(dumpData.dumpLocation, preFailoverDumpData.dumpLocation);
    assertTrue(fs.exists(new Path(preFailoverDumpData.dumpLocation)));
    assertNotEquals(reverseDumpData.dumpLocation, dumpData.dumpLocation);
    assertFalse(fs.exists(new Path(dumpPath, ReplAck.FAILOVER_READY_MARKER.toString())));
    assertTrue(new DumpMetaData(dumpPath, conf).getDumpType() == DumpType.PRE_OPTIMIZED_BOOTSTRAP);
    assertTrue(fs.exists(dumpAckFile));
    db = replica.getDatabase(replicatedDbName);
    assertTrue(MetaStoreUtils.isDbBeingPlannedFailedOverAtEndpoint(db, MetaStoreUtils.FailoverEndpoint.TARGET));
    assertTrue(MetaStoreUtils.isTargetOfReplication(db));
    primary.load(primaryDbName, replicatedDbName);
    dumpData = replica.dump(replicatedDbName, retainPrevDumpDir);

    primary.load(primaryDbName, replicatedDbName, retainPrevDumpDir)
            .run("use " + primaryDbName)
            .run("show tables")
            .verifyResults(new String[]{"t1", "t2", "t3"})
            .run("repl status " + primaryDbName)
            .verifyResult(dumpData.lastReplicationId)
            .run("select id from t1")
            .verifyResults(new String[]{"1"})
            .run("select rank from t2 order by rank")
            .verifyResults(new String[]{"11", "20"})
            .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 (3)")
            .run("insert into t2 partition(name='Bob') values(30)")
            .dump(replicatedDbName, retainPrevDumpDir);
    dumpPath = new Path(reverseDumpData.dumpLocation, ReplUtils.REPL_HIVE_BASE_DIR);
    dumpAckFile = new Path(dumpPath, DUMP_ACKNOWLEDGEMENT.toString());
    assertFalse(MetaStoreUtils.isDbBeingPlannedFailedOverAtEndpoint(replica.getDatabase(replicatedDbName),
            MetaStoreUtils.FailoverEndpoint.TARGET));
    fs.delete(dumpAckFile);
    replica.run("ALTER DATABASE " + replicatedDbName + " SET DBPROPERTIES('" + ReplConst.REPL_FAILOVER_ENDPOINT + "'='"
            + MetaStoreUtils.FailoverEndpoint.TARGET + "')");
    assertTrue(MetaStoreUtils.isDbBeingPlannedFailedOverAtEndpoint(replica.getDatabase(replicatedDbName),
            MetaStoreUtils.FailoverEndpoint.TARGET));
    assertFalse(fs.exists(dumpAckFile));
    dumpData = replica.dump(replicatedDbName, retainPrevDumpDir);
    assertEquals(reverseDumpData.dumpLocation, dumpData.dumpLocation);
    fs.exists(dumpAckFile);
    assertFalse(MetaStoreUtils.isDbBeingPlannedFailedOverAtEndpoint(replica.getDatabase(replicatedDbName),
            MetaStoreUtils.FailoverEndpoint.TARGET));

    primary.load(primaryDbName, replicatedDbName, retainPrevDumpDir)
            .run("use " + primaryDbName)
            .run("select rank from t2 order by rank")
            .verifyResults(new String[]{"11", "20", "30"})
            .run("select id from t3")
            .verifyResults(new String[]{"10", "3"});;
    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