Java 类org.apache.hadoop.hbase.HRegionLocation 实例源码

public static THRegionLocation regionLocationFromHBase(HRegionLocation hrl) {
  HRegionInfo hri = hrl.getRegionInfo();
  ServerName serverName = hrl.getServerName();

  THRegionInfo thRegionInfo = new THRegionInfo();
  THRegionLocation thRegionLocation = new THRegionLocation();
  TServerName tServerName = new TServerName();

  tServerName.setHostName(serverName.getHostname());
  tServerName.setPort(serverName.getPort());
  tServerName.setStartCode(serverName.getStartcode());

  thRegionInfo.setTableName(hri.getTable().getName());
  thRegionInfo.setEndKey(hri.getEndKey());
  thRegionInfo.setStartKey(hri.getStartKey());
  thRegionInfo.setOffline(hri.isOffline());
  thRegionInfo.setSplit(hri.isSplit());
  thRegionInfo.setReplicaId(hri.getReplicaId());

  thRegionLocation.setRegionInfo(thRegionInfo);
  thRegionLocation.setServerName(tServerName);

  return thRegionLocation;
}

/**
 * Returns the {@link ServerName} from catalog table {@link Result}
 * where the region is transitioning. It should be the same as
 * {@link HRegionInfo#getServerName(Result)} if the server is at OPEN state.
 * @param r Result to pull the transitioning server name from
 * @return A ServerName instance or {@link HRegionInfo#getServerName(Result)}
 * if necessary fields not found or empty.
 */
static ServerName getRegionServer(final Result r, int replicaId) {
  Cell cell = r.getColumnLatestCell(HConstants.CATALOG_FAMILY, getServerNameColumn(replicaId));
  if (cell == null || cell.getValueLength() == 0) {
    RegionLocations locations = MetaTableAccessor.getRegionLocations(r);
    if (locations != null) {
      HRegionLocation location = locations.getRegionLocation(replicaId);
      if (location != null) {
        return location.getServerName();
      }
    }
    return null;
  }
  return ServerName.parseServerName(Bytes.toString(cell.getValueArray(),
    cell.getValueOffset(), cell.getValueLength()));
}

@Test (timeout=180000)
public void testMove() throws Exception {
  List<HRegionLocation> regions;
  try (RegionLocator locator = systemUserConnection.getRegionLocator(TEST_TABLE)) {
    regions = locator.getAllRegionLocations();
  }
  HRegionLocation location = regions.get(0);
  final HRegionInfo hri = location.getRegionInfo();
  final ServerName server = location.getServerName();
  AccessTestAction action = new AccessTestAction() {
    @Override
    public Object run() throws Exception {
      ACCESS_CONTROLLER.preMove(ObserverContext.createAndPrepare(CP_ENV, null),
          hri, server, server);
      return null;
    }
  };

  verifyAllowed(action, SUPERUSER, USER_ADMIN, USER_OWNER, USER_GROUP_ADMIN);
  verifyDenied(action, USER_CREATE, USER_RW, USER_RO, USER_NONE, USER_GROUP_READ,
    USER_GROUP_WRITE, USER_GROUP_CREATE);
}

@Test (timeout=180000)
public void testAssign() throws Exception {
  List<HRegionLocation> regions;
  try (RegionLocator locator = systemUserConnection.getRegionLocator(TEST_TABLE)) {
    regions = locator.getAllRegionLocations();
  }
  HRegionLocation location = regions.get(0);
  final HRegionInfo hri = location.getRegionInfo();
  AccessTestAction action = new AccessTestAction() {
    @Override
    public Object run() throws Exception {
      ACCESS_CONTROLLER.preAssign(ObserverContext.createAndPrepare(CP_ENV, null), hri);
      return null;
    }
  };

  verifyAllowed(action, SUPERUSER, USER_ADMIN, USER_OWNER, USER_GROUP_ADMIN);
  verifyDenied(action, USER_CREATE, USER_RW, USER_RO, USER_NONE, USER_GROUP_READ,
    USER_GROUP_WRITE, USER_GROUP_CREATE);
}

