public void testReplCommitTransactionOnSourceDelete(String tableStorage, String tableProperty) throws Throwable {
    String tableName = "testReplCommitTransactionOnSourceDelete";
    String[] result = new String[] { "5" };

    // Do a bootstrap dump.
    WarehouseInstance.Tuple bootStrapDump = primary.dump(primaryDbName);
    replica.load(replicatedDbName, primaryDbName).run("REPL STATUS " + replicatedDbName)
        .verifyResult(bootStrapDump.lastReplicationId);

    // Add some data to the table & do a incremental dump.
    ReplicationTestUtils.insertIntoDB(primary, primaryDbName, tableName, tableProperty, tableStorage,
        new String[] { "1", "2", "3", "4", "5" });
    WarehouseInstance.Tuple incrementalDump = primary.dump(primaryDbName);

    // Keep a copy of the data, before we drop the table, so that we can copy it back to the location, in order to
    // trigger source delete at the time of checksum verification.
    Path tablePath = new Path(primary.getTable(primaryDbName, tableName).getSd().getLocation());
    Path tablePath_dupe = new Path(primary.getTable(primaryDbName, tableName).getSd().getLocation() + "_dupe");
    FileSystem fs = tablePath.getFileSystem(conf);
    DataCopyStatistics copyStatistics = new DataCopyStatistics();
    FileUtils.copy(fs, tablePath, fs, tablePath_dupe, false, false, conf, copyStatistics);

    // Drop the table.
    primary.run("drop table " + primaryDbName + "." + tableName);

    // Copy back the data to original location, so that copy happens from original location, not the CM location.
    copyStatistics = new DataCopyStatistics();
    FileUtils.copy(fs, tablePath_dupe, fs, tablePath, false, false, conf, copyStatistics);

    // Add a util to delete the original source at the time of source checksum verification.
    CopyUtils.testCallable = () -> {
      try {
        fs.delete(tablePath, true);
      } catch (Throwable throwable) {
        throwable.printStackTrace();
      }
      return null;
    };

    // Do an incremental load & verify if things are good.
    replica.loadWithoutExplain(replicatedDbName, primaryDbName)
        .run("REPL STATUS " + replicatedDbName).verifyResult(incrementalDump.lastReplicationId);
    verifyResultsInReplicaInt(Lists.newArrayList("select count(*) from " + tableName,
        "select count(*) from " + tableName + "_nopart"),
        Lists.newArrayList(result, result));

  }
TestReplicationScenariosAcidTables - test replication for ACID tables
itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/parse/TestReplicationScenariosIncrementalLoadAcidTables.java