@Test
  public void testTargetEventIdGenerationAfterFirstIncrementalInOptFailover() throws Throwable {
    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 an incremental dump.
    WarehouseInstance.Tuple tuple = primary.run("use " + primaryDbName)
            .run("create external table table1 (id int)")
            .run("insert into table table1 values (100)")
            .run("create  table table1_managed (name string)")
            .run("insert into table table1_managed values ('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")
            .verifyResult("100")
            .run("select name from table1_managed")
            .verifyResult("ABC")
            .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 ('SDC')")
            .run("insert into table table2_managed values ('A'),('B'),('C')");


    // 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 t1 (id int)")
            .run("insert into table t1 values (15),(1),(96)")
            .run("create  table t1_managed (id string)")
            .run("insert into table t1_managed values ('SA'),('PS')");

    // Do some modifications on the target database.
    replica.run("use " + replicatedDbName)
            .run("alter database "+ replicatedDbName + " set DBPROPERTIES ('key1'='value1')")
            .run("alter database "+ replicatedDbName + " set DBPROPERTIES ('key2'='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 + "reverse1");
    replicaFs.mkdirs(newReplDir);
    withClause = ReplicationTestUtils.includeExternalTableClause(true);
    withClause.add("'" + HiveConf.ConfVars.REPL_DIR.varname + "'='" + newReplDir + "'");

    tuple = replica.dump(replicatedDbName);

    // 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]), -1,
                    new DatabaseAndTableFilter(replicatedDbName, null));

    // There should be 2 events, two custom alter operations.
    assertEquals(2, nl.getEvents().size());
  }
Test replication scenarios with staging on replica.
itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenariosExclusiveReplica.java