@VisibleForTesting
LinkedBlockingQueue<PutStatus> getQueue(HRegionLocation addr) {
  FlushWorker worker = serverToFlushWorkerMap.get(addr);
  if (worker == null) {
    synchronized (this.serverToFlushWorkerMap) {
      worker = serverToFlushWorkerMap.get(addr);
      if (worker == null) {
        // Create the flush worker
        worker = new FlushWorker(workerConf, this.conn, addr, this, perRegionServerBufferQueueSize,
                pool, executor);
        this.serverToFlushWorkerMap.put(addr, worker);
        executor.scheduleAtFixedRate(worker, flushPeriod, flushPeriod, TimeUnit.MILLISECONDS);
      }
    }
  }
  return worker.getQueue();
}

@Test (timeout=300000)
public void testGetRegion() throws Exception {
  // We use actual HBaseAdmin instance instead of going via Admin interface in
  // here because makes use of an internal HBA method (TODO: Fix.).
  HBaseAdmin rawAdmin = new HBaseAdmin(TEST_UTIL.getConfiguration());

  final TableName tableName = TableName.valueOf("testGetRegion");
  LOG.info("Started " + tableName);
  HTable t = TEST_UTIL.createMultiRegionTable(tableName, HConstants.CATALOG_FAMILY);

  HRegionLocation regionLocation = t.getRegionLocation("mmm");
  HRegionInfo region = regionLocation.getRegionInfo();
  byte[] regionName = region.getRegionName();
  Pair<HRegionInfo, ServerName> pair = rawAdmin.getRegion(regionName);
  assertTrue(Bytes.equals(regionName, pair.getFirst().getRegionName()));
  pair = rawAdmin.getRegion(region.getEncodedNameAsBytes());
  assertTrue(Bytes.equals(regionName, pair.getFirst().getRegionName()));
}

/**
 * Delete a cached location, no matter what it is. Called when we were told to not use cache.
 * @param tableName tableName
 * @param row
 */
public void clearCache(final TableName tableName, final byte [] row, int replicaId) {
  ConcurrentMap<byte[], RegionLocations> tableLocations = getTableLocations(tableName);

  boolean removed = false;
  RegionLocations regionLocations = getCachedLocation(tableName, row);
  if (regionLocations != null) {
    HRegionLocation toBeRemoved = regionLocations.getRegionLocation(replicaId);
    RegionLocations updatedLocations = regionLocations.remove(replicaId);
    if (updatedLocations != regionLocations) {
      byte[] startKey = regionLocations.getRegionLocation().getRegionInfo().getStartKey();
      if (updatedLocations.isEmpty()) {
        removed = tableLocations.remove(startKey, regionLocations);
      } else {
        removed = tableLocations.replace(startKey, regionLocations, updatedLocations);
      }
    }

    if (removed && LOG.isTraceEnabled() && toBeRemoved != null) {
      LOG.trace("Removed " + toBeRemoved + " from cache");
    }
  }
}

@SuppressWarnings("deprecation")
private HConnection getMockedConnection(final Configuration conf)
throws IOException, ServiceException {
  HConnection c = Mockito.mock(HConnection.class);
  Mockito.when(c.getConfiguration()).thenReturn(conf);
  Mockito.doNothing().when(c).close();
  // Make it so we return a particular location when asked.
  final HRegionLocation loc = new HRegionLocation(HRegionInfo.FIRST_META_REGIONINFO,
      ServerName.valueOf("example.org", 1234, 0));
  Mockito.when(c.getRegionLocation((TableName) Mockito.any(),
      (byte[]) Mockito.any(), Mockito.anyBoolean())).
    thenReturn(loc);
  Mockito.when(c.locateRegion((TableName) Mockito.any(), (byte[]) Mockito.any())).
    thenReturn(loc);
  ClientProtos.ClientService.BlockingInterface hri =
    Mockito.mock(ClientProtos.ClientService.BlockingInterface.class);
  Mockito.when(hri.bulkLoadHFile((RpcController)Mockito.any(), (BulkLoadHFileRequest)Mockito.any())).
    thenThrow(new ServiceException(new IOException("injecting bulk load error")));
  Mockito.when(c.getClient(Mockito.any(ServerName.class))).
    thenReturn(hri);
  return c;
}

