@Test
  public void testIncrementalDumpCheckpointing() throws Throwable {
    WarehouseInstance.Tuple bootstrapDump = primary.run("use " + primaryDbName)
            .run("CREATE TABLE t1(a string) STORED AS TEXTFILE")
            .run("CREATE TABLE t2(a string) STORED AS TEXTFILE")
            .dump(primaryDbName);

    replica.load(replicatedDbName, primaryDbName)
            .run("select * from " + replicatedDbName + ".t1")
            .verifyResults(new String[] {})
            .run("select * from " + replicatedDbName + ".t2")
            .verifyResults(new String[] {});


    //Case 1: When the last dump finished all the events and
    //only  _finished_dump file at the hiveDumpRoot was about to be written when it failed.
    ReplDumpWork.testDeletePreviousDumpMetaPath(true);

    WarehouseInstance.Tuple incrementalDump1 = primary.run("use " + primaryDbName)
            .run("insert into t1 values (1)")
            .run("insert into t2 values (2)")
            .dump(primaryDbName);

    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());
    FileSystem fs = FileSystem.get(hiveDumpDir.toUri(), primary.hiveConf);
    assertTrue(fs.exists(ackFile));
    assertTrue(fs.exists(ackLastEventID));

    fs.delete(ackFile, false);

    long firstIncEventID = Long.parseLong(bootstrapDump.lastReplicationId) + 1;
    long lastIncEventID = Long.parseLong(incrementalDump1.lastReplicationId);
    assertTrue(lastIncEventID > (firstIncEventID + 1));
    Map<Path, Long> pathModTimeMap = new HashMap<>();
    for (long eventId=firstIncEventID; eventId<=lastIncEventID; eventId++) {
      Path eventRoot = new Path(hiveDumpDir, String.valueOf(eventId));
      if (fs.exists(eventRoot)) {
        for (FileStatus fileStatus: fs.listStatus(eventRoot)) {
          pathModTimeMap.put(fileStatus.getPath(), fileStatus.getModificationTime());
        }
      }
    }

    ReplDumpWork.testDeletePreviousDumpMetaPath(false);
    WarehouseInstance.Tuple incrementalDump2 = primary.run("use " + primaryDbName)
            .dump(primaryDbName);
    assertEquals(incrementalDump1.dumpLocation, incrementalDump2.dumpLocation);
    assertTrue(fs.exists(ackFile));
    //check events were not rewritten.
    for(Map.Entry<Path, Long> entry :pathModTimeMap.entrySet()) {
      assertEquals((long)entry.getValue(),
              fs.getFileStatus(new Path(hiveDumpDir, entry.getKey())).getModificationTime());
    }

    replica.load(replicatedDbName, primaryDbName)
            .run("select * from " + replicatedDbName + ".t1")
            .verifyResults(new String[] {"1"})
            .run("select * from " + replicatedDbName + ".t2")
            .verifyResults(new String[] {"2"});


    //Case 2: When the last dump was half way through
    ReplDumpWork.testDeletePreviousDumpMetaPath(true);

    WarehouseInstance.Tuple incrementalDump3 = primary.run("use " + primaryDbName)
            .run("insert into t1 values (3)")
            .run("insert into t2 values (4)")
            .dump(primaryDbName);

    hiveDumpDir = new Path(incrementalDump3.dumpLocation, ReplUtils.REPL_HIVE_BASE_DIR);
    ackFile = new Path(hiveDumpDir, ReplAck.DUMP_ACKNOWLEDGEMENT.toString());
    ackLastEventID = new Path(hiveDumpDir, ReplAck.EVENTS_DUMP.toString());
    fs = FileSystem.get(hiveDumpDir.toUri(), primary.hiveConf);
    assertTrue(fs.exists(ackFile));
    assertTrue(fs.exists(ackLastEventID));

    fs.delete(ackFile, false);
    //delete last three events and test if it recovers.
    long lastEventID = Long.parseLong(incrementalDump3.lastReplicationId);
    Path lastEvtRoot = new Path(hiveDumpDir + File.separator + String.valueOf(lastEventID));
    Path secondLastEvtRoot = new Path(hiveDumpDir + File.separator + String.valueOf(lastEventID - 1));
    Path thirdLastEvtRoot = new Path(hiveDumpDir + File.separator + String.valueOf(lastEventID - 2));
    assertTrue(fs.exists(lastEvtRoot));
    assertTrue(fs.exists(secondLastEvtRoot));
    assertTrue(fs.exists(thirdLastEvtRoot));

    pathModTimeMap = new HashMap<>();
    for (long idx = Long.parseLong(incrementalDump2.lastReplicationId)+1; idx < (lastEventID - 2); idx++) {
      Path eventRoot  = new Path(hiveDumpDir, String.valueOf(idx));
      if (fs.exists(eventRoot)) {
        for (FileStatus fileStatus: fs.listStatus(eventRoot)) {
          pathModTimeMap.put(fileStatus.getPath(), fileStatus.getModificationTime());
        }
      }
    }
    long lastEvtModTimeOld = fs.getFileStatus(lastEvtRoot).getModificationTime();
    long secondLastEvtModTimeOld = fs.getFileStatus(secondLastEvtRoot).getModificationTime();
    long thirdLastEvtModTimeOld = fs.getFileStatus(thirdLastEvtRoot).getModificationTime();

    fs.delete(lastEvtRoot, true);
    fs.delete(secondLastEvtRoot, true);
    fs.delete(thirdLastEvtRoot, true);
    EventsDumpMetadata eventsDumpMetadata = EventsDumpMetadata.deserialize(ackLastEventID, conf);
    eventsDumpMetadata.setLastReplId(lastEventID - 3);
    eventsDumpMetadata.setEventsDumpedCount(eventsDumpMetadata.getEventsDumpedCount() - 3);
    org.apache.hadoop.hive.ql.parse.repl.dump.Utils.writeOutput(eventsDumpMetadata.serialize(), ackLastEventID,
            primary.hiveConf);
    ReplDumpWork.testDeletePreviousDumpMetaPath(false);

    WarehouseInstance.Tuple incrementalDump4 = primary.run("use " + primaryDbName)
            .dump(primaryDbName);

    assertEquals(incrementalDump3.dumpLocation, incrementalDump4.dumpLocation);

    verifyPathExist(fs, ackFile);
    verifyPathExist(fs, ackLastEventID);
    verifyPathExist(fs, lastEvtRoot);
    verifyPathExist(fs, secondLastEvtRoot);
    verifyPathExist(fs, thirdLastEvtRoot);
    assertTrue(fs.getFileStatus(lastEvtRoot).getModificationTime() > lastEvtModTimeOld);
    assertTrue(fs.getFileStatus(secondLastEvtRoot).getModificationTime() > secondLastEvtModTimeOld);
    assertTrue(fs.getFileStatus(thirdLastEvtRoot).getModificationTime() > thirdLastEvtModTimeOld);

    //Check other event dump files have not been modified.
    for (Map.Entry<Path, Long> entry:pathModTimeMap.entrySet()) {
      assertEquals((long)entry.getValue(), fs.getFileStatus(entry.getKey()).getModificationTime());
    }

    replica.load(replicatedDbName, primaryDbName)
            .run("select * from " + replicatedDbName + ".t1")
            .verifyResults(new String[] {"1", "3"})
            .run("select * from " + replicatedDbName + ".t2")
            .verifyResults(new String[] {"2", "4"});
  }
Testcase for getting immutable dataset dump, and its corresponding repl load.
itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenariosAcidTables.java