@Test
  public void testMRIncrementalLoadWithLocalityMultiCluster() throws Exception {
    // Start cluster A
    util = new HBaseTestingUtil();
    Configuration confA = util.getConfiguration();
    int hostCount = 3;
    int regionNum = 20;
    String[] hostnames = new String[hostCount];
    for (int i = 0; i < hostCount; ++i) {
      hostnames[i] = "datanode_" + i;
    }
    StartTestingClusterOption option = StartTestingClusterOption.builder()
      .numRegionServers(hostCount).dataNodeHosts(hostnames).build();
    util.startMiniCluster(option);

    // Start cluster B
    HBaseTestingUtil utilB = new HBaseTestingUtil();
    Configuration confB = utilB.getConfiguration();
    utilB.startMiniCluster(option);

    Path testDir = util.getDataTestDirOnTestFS("testLocalMRIncrementalLoad");

    byte[][] splitKeys = generateRandomSplitKeys(regionNum - 1);
    TableName tableName = TableName.valueOf("table");
    // Create table in cluster B
    try (Table table = utilB.createTable(tableName, FAMILIES, splitKeys);
      RegionLocator r = utilB.getConnection().getRegionLocator(tableName)) {
      // Generate the bulk load files
      // Job has zookeeper configuration for cluster A
      // Assume reading from cluster A by TableInputFormat and creating hfiles to cluster B
      Job job = new Job(confA, "testLocalMRIncrementalLoad");
      Configuration jobConf = job.getConfiguration();
      final UUID key = ConfigurationCaptorConnection.configureConnectionImpl(jobConf);
      job.setWorkingDirectory(util.getDataTestDirOnTestFS("runIncrementalPELoad"));
      setupRandomGeneratorMapper(job, false);
      HFileOutputFormat2.configureIncrementalLoad(job, table, r);

      assertEquals(confB.get(HConstants.ZOOKEEPER_QUORUM),
        jobConf.get(HFileOutputFormat2.REMOTE_CLUSTER_ZOOKEEPER_QUORUM_CONF_KEY));
      assertEquals(confB.get(HConstants.ZOOKEEPER_CLIENT_PORT),
        jobConf.get(HFileOutputFormat2.REMOTE_CLUSTER_ZOOKEEPER_CLIENT_PORT_CONF_KEY));
      assertEquals(confB.get(HConstants.ZOOKEEPER_ZNODE_PARENT),
        jobConf.get(HFileOutputFormat2.REMOTE_CLUSTER_ZOOKEEPER_ZNODE_PARENT_CONF_KEY));

      String bSpecificConfigKey = "my.override.config.for.b";
      String bSpecificConfigValue = "b-specific-value";
      jobConf.set(HFileOutputFormat2.REMOTE_CLUSTER_CONF_PREFIX + bSpecificConfigKey,
        bSpecificConfigValue);

      FileOutputFormat.setOutputPath(job, testDir);

      assertFalse(util.getTestFileSystem().exists(testDir));

      assertTrue(job.waitForCompletion(true));

      final List<Configuration> configs =
        ConfigurationCaptorConnection.getCapturedConfigarutions(key);

      assertFalse(configs.isEmpty());
      for (Configuration config : configs) {
        assertEquals(confB.get(HConstants.ZOOKEEPER_QUORUM),
          config.get(HConstants.ZOOKEEPER_QUORUM));
        assertEquals(confB.get(HConstants.ZOOKEEPER_CLIENT_PORT),
          config.get(HConstants.ZOOKEEPER_CLIENT_PORT));
        assertEquals(confB.get(HConstants.ZOOKEEPER_ZNODE_PARENT),
          config.get(HConstants.ZOOKEEPER_ZNODE_PARENT));

        assertEquals(bSpecificConfigValue, config.get(bSpecificConfigKey));
      }
    } finally {
      utilB.deleteTable(tableName);
      testDir.getFileSystem(confA).delete(testDir, true);
      util.shutdownMiniCluster();
      utilB.shutdownMiniCluster();
    }
  }
This test is to test the scenario happened in HBASE-6901. All files are bulk loaded and
excluded from minor compaction. Without the fix of HBASE-6901, an
ArrayIndexOutOfBoundsException will be thrown.
hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestHFileOutputFormat2.java