private void printLocations(Result r) {
  RegionLocations rl = null;
  if (r == null) {
    LOG.info("FAILED FOR null Result");
    return;
  }
  LOG.info("FAILED FOR " + resultToString(r) + " Stale " + r.isStale());
  if (r.getRow() == null) {
    return;
  }
  try {
    rl = ((ClusterConnection)connection).locateRegion(tableName, r.getRow(), true, true);
  } catch (IOException e) {
    LOG.warn("Couldn't get locations for row " + Bytes.toString(r.getRow()));
  }
  HRegionLocation locations[] = rl.getRegionLocations();
  for (HRegionLocation h : locations) {
    LOG.info("LOCATION " + h);
  }
}

private void verifyRegionResults(RegionLocator regionLocator,
    Map<byte[], String> results, String expected, byte[] row)
throws Exception {
  for (Map.Entry<byte [], String> e: results.entrySet()) {
    LOG.info("row=" + Bytes.toString(row) + ", expected=" + expected +
     ", result key=" + Bytes.toString(e.getKey()) +
     ", value=" + e.getValue());
  }
  HRegionLocation loc = regionLocator.getRegionLocation(row, true);
  byte[] region = loc.getRegionInfo().getRegionName();
  assertTrue("Results should contain region " +
    Bytes.toStringBinary(region) + " for row '" + Bytes.toStringBinary(row)+ "'",
    results.containsKey(region));
  assertEquals("Invalid result for row '"+Bytes.toStringBinary(row)+"'",
    expected, results.get(region));
}

private void addNewScanner(boolean isSync) {
  HRegionLocation regionLocation = null;
  synchronized (regionLocationQueue) {
    regionLocation = regionLocationQueue.poll();
  }
  if (regionLocation == null) return;
  runningGet.incrementAndGet();
  if (!isSync) {
    new Thread(new GetScannerRunnable(regionLocation)).start();
    return;
  }
  try {
    innerAddScanner(regionLocation);
  } catch (IOException e) {
    e.printStackTrace();
    abortException = e;
  }
}

private void innerAddScanner(HRegionLocation regionLocation) throws IOException {
  if (INIT_REGION_SIZE != getRegionNumber()) {
    throw new IOException(
        "region number changed from " + INIT_REGION_SIZE + " to " + getRegionNumber());
  }
  Scan newScan = new Scan(rawScan);
  if (regionLocation.getRegionInfo().getStartKey() != null)
    newScan.setStartRow(regionLocation.getRegionInfo().getStartKey());
  if (regionLocation.getRegionInfo().getEndKey() != null)
    newScan.setStopRow(regionLocation.getRegionInfo().getEndKey());
  newScan.setAttribute(IndexConstants.SCAN_WITH_INDEX, Bytes.toBytes("Hi"));
  newScan.setFilter(rangeList.toFilterList());
  newScan.setAttribute(IndexConstants.MAX_SCAN_SCALE, Bytes.toBytes(1.0f));
  ResultScanner scanner = table.getScanner(newScan);
  synchronized (scannerList) {
    scannerList.add(scanner);
  }
  runningGet.decrementAndGet();
}

private ResultScanner getNextScanner() throws IOException {
  if (INIT_REGION_SIZE != getRegionNumber()) {
    throw new IOException(
        "region number changed from " + INIT_REGION_SIZE + " to " + getRegionNumber());
  }
  if (regionLocationQueue.isEmpty()) return null;
  HRegionLocation regionLocation = regionLocationQueue.poll();
  Scan newScan = new Scan(rawScan);
  if (regionLocation.getRegionInfo().getStartKey() != null)
    newScan.setStartRow(regionLocation.getRegionInfo().getStartKey());
  if (regionLocation.getRegionInfo().getEndKey() != null)
    newScan.setStopRow(regionLocation.getRegionInfo().getEndKey());
  newScan.setAttribute(IndexConstants.SCAN_WITH_INDEX, Bytes.toBytes("Hi"));
  newScan.setFilter(rangeList.toFilterList());
  newScan.setAttribute(IndexConstants.MAX_SCAN_SCALE, Bytes.toBytes(1.0f));
  newScan.setId(rawScan.getId());
  newScan.setCacheBlocks(rawScan.getCacheBlocks());
  newScan.setCaching(rawScan.getCaching());
  return table.getScanner(newScan);
}

