@Test
  public void testBatchingOfIncrementalEvents() throws Throwable {
    final int REPL_MAX_LOAD_TASKS = 5;
    List<String> incrementalBatchConfigs = Arrays.asList(
            String.format("'%s'='%s'", HiveConf.ConfVars.REPL_BATCH_INCREMENTAL_EVENTS, "true"),
            String.format("'%s'='%d'", HiveConf.ConfVars.REPL_APPROX_MAX_LOAD_TASKS, REPL_MAX_LOAD_TASKS)
    );

    //bootstrap run, config should have no effect
    WarehouseInstance.Tuple bootstrapDump = primary.run("use " + primaryDbName)
            .run("create table t1 (id int)")
            .run("insert into table t1 values (1)")
            .dump(primaryDbName, incrementalBatchConfigs);

    FileSystem fs = new Path(bootstrapDump.dumpLocation).getFileSystem(conf);

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

    //incremental run
    WarehouseInstance.Tuple incrementalDump = primary.run("use " + primaryDbName)
            .run("insert into t1 values(2)")
            .run("insert into t1 values(3)")
            .run("insert into t1 values(4)")
            .run("insert into t1 values(5)")
            .dump(primaryDbName, incrementalBatchConfigs);

    Path dumpPath = new Path(incrementalDump.dumpLocation, ReplUtils.REPL_HIVE_BASE_DIR);
    Path ackLastEventID = new Path(dumpPath, ReplAck.EVENTS_DUMP.toString());
    EventsDumpMetadata eventsDumpMetadata = EventsDumpMetadata.deserialize(ackLastEventID, conf);
    assertTrue(eventsDumpMetadata.isEventsBatched());

    int eventsCountInAckFile = eventsDumpMetadata.getEventsDumpedCount(), expectedEventsCount = 0;
    String eventsBatchDirPrefix = ReplUtils.INC_EVENTS_BATCH.replaceAll("%d", "");

    List<FileStatus> batchFiles = Arrays.stream(fs.listStatus(dumpPath))
            .filter(fileStatus -> fileStatus.getPath().getName()
                    .startsWith(eventsBatchDirPrefix)).collect(Collectors.toList());


    for (FileStatus fileStatus : batchFiles) {
      int eventsPerBatch = fs.listStatus(fileStatus.getPath()).length;
      assertTrue(eventsPerBatch <= REPL_MAX_LOAD_TASKS);
      expectedEventsCount += eventsPerBatch;
    }
    assertEquals(eventsCountInAckFile, expectedEventsCount);

    // Repl Load should be agnostic of batch size and REPL_BATCH_INCREMENTAL_EVENTS config.
    // hence not passing incrementalBatchConfigs here.
    replica.load(replicatedDbName, primaryDbName)
            .run("use " + replicatedDbName)
            .run("select * from t1")
            .verifyResults(new String[]{"1", "2", "3", "4", "5"});

    assertTrue(fs.exists(new Path(dumpPath, LOAD_ACKNOWLEDGEMENT.toString())));

    ReplDumpWork.testDeletePreviousDumpMetaPath(true);
    //second round of incremental dump.
    incrementalDump = primary.run("use " + primaryDbName)
            .run("insert into t1 values(6)")
            .run("insert into t1 values(7)")
            .run("insert into t1 values(8)")
            .run("insert into t1 values(9)")
            .dump(primaryDbName, incrementalBatchConfigs);

    // simulate a failure in repl dump when batching was enabled.
    dumpPath = new Path(incrementalDump.dumpLocation, ReplUtils.REPL_HIVE_BASE_DIR);
    batchFiles = Arrays.stream(fs.listStatus(dumpPath))
            .filter(fileStatus -> fileStatus.getPath().getName()
                    .startsWith(eventsBatchDirPrefix))
            .sorted(new EventDumpDirComparator()).collect(Collectors.toList());


    FileStatus lastBatch = batchFiles.get(batchFiles.size() - 1);
    Path lastBatchPath = lastBatch.getPath();

    FileStatus[] eventsOfLastBatch = fs.listStatus(lastBatchPath);
    Arrays.sort(eventsOfLastBatch, new EventDumpDirComparator());
    Map<FileStatus, Long> modificationTimes = batchFiles.stream().filter(file -> !Objects.equals(lastBatch, file))
            .collect(Collectors.toMap(Function.identity(), FileStatus::getModificationTime));

    long lastReplId = Long.parseLong(eventsOfLastBatch[0].getPath().getName()) - 1;
    ackLastEventID = new Path(dumpPath, ReplAck.EVENTS_DUMP.toString());
    eventsDumpMetadata = EventsDumpMetadata.deserialize(ackLastEventID, conf);
    eventsDumpMetadata.setLastReplId(lastReplId);
    eventsDumpMetadata.setEventsDumpedCount(eventsDumpMetadata.getEventsDumpedCount() - (eventsOfLastBatch.length - 1));
    org.apache.hadoop.hive.ql.parse.repl.dump.Utils.writeOutput(eventsDumpMetadata.serialize(),
            ackLastEventID,
            primary.hiveConf);
    fs.delete(new Path(dumpPath, DUMP_ACKNOWLEDGEMENT.toString()), false);
    //delete all events of last batch except one.
    for (int idx = 1; idx < eventsOfLastBatch.length; idx++)
      fs.delete(eventsOfLastBatch[idx].getPath(), true);

    // when we try to resume a failed dump which was batched without setting REPL_BATCH_INCREMENTAL_EVENTS = true
    // in the next run dump should fail.
    primary.dumpFailure(primaryDbName);
    if (ReplUtils.failedWithNonRecoverableError(dumpPath, conf)) {
      fs.delete(new Path(dumpPath, ReplAck.NON_RECOVERABLE_MARKER.toString()), false);
    }

    WarehouseInstance.Tuple dumpAfterFailure = primary.dump(primaryDbName, incrementalBatchConfigs);
    //ensure dump did recover and dump location of new dump is same as the previous one.
    assertEquals(dumpAfterFailure.dumpLocation, incrementalDump.dumpLocation);

    List<FileStatus> filesAfterFailedDump = Arrays.stream(fs.listStatus(dumpPath))
            .filter(fileStatus -> fileStatus.getPath().getName()
                    .startsWith(eventsBatchDirPrefix))
            .sorted(new EventDumpDirComparator()).collect(Collectors.toList());

    //ensure all event files are dumped again.
    assertEquals(batchFiles, filesAfterFailedDump);

    //ensure last batch had events dumped and was indeed modified.
    assertNotEquals(lastBatch.getModificationTime(),
            filesAfterFailedDump.get(filesAfterFailedDump.size() - 1).getModificationTime());

    assertArrayEquals(fs.listStatus(lastBatchPath),
            fs.listStatus(filesAfterFailedDump.get(filesAfterFailedDump.size() - 1).getPath()));

    //ensure remaining batches were not modified.
    assertTrue(filesAfterFailedDump.stream()
            .filter(file -> !Objects.equals(file, lastBatch))
            .allMatch(file -> file.getModificationTime() == modificationTimes.get(file)));
    //ensure successful repl load.
    replica.load(replicatedDbName, primaryDbName)
            .run("use " + replicatedDbName)
            .run("select * from t1")
            .verifyResults(new String[]{"1", "2", "3", "4", "5", "6", "7", "8", "9"});

    ReplDumpWork.testDeletePreviousDumpMetaPath(false);
  }
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