@Test
  public void testTargetEventIdGenerationInOptmisedFailover() throws Throwable {
    // Do a a cycle of bootstrap dump & load.
    List<String> withClause = ReplicationTestUtils.includeExternalTableClause(true);
    withClause.add("'" + HiveConf.ConfVars.REPL_DIR.varname + "'='" + primary.repldDir + "'");

    // Do a bootstrap cycle(A->B)
    primary.dump(primaryDbName, withClause);
    replica.load(replicatedDbName, primaryDbName, withClause);

    // Add some table & do the first incremental dump.
    primary.run("use " + primaryDbName)
            .run("create external table tablei1 (id int)")
            .run("create external table tablei2 (id int)")
            .run("create table tablem1 (id int)")
            .run("create table tablem2 (id int)")
            .run("insert into table tablei1 values(1),(2),(3),(4)")
            .run("insert into table tablei2 values(10),(20),(30),(40)")
            .run("insert into table tablem1 values(5),(10),(15),(20)")
            .run("insert into table tablem2 values(6),(12),(18),(24)")
            .dump(primaryDbName, withClause);

    // Do the incremental load, and check everything is intact.
    replica.load(replicatedDbName, primaryDbName, withClause)
            .run("use "+ replicatedDbName)
            .run("select id from tablei1")
            .verifyResults(new String[]{"1","2","3","4"})
            .run("select id from tablei2")
            .verifyResults(new String[]{"10","20","30","40"})
            .run("select id from tablem1")
            .verifyResults(new String[]{"5","10","15","20"})
            .run("select id from tablem2")
            .verifyResults(new String[]{"6","12","18","24"});

    // Do some modifications & call for the second cycle of incremental dump & load.
    WarehouseInstance.Tuple tuple = primary.run("use " + primaryDbName)
            .run("create external table table1 (id int)")
            .run("insert into table table1 values (25),(35),(82)")
            .run("create  table table1_managed (name string)")
            .run("insert into table table1_managed values ('CAD'),('DAS'),('MSA')")
            .run("insert into table tablei1 values(15),(62),(25),(62)")
            .run("insert into table tablei2 values(10),(22),(11),(22)")
            .run("insert into table tablem1 values(5),(10),(15),(20)")
            .run("alter table table1 set TBLPROPERTIES('comment'='abc')")
            .dump(primaryDbName, withClause);

    // Do an incremental load
    replica.load(replicatedDbName, primaryDbName, withClause);

    // Get the latest notification from the notification log for the target database, just after replication.
    CurrentNotificationEventId notificationIdAfterRepl = replica.getCurrentNotificationEventId();

    // Check the tables are there post incremental load.
    replica.run("repl status " + replicatedDbName)
            .verifyResult(tuple.lastReplicationId)
            .run("use " + replicatedDbName)
            .run("select id from table1")
            .verifyResults(new String[]{"25", "35", "82"})
            .run("select name from table1_managed")
            .verifyResults(new String[]{"CAD", "DAS", "MSA"})
            .verifyReplTargetProperty(replicatedDbName);

    // Do some modifications on the source cluster, so we have some entries in the table diff.
    primary.run("use " + primaryDbName)
            .run("create table table2_managed (id string)")
            .run("insert into table table1_managed values ('AAA'),('BBB')")
            .run("insert into table table2_managed values ('A1'),('B1'),('C2')");


    // Do some modifications in another database to have unrelated events as well after the last load, which should
    // get filtered.

    primary.run("create database " + extraPrimaryDb)
            .run("use " + extraPrimaryDb)
            .run("create external table table1 (id int)")
            .run("insert into table table1 values (15),(1),(96)")
            .run("create  table table1_managed (id string)")
            .run("insert into table table1_managed values ('SAA'),('PSA')");

    // Do some modifications on the target database.
    replica.run("use " + replicatedDbName)
            .run("alter database "+ replicatedDbName + " set DBPROPERTIES ('repl1'='value1')")
            .run("alter database "+ replicatedDbName + " set DBPROPERTIES ('repl2'='value2')");

    // Validate the current replication id on original target has changed now.
    assertNotEquals(replica.getCurrentNotificationEventId().getEventId(), notificationIdAfterRepl.getEventId());

    // Prepare for reverse replication.
    DistributedFileSystem replicaFs = replica.miniDFSCluster.getFileSystem();
    Path newReplDir = new Path(replica.repldDir + "reverse01");
    replicaFs.mkdirs(newReplDir);
    withClause = ReplicationTestUtils.includeExternalTableClause(true);
    withClause.add("'" + HiveConf.ConfVars.REPL_DIR.varname + "'='" + newReplDir + "'");

    tuple = replica.dump(replicatedDbName, withClause);

    // Check event ack file should get created.
    assertTrue(new Path(tuple.dumpLocation, EVENT_ACK_FILE).toString() + " doesn't exist",
            replicaFs.exists(new Path(tuple.dumpLocation, EVENT_ACK_FILE)));

    // Get the target event id.
    NotificationEventResponse nl = new HiveMetaStoreClient(replica.hiveConf)
            .getNextNotification(Long.parseLong(getEventIdFromFile(new Path(tuple.dumpLocation), conf)[1]), 10,
                    new DatabaseAndTableFilter(replicatedDbName, null));

    assertEquals(0, nl.getEventsSize());
  }
Test replication scenarios with staging on replica.
itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenariosExclusiveReplica.java