private void addNewScanner(boolean isSync) {
  HRegionLocation regionLocation = null;
  synchronized (regionLocationQueue) {
    regionLocation = regionLocationQueue.poll();
  }
  if (regionLocation == null) return;
  if (!isSync) {
    new Thread(new GetScannerRunnable(regionLocation)).start();
    return;
  }
  try {
    innerAddScanner(regionLocation);
  } catch (IOException e) {
    e.printStackTrace();
    abortException = e;
  }
}

public void clearCache(final HRegionLocation location) {
  if (location == null) {
    return;
  }
  TableName tableName = location.getRegionInfo().getTable();
  ConcurrentMap<byte[], RegionLocations> tableLocations = getTableLocations(tableName);
  RegionLocations regionLocations = tableLocations.get(location.getRegionInfo().getStartKey());
  if (regionLocations != null) {
    RegionLocations updatedLocations = regionLocations.remove(location);
    boolean removed = false;
    if (updatedLocations != regionLocations) {
      if (updatedLocations.isEmpty()) {
        removed = tableLocations.remove(location.getRegionInfo().getStartKey(), regionLocations);
      } else {
        removed = tableLocations.replace(location.getRegionInfo().getStartKey(), regionLocations, updatedLocations);
      }
      if (removed && LOG.isTraceEnabled()) {
        LOG.trace("Removed " + location + " from cache");
      }
    }
  }
}

protected static int lookupRegionIndex(List<HRegionLocation> list, byte[] key) {
  if (list.size() == 1 || Bytes.compareTo(key, list.get(0).getRegionInfo().getStartKey()) <= 0)
    return 0;
  if (Bytes.compareTo(key, list.get(list.size() - 1).getRegionInfo().getStartKey()) >= 0)
    return list.size() - 1;
  int l = 0, r = list.size() - 1;
  while (l < r) {
    int mid = (l + r) / 2;
    int cmp = Bytes.compareTo(key, list.get(mid).getRegionInfo().getStartKey());
    if (cmp == 0) {
      return mid;
    } else if (cmp > 0) {
      if (Bytes.compareTo(key, list.get(mid + 1).getRegionInfo().getStartKey()) < 0) return mid;
      l = mid + 1;
    } else {
      r = mid - 1;
    }
  }
  return l;
}

@Override
public List<HRegionLocation> locateRegions(final TableName tableName,
    final boolean useCache, final boolean offlined) throws IOException {
  NavigableMap<HRegionInfo, ServerName> regions = MetaScanner.allTableRegions(this, tableName);
  final List<HRegionLocation> locations = new ArrayList<HRegionLocation>();
  for (HRegionInfo regionInfo : regions.keySet()) {
    RegionLocations list = locateRegion(tableName, regionInfo.getStartKey(), useCache, true);
    if (list != null) {
      for (HRegionLocation loc : list.getRegionLocations()) {
        if (loc != null) {
          locations.add(loc);
        }
      }
    }
  }
  return locations;
}

private HRegionLocation getReplicaLocationOrFail(Action<Row> action) {
  // We are going to try get location once again. For each action, we'll do it once
  // from cache, because the previous calls in the loop might populate it.
  int replicaId = action.getReplicaId();
  RegionLocations locs = findAllLocationsOrFail(action, true);
  if (locs == null) return null; // manageError already called
  HRegionLocation loc = locs.getRegionLocation(replicaId);
  if (loc == null || loc.getServerName() == null) {
    locs = findAllLocationsOrFail(action, false);
    if (locs == null) return null; // manageError already called
    loc = locs.getRegionLocation(replicaId);
  }
  if (loc == null || loc.getServerName() == null) {
    manageLocationError(action, null);
    return null;
  }
  return loc;
}

private T updateStatsAndUnwrap(T result, RetryingCallable<T> callable) {
  // don't track stats about requests that aren't to regionservers
  if (!(callable instanceof RegionServerCallable)) {
    return result;
  }

  // mutli-server callables span multiple regions, so they don't have a location,
  // but they are region server callables, so we have to handle them when we process the
  // result in AsyncProcess#receiveMultiAction, not in here
  if (callable instanceof MultiServerCallable) {
    return result;
  }

  // update the stats for the single server callable
  RegionServerCallable<T> regionCallable = (RegionServerCallable) callable;
  HRegionLocation location = regionCallable.getLocation();
  return ResultStatsUtil.updateStats(result, stats, location);
}

