private void doIncrementalLoadTest(boolean shouldChangeRegions, boolean shouldKeepLocality,
    boolean putSortReducer, List<String> tableStr) throws Exception {
    util = new HBaseTestingUtil();
    Configuration conf = util.getConfiguration();
    conf.setBoolean(MultiTableHFileOutputFormat.LOCALITY_SENSITIVE_CONF_KEY, shouldKeepLocality);
    int hostCount = 1;
    int regionNum = 5;
    if (shouldKeepLocality) {
      // We should change host count higher than hdfs replica count when MiniHBaseCluster supports
      // explicit hostnames parameter just like MiniDFSCluster does.
      hostCount = 3;
      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);

    Map<String, Table> allTables = new HashMap<>(tableStr.size());
    List<HFileOutputFormat2.TableInfo> tableInfo = new ArrayList<>(tableStr.size());
    boolean writeMultipleTables = tableStr.size() > 1;
    for (String tableStrSingle : tableStr) {
      byte[][] splitKeys = generateRandomSplitKeys(regionNum - 1);
      TableName tableName = TableName.valueOf(tableStrSingle);
      Table table = util.createTable(tableName, FAMILIES, splitKeys);

      RegionLocator r = util.getConnection().getRegionLocator(tableName);
      assertEquals("Should start with empty table", 0, util.countRows(table));
      int numRegions = r.getStartKeys().length;
      assertEquals("Should make " + regionNum + " regions", numRegions, regionNum);

      allTables.put(tableStrSingle, table);
      tableInfo.add(new HFileOutputFormat2.TableInfo(table.getDescriptor(), r));
    }
    Path testDir = util.getDataTestDirOnTestFS("testLocalMRIncrementalLoad");
    // Generate the bulk load files
    runIncrementalPELoad(conf, tableInfo, testDir, putSortReducer);
    if (writeMultipleTables) {
      testDir = new Path(testDir, "default");
    }

    for (Table tableSingle : allTables.values()) {
      // This doesn't write into the table, just makes files
      assertEquals("HFOF should not touch actual table", 0, util.countRows(tableSingle));
    }
    int numTableDirs = 0;
    FileStatus[] fss = testDir.getFileSystem(conf).listStatus(testDir);
    for (FileStatus tf : fss) {
      Path tablePath = testDir;
      if (writeMultipleTables) {
        if (allTables.containsKey(tf.getPath().getName())) {
          ++numTableDirs;
          tablePath = tf.getPath();
        } else {
          continue;
        }
      }

      // Make sure that a directory was created for every CF
      int dir = 0;
      fss = tablePath.getFileSystem(conf).listStatus(tablePath);
      for (FileStatus f : fss) {
        for (byte[] family : FAMILIES) {
          if (Bytes.toString(family).equals(f.getPath().getName())) {
            ++dir;
          }
        }
      }
      assertEquals("Column family not found in FS.", FAMILIES.length, dir);
    }
    if (writeMultipleTables) {
      assertEquals("Dir for all input tables not created", numTableDirs, allTables.size());
    }

    Admin admin = util.getConnection().getAdmin();
    try {
      // handle the split case
      if (shouldChangeRegions) {
        Table chosenTable = allTables.values().iterator().next();
        // Choose a semi-random table if multiple tables are available
        LOG.info("Changing regions in table " + chosenTable.getName().getNameAsString());
        admin.disableTable(chosenTable.getName());
        util.waitUntilNoRegionsInTransition();

        util.deleteTable(chosenTable.getName());
        byte[][] newSplitKeys = generateRandomSplitKeys(14);
        Table table = util.createTable(chosenTable.getName(), FAMILIES, newSplitKeys);

        while (
          util.getConnection().getRegionLocator(chosenTable.getName()).getAllRegionLocations()
            .size() != 15 || !admin.isTableAvailable(table.getName())
        ) {
          Thread.sleep(200);
          LOG.info("Waiting for new region assignment to happen");
        }
      }

      // Perform the actual load
      for (HFileOutputFormat2.TableInfo singleTableInfo : tableInfo) {
        Path tableDir = testDir;
        String tableNameStr = singleTableInfo.getTableDescriptor().getTableName().getNameAsString();
        LOG.info("Running BulkLoadHFiles on table" + tableNameStr);
        if (writeMultipleTables) {
          tableDir = new Path(testDir, tableNameStr);
        }
        Table currentTable = allTables.get(tableNameStr);
        TableName currentTableName = currentTable.getName();
        BulkLoadHFiles.create(conf).bulkLoad(currentTableName, tableDir);

        // Ensure data shows up
        int expectedRows = 0;
        if (putSortReducer) {
          // no rows should be extracted
          assertEquals("BulkLoadHFiles should put expected data in table", expectedRows,
            util.countRows(currentTable));
        } else {
          expectedRows = NMapInputFormat.getNumMapTasks(conf) * ROWSPERSPLIT;
          assertEquals("BulkLoadHFiles should put expected data in table", expectedRows,
            util.countRows(currentTable));
          Scan scan = new Scan();
          ResultScanner results = currentTable.getScanner(scan);
          for (Result res : results) {
            assertEquals(FAMILIES.length, res.rawCells().length);
            Cell first = res.rawCells()[0];
            for (Cell kv : res.rawCells()) {
              assertTrue(CellUtil.matchingRows(first, kv));
              assertTrue(Bytes.equals(CellUtil.cloneValue(first), CellUtil.cloneValue(kv)));
            }
          }
          results.close();
        }
        String tableDigestBefore = util.checksumRows(currentTable);
        // Check region locality
        HDFSBlocksDistribution hbd = new HDFSBlocksDistribution();
        for (HRegion region : util.getHBaseCluster().getRegions(currentTableName)) {
          hbd.add(region.getHDFSBlocksDistribution());
        }
        for (String hostname : hostnames) {
          float locality = hbd.getBlockLocalityIndex(hostname);
          LOG.info("locality of [" + hostname + "]: " + locality);
          assertEquals(100, (int) (locality * 100));
        }

        // Cause regions to reopen
        admin.disableTable(currentTableName);
        while (!admin.isTableDisabled(currentTableName)) {
          Thread.sleep(200);
          LOG.info("Waiting for table to disable");
        }
        admin.enableTable(currentTableName);
        util.waitTableAvailable(currentTableName);
        assertEquals("Data should remain after reopening of regions", tableDigestBefore,
          util.checksumRows(currentTable));
      }
    } finally {
      for (HFileOutputFormat2.TableInfo tableInfoSingle : tableInfo) {
        tableInfoSingle.getRegionLocator().close();
      }
      for (Entry<String, Table> singleTable : allTables.entrySet()) {
        singleTable.getValue().close();
        util.deleteTable(singleTable.getValue().getName());
      }
      testDir.getFileSystem(conf).delete(testDir, true);
      util.shutdownMiniCluster();
    }
  }
Test for HFileOutputFormat2.LOCALITY_SENSITIVE_CONF_KEY = true This test could only check the
correctness of original logic if LOCALITY_SENSITIVE_CONF_KEY is set to true. Because
MiniHBaseCluster always run with single hostname (and different ports), it's not possible to
check the region locality by comparing region locations and DN hostnames. When MiniHBaseCluster
supports explicit hostnames parameter (just like MiniDFSCluster does), we could test region
locality features more easily.
hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestHFileOutputFormat2.java