@Test
  public void externalTableIncrementalCheckpointing() throws Throwable {
    List<String> withClause = ReplicationTestUtils.includeExternalTableClause(true);
    WarehouseInstance.Tuple tuple = primary
            .run("use " + primaryDbName)
            .run("create external table t1 (id int)")
            .run("insert into table t1 values (1)")
            .run("insert into table t1 values (2)")
            .run("create external table t2 (id int)")
            .run("insert into table t2 values (3)")
            .run("insert into table t2 values (4)")
            .dump(primaryDbName, withClause);

    ReplicationTestUtils.assertExternalFileList(Arrays.asList(new String[]{"t1", "t2"}), tuple.dumpLocation, primary);

    replica.load(replicatedDbName, primaryDbName)
            .status(replicatedDbName)
            .verifyResult(tuple.lastReplicationId)
            .run("use " + replicatedDbName)
            .run("select * from t1")
            .verifyResults(new String[] {"1", "2"})
            .run("select * from t2")
            .verifyResults(new String[] {"3", "4"})
            .verifyReplTargetProperty(replicatedDbName);

    ReplDumpWork.testDeletePreviousDumpMetaPath(true);

    withClause = ReplicationTestUtils.externalTableWithClause(new ArrayList<>(), true, true);
    WarehouseInstance.Tuple incrementalDump1 = primary.run("use " + primaryDbName)
            .run("drop table t1")
            .run("insert into table t2 values (5)")
            .run("insert into table t2 values (6)")
            .run("create external table t3 (id int)")
            .run("insert into table t3 values (8)")
            .dump(primaryDbName, withClause);

    // verify that the external table list is written correctly for incremental
    ReplicationTestUtils.assertExternalFileList(Arrays.asList("t2", "t3"), incrementalDump1.dumpLocation, primary);

    FileSystem fs = primary.miniDFSCluster.getFileSystem();
    Path hiveDumpDir = new Path(incrementalDump1.dumpLocation, ReplUtils.REPL_HIVE_BASE_DIR);
    Path ackFile = new Path(hiveDumpDir, ReplAck.DUMP_ACKNOWLEDGEMENT.toString());
    Path ackLastEventID = new Path(hiveDumpDir, ReplAck.EVENTS_DUMP.toString());
    assertTrue(fs.exists(ackFile));
    assertTrue(fs.exists(ackLastEventID));
    Path bootstrapDir = new Path(hiveDumpDir, ReplUtils.INC_BOOTSTRAP_ROOT_DIR_NAME);
    Path metaDir = new Path(bootstrapDir, EximUtil.METADATA_PATH_NAME);
    Path dataDir = new Path(bootstrapDir, EximUtil.DATA_PATH_NAME);
    assertFalse(fs.exists(dataDir));
    long oldMetadirModTime = fs.getFileStatus(metaDir).getModificationTime();
    fs.delete(ackFile, false);
    EventsDumpMetadata eventsDumpMetadata = EventsDumpMetadata.deserialize(ackLastEventID, conf);
    fs.delete(ackLastEventID, false);
    //delete all the event folders except first event
    long startEvent = -1;
    long endEvent = Long.valueOf(incrementalDump1.lastReplicationId);
    int deletedEventsCount = 0;
    for (long eventDir = Long.valueOf(tuple.lastReplicationId) + 1;  eventDir <= endEvent; eventDir++) {
      Path eventRoot = new Path(hiveDumpDir, String.valueOf(eventDir));
      if (fs.exists(eventRoot)) {
        if (startEvent == -1){
          startEvent = eventDir;
        } else {
          deletedEventsCount++;
          fs.delete(eventRoot, true);
        }
      }
    }
    Path startEventRoot = new Path(hiveDumpDir, String.valueOf(startEvent));
    Map<Path, Long> firstEventModTimeMap = new HashMap<>();
    for (FileStatus fileStatus: fs.listStatus(startEventRoot)) {
      firstEventModTimeMap.put(fileStatus.getPath(), fileStatus.getModificationTime());
    }
    assertTrue(endEvent - startEvent > 1);
    eventsDumpMetadata.setEventsDumpedCount(eventsDumpMetadata.getEventsDumpedCount() - deletedEventsCount);
    Utils.writeOutput(eventsDumpMetadata.serialize(), ackLastEventID, primary.hiveConf);
    WarehouseInstance.Tuple incrementalDump2 = primary.dump(primaryDbName, withClause);
    assertEquals(incrementalDump1.dumpLocation, incrementalDump2.dumpLocation);
    assertTrue(fs.getFileStatus(metaDir).getModificationTime() > oldMetadirModTime);
    ReplicationTestUtils.assertExternalFileList(Arrays.asList("t2", "t3"), incrementalDump2.dumpLocation, primary);
    //first event meta is not rewritten
    for (Map.Entry<Path, Long> entry: firstEventModTimeMap.entrySet()) {
      assertEquals((long)entry.getValue(), fs.getFileStatus(entry.getKey()).getModificationTime());
    }
    replica.load(replicatedDbName, primaryDbName)
            .status(replicatedDbName)
            .verifyResult(incrementalDump2.lastReplicationId)
            .run("use " + replicatedDbName)
            .run("show tables like 't1'")
            .verifyFailure(new String[] {"t1"})
            .run("select * from t2")
            .verifyResults(new String[] {"3", "4", "5", "6"})
            .run("select * from t3")
            .verifyResult("8")
            .verifyReplTargetProperty(replicatedDbName);
  }
@param sourceTableName  -- Provide the fully qualified table name
@param replicaTableName -- Provide the fully qualified table name
itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenariosExternalTables.java