private void init() {
  logger.debug("Getting region locations");
  TableName tableName = TableName.valueOf(hbaseScanSpec.getTableName());
  Connection conn = storagePlugin.getConnection();

  try (Admin admin = conn.getAdmin();
       RegionLocator locator = conn.getRegionLocator(tableName)) {
    this.hTableDesc = admin.getTableDescriptor(tableName);
    List<HRegionLocation> regionLocations = locator.getAllRegionLocations();
    statsCalculator = new TableStatsCalculator(conn, hbaseScanSpec, storagePlugin.getContext().getConfig(), storagePluginConfig);

    boolean foundStartRegion = false;
    regionsToScan = new TreeMap<>();
    for (HRegionLocation regionLocation : regionLocations) {
      HRegionInfo regionInfo = regionLocation.getRegionInfo();
      if (!foundStartRegion && hbaseScanSpec.getStartRow() != null && hbaseScanSpec.getStartRow().length != 0 && !regionInfo.containsRow(hbaseScanSpec.getStartRow())) {
        continue;
      }
      foundStartRegion = true;
      regionsToScan.put(regionInfo, regionLocation.getServerName());
      scanSizeInBytes += statsCalculator.getRegionSizeInBytes(regionInfo.getRegionName());
      if (hbaseScanSpec.getStopRow() != null && hbaseScanSpec.getStopRow().length != 0 && regionInfo.containsRow(hbaseScanSpec.getStopRow())) {
        break;
      }
    }
  } catch (IOException e) {
    throw new RuntimeException("Error getting region info for table: " + hbaseScanSpec.getTableName(), e);
  }
  verifyColumns();
}

@Override
public HRegionLocation getRegionLocation(byte[] bytes, boolean reload) throws IOException {
    for(HRegionLocation region : getRegions(reload)) {
        if (region.getRegionInfo().containsRow(bytes)) {
            return region;
        }
    }
    throw new IOException("Region not found for row: " + Bytes.toStringBinary(bytes));
}

@Override
public Pair<byte[][], byte[][]> getStartEndKeys() throws IOException {
    List<HRegionLocation> regions = getAllRegionLocations();
    byte[][] starts = new byte[regions.size()][];
    byte[][] ends = new byte[regions.size()][];
    int i = 0;
    for(HRegionLocation region : regions) {
        starts[i] = region.getRegionInfo().getStartKey();
        ends[i] = region.getRegionInfo().getEndKey();
        i++;
    }
    return Pair.newPair(starts, ends);
}

public static List<THRegionLocation> regionLocationsFromHBase(List<HRegionLocation> locations) {
  List<THRegionLocation> tlocations = new ArrayList<THRegionLocation>(locations.size());
  for (HRegionLocation hrl:locations) {
    tlocations.add(regionLocationFromHBase(hrl));
  }
  return tlocations;
}

public RegionReplicaReplayCallable(ClusterConnection connection,
    RpcControllerFactory rpcControllerFactory, TableName tableName,
    HRegionLocation location, HRegionInfo regionInfo, byte[] row,List<Entry> entries,
    AtomicLong skippedEntries) {
  super(connection, rpcControllerFactory, location, tableName, row, regionInfo.getReplicaId());
  this.entries = entries;
  this.skippedEntries = skippedEntries;
  this.initialEncodedRegionName = regionInfo.getEncodedNameAsBytes();
}

private void init(RegionLocator regionLocator, Admin admin)
    throws IOException {
  if (!enabled(admin.getConfiguration())) {
    LOG.info("Region size calculation disabled.");
    return;
  }

  LOG.info("Calculating region sizes for table \"" + regionLocator.getName() + "\".");

  //get regions for table
  List<HRegionLocation> tableRegionInfos = regionLocator.getAllRegionLocations();
  Set<byte[]> tableRegions = new TreeSet<byte[]>(Bytes.BYTES_COMPARATOR);
  for (HRegionLocation regionInfo : tableRegionInfos) {
    tableRegions.add(regionInfo.getRegionInfo().getRegionName());
  }

  ClusterStatus clusterStatus = admin.getClusterStatus();
  Collection<ServerName> servers = clusterStatus.getServers();
  final long megaByte = 1024L * 1024L;

  //iterate all cluster regions, filter regions from our table and compute their size
  for (ServerName serverName: servers) {
    ServerLoad serverLoad = clusterStatus.getLoad(serverName);

    for (RegionLoad regionLoad: serverLoad.getRegionsLoad().values()) {
      byte[] regionId = regionLoad.getName();

      if (tableRegions.contains(regionId)) {

        long regionSizeBytes = regionLoad.getStorefileSizeMB() * megaByte;
        sizeMap.put(regionId, regionSizeBytes);

        if (LOG.isDebugEnabled()) {
          LOG.debug("Region " + regionLoad.getNameAsString() + " has size " + regionSizeBytes);
        }
      }
    }
  }
  LOG.debug("Region sizes calculated");
}

