@Test
  public void testReplConfiguredCleanupOfNotificationEvents() throws Exception {

    boolean verifySetupOriginal = verifySetupSteps;
    verifySetupSteps = true;
    final int cleanerTtlSeconds = 1;
    final int cleanerIntervalSeconds = 1;
    String nameOfTest = testName.getMethodName();
    String dbName = createDB(nameOfTest, driver);
    String replDbName = dbName + "_dupe";

    run("CREATE TABLE " + dbName + ".unptned(a string) STORED AS TEXTFILE", driver);
    run("CREATE TABLE " + dbName + ".ptned(a string) partitioned by (b int) STORED AS TEXTFILE", driver);

    //bootstrap
    bootstrapLoadAndVerify(dbName, replDbName);

    String[] unptnData = new String[] {"eleven", "twelve"};
    String[] ptnData1 = new String[] {"thirteen", "fourteen", "fifteen"};
    String[] ptnData2 = new String[] {"fifteen", "sixteen", "seventeen"};
    String[] empty = new String[] {};

    String unptnLocn = new Path(TEST_PATH, nameOfTest + "_unptn").toUri().getPath();
    String ptnLocn1 = new Path(TEST_PATH, nameOfTest + "_ptn1").toUri().getPath();
    String ptnLocn2 = new Path(TEST_PATH, nameOfTest + "_ptn2").toUri().getPath();

    createTestDataFile(unptnLocn, unptnData);
    createTestDataFile(ptnLocn1, ptnData1);
    createTestDataFile(ptnLocn2, ptnData2);

    run("LOAD DATA LOCAL INPATH '" + unptnLocn + "' OVERWRITE INTO TABLE " + dbName + ".unptned", driver);
    verifySetup("SELECT * from " + dbName + ".unptned", unptnData, driver);
    run("CREATE TABLE " + dbName + ".unptned_late LIKE " + dbName + ".unptned", driver);
    run("INSERT INTO TABLE " + dbName + ".unptned_late SELECT * FROM " + dbName + ".unptned", driver);
    verifySetup("SELECT * from " + dbName + ".unptned_late", unptnData, driver);

    // CM was enabled during setup, REPL_EVENT_DB_LISTENER_TTL should be used, set the other one to a low value
    MetastoreConf.setTimeVar(hconf, MetastoreConf.ConfVars.EVENT_DB_LISTENER_TTL, cleanerTtlSeconds, TimeUnit.SECONDS);
    MetastoreConf.setTimeVar(hconf, MetastoreConf.ConfVars.EVENT_DB_LISTENER_CLEAN_INTERVAL, cleanerIntervalSeconds, TimeUnit.SECONDS);
    DbNotificationListener.resetCleaner(hconf);

    //sleep to ensure correct conf(REPL_EVENT_DB_LISTENER_TTL) is used
    try {
      Thread.sleep(cleanerIntervalSeconds * 1000 * 10);
    } catch (InterruptedException e) {
      LOG.warn("Sleep unsuccesful", e);
    }

    //verify events get replicated
    Tuple incrDump = replDumpDb(dbName);
    loadAndVerify(replDbName, dbName, incrDump.lastReplId);
    verifyRun("SELECT * from " + replDbName + ".unptned", unptnData, driverMirror);
    verifyRun("SELECT * from " + replDbName + ".unptned_late", unptnData, driverMirror);


    // For next run, CM is enabled, set REPL_EVENT_DB_LISTENER_TTL to low value for events to get deleted
    MetastoreConf.setTimeVar(hconf, MetastoreConf.ConfVars.EVENT_DB_LISTENER_TTL, cleanerTtlSeconds * 60 * 60, TimeUnit.SECONDS);
    MetastoreConf.setTimeVar(hconf, REPL_EVENT_DB_LISTENER_TTL, cleanerTtlSeconds , TimeUnit.SECONDS);
    DbNotificationListener.resetCleaner(hconf);

    run("ALTER TABLE " + dbName + ".ptned ADD PARTITION (b=1)", driver);
    run("LOAD DATA LOCAL INPATH '" + ptnLocn1 + "' OVERWRITE INTO TABLE " + dbName
            + ".ptned PARTITION(b=1)", driver);
    verifySetup("SELECT a from " + dbName + ".ptned WHERE b=1", ptnData1, driver);
    run("LOAD DATA LOCAL INPATH '" + ptnLocn2 + "' OVERWRITE INTO TABLE " + dbName
            + ".ptned PARTITION(b=2)", driver);
    verifySetup("SELECT a from " + dbName + ".ptned WHERE b=2", ptnData2, driver);

    try {
      Thread.sleep(cleanerIntervalSeconds * 1000 * 10);
    } catch (InterruptedException e) {
      LOG.warn("Sleep unsuccesful", e);
    }

    incrDump = replDumpDb(dbName);

    // expected empty data because REPL_EVENT_DB_LISTENER_TTL should have been exceeded before dump
    loadAndVerify(replDbName, dbName, incrDump.lastReplId);
    verifyRun("SELECT a from " + replDbName + ".ptned WHERE b=1", empty, driverMirror);
    verifyRun("SELECT a from " + replDbName + ".ptned WHERE b=2", empty, driverMirror);

    // With CM disabled, EVENT_DB_LISTENER_TTL should be used.
    // First check with high ttl
    MetastoreConf.setBoolVar(hconf, MetastoreConf.ConfVars.REPLCMENABLED, false);
    MetastoreConf.setTimeVar(hconf, MetastoreConf.ConfVars.EVENT_DB_LISTENER_TTL, cleanerTtlSeconds  * 60 * 60, TimeUnit.SECONDS);
    MetastoreConf.setTimeVar(hconf, REPL_EVENT_DB_LISTENER_TTL, cleanerTtlSeconds, TimeUnit.SECONDS);
    DbNotificationListener.resetCleaner(hconf);

    run("CREATE TABLE " + dbName
            + ".ptned_late(a string) PARTITIONED BY (b int) STORED AS TEXTFILE", driver);
    run("INSERT INTO TABLE " + dbName + ".ptned_late PARTITION(b=1) SELECT a FROM " + dbName
            + ".ptned WHERE b=1", driver);
    verifySetup("SELECT a from " + dbName + ".ptned_late WHERE b=1", ptnData1, driver);


    //sleep to ensure correct conf(EVENT_DB_LISTENER_TTL) is used
    try {
      Thread.sleep(cleanerIntervalSeconds * 1000 * 10);
    } catch (InterruptedException e) {
      LOG.warn("Sleep unsuccesful", e);
    }

    //check replication success
    incrDump = replDumpDb(dbName);
    loadAndVerify(replDbName, dbName, incrDump.lastReplId);
    verifyRun("SELECT a from " + replDbName + ".ptned_late WHERE b=1", ptnData1, driverMirror);


    //With CM disabled, set a low ttl for events to get deleted
    MetastoreConf.setBoolVar(hconf, MetastoreConf.ConfVars.REPLCMENABLED, false);
    MetastoreConf.setTimeVar(hconf, MetastoreConf.ConfVars.EVENT_DB_LISTENER_TTL, cleanerTtlSeconds, TimeUnit.SECONDS);
    MetastoreConf.setTimeVar(hconf, REPL_EVENT_DB_LISTENER_TTL, cleanerTtlSeconds   * 60 * 60, TimeUnit.SECONDS);
    DbNotificationListener.resetCleaner(hconf);

    run("INSERT INTO TABLE " + dbName + ".ptned_late PARTITION(b=2) SELECT a FROM " + dbName
            + ".ptned WHERE b=2", driver);
    verifySetup("SELECT a from " + dbName + ".ptned_late WHERE b=2", ptnData2, driver);


    try {
      Thread.sleep(cleanerIntervalSeconds * 1000 * 10);
    } catch (InterruptedException e) {
      LOG.warn("Sleep unsuccesful", e);
    }

    //events should be deleted before dump
    incrDump = replDumpDb(dbName);
    loadAndVerify(replDbName, dbName, incrDump.lastReplId);
    verifyRun("SELECT a from " + replDbName + ".ptned_late WHERE b=2", empty, driverMirror);


    //restore original values
    MetastoreConf.setBoolVar(hconf, MetastoreConf.ConfVars.REPLCMENABLED, true);
    MetastoreConf.setTimeVar(hconf, MetastoreConf.ConfVars.EVENT_DB_LISTENER_TTL, 86400, TimeUnit.SECONDS);
    MetastoreConf.setTimeVar(hconf, REPL_EVENT_DB_LISTENER_TTL, 864000, TimeUnit.SECONDS);
    MetastoreConf.setTimeVar(hconf, MetastoreConf.ConfVars.EVENT_DB_LISTENER_CLEAN_INTERVAL, 7200, TimeUnit.SECONDS);
    DbNotificationListener.resetCleaner(hconf);
    verifySetupSteps = verifySetupOriginal;
  }
Tests basic operation - creates a db, with 4 tables, 2 ptned and 2 unptned.
Inserts data into one of the ptned tables, and one of the unptned tables,
and verifies that a REPL DUMP followed by a REPL LOAD is able to load it
appropriately. This tests bootstrap behaviour primarily.
itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenarios.java