@Test
  public void testEnablementOfReplBackgroundThreadDuringFailover() throws Throwable{
    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);

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

    dumpData = 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)")
            .dump(primaryDbName, failoverConfigs);

    dumpPath = new Path(dumpData.dumpLocation, ReplUtils.REPL_HIVE_BASE_DIR);
    Path failoverReadyMarker = new Path(dumpPath, ReplAck.FAILOVER_READY_MARKER.toString());
    assertTrue(fs.exists(new Path(dumpPath, DUMP_ACKNOWLEDGEMENT.toString())));
    assertTrue(fs.exists(new Path(dumpPath, FailoverMetaData.FAILOVER_METADATA)));
    assertTrue(fs.exists(failoverReadyMarker));
    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[]{"10", "11"});

    assertTrue(fs.exists(new Path(dumpPath, LOAD_ACKNOWLEDGEMENT.toString())));
    db = replica.getDatabase(replicatedDbName);
    assertTrue(MetaStoreUtils.isTargetOfReplication(db));
    assertTrue(MetaStoreUtils.isDbBeingPlannedFailedOverAtEndpoint(db, MetaStoreUtils.FailoverEndpoint.TARGET));
    assert "true".equals(db.getParameters().get(REPL_ENABLE_BACKGROUND_THREAD));
  }
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