/**
 * Record the location of the hbase:meta region as found in ZooKeeper.
 */
private boolean recordMetaRegion() throws IOException {
  RegionLocations rl = ((ClusterConnection)connection).locateRegion(TableName.META_TABLE_NAME,
      HConstants.EMPTY_START_ROW, false, false);
  if (rl == null) {
    errors.reportError(ERROR_CODE.NULL_META_REGION,
        "META region or some of its attributes are null.");
    return false;
  }
  for (HRegionLocation metaLocation : rl.getRegionLocations()) {
    // Check if Meta region is valid and existing
    if (metaLocation == null || metaLocation.getRegionInfo() == null ||
        metaLocation.getHostname() == null) {
      errors.reportError(ERROR_CODE.NULL_META_REGION,
          "META region or some of its attributes are null.");
      return false;
    }
    ServerName sn = metaLocation.getServerName();
    MetaEntry m = new MetaEntry(metaLocation.getRegionInfo(), sn, EnvironmentEdgeManager.currentTime());
    HbckInfo hbckInfo = regionInfoMap.get(metaLocation.getRegionInfo().getEncodedName());
    if (hbckInfo == null) {
      regionInfoMap.put(metaLocation.getRegionInfo().getEncodedName(), new HbckInfo(m));
    } else {
      hbckInfo.metaEntry = m;
    }
  }
  return true;
}

ReplayServerCallable(final HConnection connection, final TableName tableName,
    final HRegionLocation regionLoc, final HRegionInfo regionInfo,
    final List<Entry> entries) {
  super(connection, tableName, null);
  this.entries = entries;
  this.regionInfo = regionInfo;
  setLocation(regionLoc);
}

public HRegionLocation getLocation(boolean useCache) throws IOException {
  RegionLocations rl = getRegionLocations(connection, tableName, row, useCache, replicaId);
  if (rl == null) {
    throw new HBaseIOException(getExceptionMessage());
  }
  HRegionLocation location = rl.getRegionLocation(replicaId);
  if (location == null) {
    throw new HBaseIOException(getExceptionMessage());
  }

  return location;
}

@Test
public void testGetStartEndKeysWithRegionReplicas() throws IOException {
  HTableDescriptor htd = new HTableDescriptor(TableName.valueOf("testGetStartEndKeys"));
  HColumnDescriptor fam = new HColumnDescriptor(FAMILY);
  htd.addFamily(fam);
  byte[][] KEYS = HBaseTestingUtility.KEYS_FOR_HBA_CREATE_TABLE;
  HBaseAdmin admin = TEST_UTIL.getHBaseAdmin();
  admin.createTable(htd, KEYS);
  List<HRegionInfo> regions = admin.getTableRegions(htd.getTableName());

  HRegionLocator locator =
      (HRegionLocator) admin.getConnection().getRegionLocator(htd.getTableName());
  for (int regionReplication = 1; regionReplication < 4; regionReplication++) {
    List<RegionLocations> regionLocations = new ArrayList<RegionLocations>();

    // mock region locations coming from meta with multiple replicas
    for (HRegionInfo region : regions) {
      HRegionLocation[] arr = new HRegionLocation[regionReplication];
      for (int i = 0; i < arr.length; i++) {
        arr[i] = new HRegionLocation(RegionReplicaUtil.getRegionInfoForReplica(region, i), null);
      }
      regionLocations.add(new RegionLocations(arr));
    }

    Pair<byte[][], byte[][]> startEndKeys = locator.getStartEndKeys(regionLocations);

    assertEquals(KEYS.length + 1, startEndKeys.getFirst().length);

    for (int i = 0; i < KEYS.length + 1; i++) {
      byte[] startKey = i == 0 ? HConstants.EMPTY_START_ROW : KEYS[i - 1];
      byte[] endKey = i == KEYS.length ? HConstants.EMPTY_END_ROW : KEYS[i];
      assertArrayEquals(startKey, startEndKeys.getFirst()[i]);
      assertArrayEquals(endKey, startEndKeys.getSecond()[i]);
    }
  }
}

MultiServerCallable(final ClusterConnection connection, final TableName tableName,
    final ServerName location, RpcControllerFactory rpcFactory, final MultiAction<R> multi) {
  super(connection, tableName, null);
  this.multiAction = multi;
  // RegionServerCallable has HRegionLocation field, but this is a multi-region request.
  // Using region info from parent HRegionLocation would be a mistake for this class; so
  // we will store the server here, and throw if someone tries to obtain location/regioninfo.
  this.location = new HRegionLocation(null, location);
  this.cellBlock = isCellBlock();
  controller = rpcFactory.newController();
}

@Override
public RegionLocations locateRegion(TableName tableName,
    byte[] row, boolean useCache, boolean retry, int replicaId) throws IOException {
  int i = 0;
  for (HRegionLocation hr : hrl){
    if (Arrays.equals(row, hr.getRegionInfo().getStartKey())) {
      usedRegions[i] = true;
      return new RegionLocations(hr);
    }
    i++;
  }
  return null;
}

/**
 * Makes some table with given region names.
 * */
private RegionLocator mockRegionLocator(String... regionNames) throws IOException {
  RegionLocator mockedTable = Mockito.mock(RegionLocator.class);
  when(mockedTable.getName()).thenReturn(TableName.valueOf("sizeTestTable"));
  List<HRegionLocation> regionLocations = new ArrayList<>();
  when(mockedTable.getAllRegionLocations()).thenReturn(regionLocations);

  for (String regionName : regionNames) {
    HRegionInfo info = Mockito.mock(HRegionInfo.class);
    when(info.getRegionName()).thenReturn(regionName.getBytes());
    regionLocations.add(new HRegionLocation(info, null));//we are not interested in values
  }

  return mockedTable;
}

public static void blockUntilRegionIsInMeta(Connection conn, long timeout, HRegionInfo hri)
    throws IOException, InterruptedException {
  log("blocking until region is in META: " + hri.getRegionNameAsString());
  long start = System.currentTimeMillis();
  while (System.currentTimeMillis() - start < timeout) {
    HRegionLocation loc = MetaTableAccessor.getRegionLocation(conn, hri);
    if (loc != null && !loc.getRegionInfo().isOffline()) {
      log("found region in META: " + hri.getRegionNameAsString());
      break;
    }
    Threads.sleep(10);
  }
}

@Override
public List<HRegionLocation> getAllRegionLocations() throws IOException {
  NavigableMap<HRegionInfo, ServerName> locations =
      MetaScanner.allTableRegions(this.connection, getName());
  ArrayList<HRegionLocation> regions = new ArrayList<>(locations.size());
  for (Entry<HRegionInfo, ServerName> entry : locations.entrySet()) {
    regions.add(new HRegionLocation(entry.getKey(), entry.getValue()));
  }
  return regions;
}

private void innerAddScanner(HRegionLocation regionLocation) throws IOException {
  Scan newScan = new Scan(rawScan);
  if (regionLocation.getRegionInfo().getStartKey() != null)
    newScan.setStartRow(regionLocation.getRegionInfo().getStartKey());
  if (regionLocation.getRegionInfo().getEndKey() != null)
    newScan.setStopRow(regionLocation.getRegionInfo().getEndKey());
  newScan.setAttribute(IndexConstants.SCAN_WITH_INDEX, Bytes.toBytes("Hi"));
  ResultScanner scanner = table.getScanner(newScan);
  synchronized (scannerList) {
    scannerList.add(scanner);
  }
}

/**
 * count covering regions for [start, end], used in clustering index
 *
 * @param tableName
 * @param start
 * @param end
 * @return
 * @throws IOException
 */
protected static int countCoveringRegions(Connection conn, TableName tableName, byte[] start,
    byte[] end) throws IOException {
  RegionLocator locator = conn.getRegionLocator(tableName);
  List<HRegionLocation> list = locator.getAllRegionLocations();
  localTest(list);
  int left = start == null ? 0 : lookupRegionIndex(list, start);
  int right = end == null ? list.size() - 1 : lookupRegionIndex(list, end);
  return right - left + 1;
}

private void addReplicaActionsAgain(
    Action<Row> action, Map<ServerName, MultiAction<Row>> actionsByServer) {
  if (action.getReplicaId() == RegionReplicaUtil.DEFAULT_REPLICA_ID) {
    throw new AssertionError("Cannot have default replica here");
  }
  HRegionLocation loc = getReplicaLocationOrFail(action);
  if (loc == null) return;
  addAction(loc.getServerName(), loc.getRegionInfo().getRegionName(),
      action, actionsByServer, nonceGroup);
}

/**
 * Put a newly discovered HRegionLocation into the cache.
 * @param tableName The table name.
 * @param source the source of the new location
 * @param location the new location
 */
public void cacheLocation(final TableName tableName, final ServerName source,
    final HRegionLocation location) {
  assert source != null;
  byte [] startKey = location.getRegionInfo().getStartKey();
  ConcurrentMap<byte[], RegionLocations> tableLocations = getTableLocations(tableName);
  RegionLocations locations = new RegionLocations(new HRegionLocation[] {location}) ;
  RegionLocations oldLocations = tableLocations.putIfAbsent(startKey, locations);
  boolean isNewCacheEntry = (oldLocations == null);
  if (isNewCacheEntry) {
    if (LOG.isTraceEnabled()) {
      LOG.trace("Cached location: " + location);
    }
    addToCachedServers(locations);
    return;
  }

  // If the server in cache sends us a redirect, assume it's always valid.
  HRegionLocation oldLocation = oldLocations.getRegionLocation(
    location.getRegionInfo().getReplicaId());
  boolean force = oldLocation != null && oldLocation.getServerName() != null
      && oldLocation.getServerName().equals(source);

  // For redirect if the number is equal to previous
  // record, the most common case is that first the region was closed with seqNum, and then
  // opened with the same seqNum; hence we will ignore the redirect.
  // There are so many corner cases with various combinations of opens and closes that
  // an additional counter on top of seqNum would be necessary to handle them all.
  RegionLocations updatedLocations = oldLocations.updateLocation(location, false, force);
  if (oldLocations != updatedLocations) {
    boolean replaced = tableLocations.replace(startKey, oldLocations, updatedLocations);
    if (replaced && LOG.isTraceEnabled()) {
      LOG.trace("Changed cached location to: " + location);
    }
    addToCachedServers(updatedLocations);
  }
}

/**
 * Lists all of the regions currently in META.
 * @param conf configuration
 * @param connection to connect with
 * @param offlined True if we are to include offlined regions, false and we'll
 * leave out offlined regions from returned list.
 * @return List of all user-space regions.
 * @throws IOException
 */
@VisibleForTesting // And for hbck.
public static List<HRegionInfo> listAllRegions(Configuration conf, Connection connection,
    final boolean offlined)
throws IOException {
  final List<HRegionInfo> regions = new ArrayList<HRegionInfo>();
  MetaScannerVisitor visitor = new MetaScannerVisitorBase() {
      @Override
      public boolean processRow(Result result) throws IOException {
        if (result == null || result.isEmpty()) {
          return true;
        }

        RegionLocations locations = MetaTableAccessor.getRegionLocations(result);
        if (locations == null) return true;
        for (HRegionLocation loc : locations.getRegionLocations()) {
          if (loc != null) {
            HRegionInfo regionInfo = loc.getRegionInfo();
            // If region offline AND we are not to include offlined regions, return.
            if (regionInfo.isOffline() && !offlined) continue;
            regions.add(regionInfo);
          }
        }
        return true;
      }
  };
  metaScan(connection, visitor);
  return regions;
}

public ReplicaRegionServerCallable(int id, HRegionLocation location) {
  super(RpcRetryingCallerWithReadReplicas.this.cConnection,
      RpcRetryingCallerWithReadReplicas.this.tableName, get.getRow());
  this.id = id;
  this.location = location;
  this.controller = rpcControllerFactory.newController();
  controller.setPriority(tableName);
}