Java 类org.apache.hadoop.hbase.client.RegionReplicaUtil 实例源码

/**
 * Return the replicas (including default) for the regions grouped by ServerName
 * @param regions
 * @return a pair containing the groupings as a map
 */
synchronized Map<ServerName, List<HRegionInfo>> getRegionAssignments(
  Collection<HRegionInfo> regions) {
  Map<ServerName, List<HRegionInfo>> map = new HashMap<ServerName, List<HRegionInfo>>();
  for (HRegionInfo region : regions) {
    HRegionInfo defaultReplica = RegionReplicaUtil.getRegionInfoForDefaultReplica(region);
    Set<HRegionInfo> allReplicas = defaultReplicaToOtherReplicas.get(defaultReplica);
    if (allReplicas != null) {
      for (HRegionInfo hri : allReplicas) {
        ServerName server = regionAssignments.get(hri);
        if (server != null) {
          List<HRegionInfo> regionsOnServer = map.get(server);
          if (regionsOnServer == null) {
            regionsOnServer = new ArrayList<HRegionInfo>(1);
            map.put(server, regionsOnServer);
          }
          regionsOnServer.add(hri);
        }
      }
    }
  }
  return map;
}

/**
 * Create any replicas for the regions (the default replicas that was
 * already created is passed to the method)
 * @param hTableDescriptor
 * @param regions default replicas
 * @return the combined list of default and non-default replicas
 */
protected List<HRegionInfo> addReplicas(HTableDescriptor hTableDescriptor,
    List<HRegionInfo> regions) {
  int numRegionReplicas = hTableDescriptor.getRegionReplication() - 1;
  if (numRegionReplicas <= 0) {
    return regions;
  }
  List<HRegionInfo> hRegionInfos =
      new ArrayList<HRegionInfo>((numRegionReplicas+1)*regions.size());
  for (int i = 0; i < regions.size(); i++) {
    for (int j = 1; j <= numRegionReplicas; j++) {
      hRegionInfos.add(RegionReplicaUtil.getRegionInfoForReplica(regions.get(i), j));
    }
  }
  hRegionInfos.addAll(regions);
  return hRegionInfos;
}

/**
 * Create any replicas for the regions (the default replicas that was
 * already created is passed to the method)
 * @param hTableDescriptor descriptor to use
 * @param regions default replicas
 * @return the combined list of default and non-default replicas
 */
private static List<HRegionInfo> addReplicas(final MasterProcedureEnv env,
    final HTableDescriptor hTableDescriptor,
    final List<HRegionInfo> regions) {
  int numRegionReplicas = hTableDescriptor.getRegionReplication() - 1;
  if (numRegionReplicas <= 0) {
    return regions;
  }
  List<HRegionInfo> hRegionInfos =
      new ArrayList<HRegionInfo>((numRegionReplicas+1)*regions.size());
  for (int i = 0; i < regions.size(); i++) {
    for (int j = 1; j <= numRegionReplicas; j++) {
      hRegionInfos.add(RegionReplicaUtil.getRegionInfoForReplica(regions.get(i), j));
    }
  }
  hRegionInfos.addAll(regions);
  return hRegionInfos;
}

/**
 * Get a list of replica regions that are:
 * not recorded in meta yet. We might not have recorded the locations
 * for the replicas since the replicas may not have been online yet, master restarted
 * in the middle of assigning, ZK erased, etc.
 * @param regionsRecordedInMeta the list of regions we know are recorded in meta
 * either as a default, or, as the location of a replica
 * @param master
 * @return list of replica regions
 * @throws IOException
 */
public static List<HRegionInfo> replicaRegionsNotRecordedInMeta(
    Set<HRegionInfo> regionsRecordedInMeta, MasterServices master)throws IOException {
  List<HRegionInfo> regionsNotRecordedInMeta = new ArrayList<HRegionInfo>();
  for (HRegionInfo hri : regionsRecordedInMeta) {
    TableName table = hri.getTable();
    HTableDescriptor htd = master.getTableDescriptors().get(table);
    // look at the HTD for the replica count. That's the source of truth
    int desiredRegionReplication = htd.getRegionReplication();
    for (int i = 0; i < desiredRegionReplication; i++) {
      HRegionInfo replica = RegionReplicaUtil.getRegionInfoForReplica(hri, i);
      if (regionsRecordedInMeta.contains(replica)) continue;
      regionsNotRecordedInMeta.add(replica);
    }
  }
  return regionsNotRecordedInMeta;
}

private void undeployRegions(HbckInfo hi) throws IOException, InterruptedException {
  undeployRegionsForHbi(hi);
  // undeploy replicas of the region (but only if the method is invoked for the primary)
  if (hi.getReplicaId() != HRegionInfo.DEFAULT_REPLICA_ID) {
    return;
  }
  int numReplicas = admin.getTableDescriptor(hi.getTableName()).getRegionReplication();
  for (int i = 1; i < numReplicas; i++) {
    if (hi.getPrimaryHRIForDeployedReplica() == null) continue;
    HRegionInfo hri = RegionReplicaUtil.getRegionInfoForReplica(
        hi.getPrimaryHRIForDeployedReplica(), i);
    HbckInfo h = regionInfoMap.get(hri.getEncodedName());
    if (h != null) {
      undeployRegionsForHbi(h);
      //set skip checks; we undeployed it, and we don't want to evaluate this anymore
      //in consistency checks
      h.setSkipChecks(true);
    }
  }
}

@Override public OperationStatus[] batchReplay(MutationReplay[] mutations, long replaySeqId)
    throws IOException {
  if (!RegionReplicaUtil.isDefaultReplica(getRegionInfo())
      && replaySeqId < lastReplayedOpenRegionSeqId) {
    // if it is a secondary replica we should ignore these entries silently
    // since they are coming out of order
    if (LOG.isTraceEnabled()) {
      LOG.trace(getRegionInfo().getEncodedName() + " : " + "Skipping " + mutations.length
          + " mutations with replaySeqId=" + replaySeqId
          + " which is < than lastReplayedOpenRegionSeqId=" + lastReplayedOpenRegionSeqId);
      for (MutationReplay mut : mutations) {
        LOG.trace(getRegionInfo().getEncodedName() + " : Skipping : " + mut.mutation);
      }
    }

    OperationStatus[] statuses = new OperationStatus[mutations.length];
    for (int i = 0; i < statuses.length; i++) {
      statuses[i] = OperationStatus.SUCCESS;
    }
    return statuses;
  }
  return batchMutate(new ReplayBatch(mutations, replaySeqId));
}

/**
 * Checks whether the given regionName is either equal to our region, or that the regionName is
 * the primary region to our corresponding range for the secondary replica.
 */
private void checkTargetRegion(byte[] encodedRegionName, String exceptionMsg, Object payload)
    throws WrongRegionException {
  if (Bytes.equals(this.getRegionInfo().getEncodedNameAsBytes(), encodedRegionName)) {
    return;
  }

  if (!RegionReplicaUtil.isDefaultReplica(this.getRegionInfo()) && Bytes
      .equals(encodedRegionName, this.fs.getRegionInfoForFS().getEncodedNameAsBytes())) {
    return;
  }

  throw new WrongRegionException(
      exceptionMsg + payload + " targetted for region " + Bytes.toStringBinary(encodedRegionName)
          + " does not match this region: " + this.getRegionInfo());
}

private Region getBiggestMemstoreOfRegionReplica(SortedMap<Long, Region> regionsBySize,
    Set<Region> excludedRegions) {
  synchronized (regionsInQueue) {
    for (Region region : regionsBySize.values()) {
      if (excludedRegions.contains(region)) {
        continue;
      }

      if (RegionReplicaUtil.isDefaultReplica(region.getRegionInfo())) {
        continue;
      }

      return region;
    }
  }
  return null;
}

/**
 * Adds a hbase:meta row for each of the specified new regions.
 * @param connection connection we're using
 * @param regionInfos region information list
 * @param regionReplication
 * @param ts desired timestamp
 * @throws IOException if problem connecting or updating meta
 */
public static void addRegionsToMeta(Connection connection,
    List<HRegionInfo> regionInfos, int regionReplication, long ts)
        throws IOException {
  List<Put> puts = new ArrayList<Put>();
  for (HRegionInfo regionInfo : regionInfos) {
    if (RegionReplicaUtil.isDefaultReplica(regionInfo)) {
      Put put = makePutFromRegionInfo(regionInfo, ts);
      // Add empty locations for region replicas so that number of replicas can be cached
      // whenever the primary region is looked up from meta
      for (int i = 1; i < regionReplication; i++) {
        addEmptyLocation(put, i);
      }
      puts.add(put);
    }
  }
  putsToMetaTable(connection, puts);
  LOG.info("Added " + puts.size());
}

/**
 * Return the replicas (including default) for the regions grouped by ServerName
 * @param regions
 * @return a pair containing the groupings as a map
 */
synchronized Map<ServerName, List<HRegionInfo>> getRegionAssignments(
  Collection<HRegionInfo> regions) {
  Map<ServerName, List<HRegionInfo>> map = new HashMap<ServerName, List<HRegionInfo>>();
  for (HRegionInfo region : regions) {
    HRegionInfo defaultReplica = RegionReplicaUtil.getRegionInfoForDefaultReplica(region);
    Set<HRegionInfo> allReplicas = defaultReplicaToOtherReplicas.get(defaultReplica);
    if (allReplicas != null) {
      for (HRegionInfo hri : allReplicas) {
        ServerName server = regionAssignments.get(hri);
        if (server != null) {
          List<HRegionInfo> regionsOnServer = map.get(server);
          if (regionsOnServer == null) {
            regionsOnServer = new ArrayList<HRegionInfo>(1);
            map.put(server, regionsOnServer);
          }
          regionsOnServer.add(hri);
        }
      }
    }
  }
  return map;
}

/**
 * Create any replicas for the regions (the default replicas that was
 * already created is passed to the method)
 * @param hTableDescriptor
 * @param regions default replicas
 * @return the combined list of default and non-default replicas
 */
protected List<HRegionInfo> addReplicas(HTableDescriptor hTableDescriptor,
    List<HRegionInfo> regions) {
  int numRegionReplicas = hTableDescriptor.getRegionReplication() - 1;
  if (numRegionReplicas <= 0) {
    return regions;
  }
  List<HRegionInfo> hRegionInfos =
      new ArrayList<HRegionInfo>((numRegionReplicas+1)*regions.size());
  for (int i = 0; i < regions.size(); i++) {
    for (int j = 1; j <= numRegionReplicas; j++) {
      hRegionInfos.add(RegionReplicaUtil.getRegionInfoForReplica(regions.get(i), j));
    }
  }
  hRegionInfos.addAll(regions);
  return hRegionInfos;
}

/**
 * Get a list of replica regions that are:
 * not recorded in meta yet. We might not have recorded the locations
 * for the replicas since the replicas may not have been online yet, master restarted
 * in the middle of assigning, ZK erased, etc.
 * @param regionsRecordedInMeta the list of regions we know are recorded in meta
 * either as a default, or, as the location of a replica
 * @param master
 * @return list of replica regions
 * @throws IOException
 */
public static List<HRegionInfo> replicaRegionsNotRecordedInMeta(
    Set<HRegionInfo> regionsRecordedInMeta, MasterServices master)throws IOException {
  List<HRegionInfo> regionsNotRecordedInMeta = new ArrayList<HRegionInfo>();
  for (HRegionInfo hri : regionsRecordedInMeta) {
    TableName table = hri.getTable();
    HTableDescriptor htd = master.getTableDescriptors().get(table);
    // look at the HTD for the replica count. That's the source of truth
    int desiredRegionReplication = htd.getRegionReplication();
    for (int i = 0; i < desiredRegionReplication; i++) {
      HRegionInfo replica = RegionReplicaUtil.getRegionInfoForReplica(hri, i);
      if (regionsRecordedInMeta.contains(replica)) continue;
      regionsNotRecordedInMeta.add(replica);
    }
  }
  return regionsNotRecordedInMeta;
}

/**
 * Rollback close regions
 * @param env MasterProcedureEnv
 **/
private void rollbackCloseRegionsForMerge(final MasterProcedureEnv env) throws IOException {
  // Check whether the region is closed; if so, open it in the same server
  final int regionReplication = getRegionReplication(env);
  final ServerName serverName = getServerName(env);

  final AssignProcedure[] procs =
      new AssignProcedure[regionsToMerge.length * regionReplication];
  int procsIdx = 0;
  for (int i = 0; i < regionsToMerge.length; ++i) {
    for (int j = 0; j < regionReplication; ++j) {
      final RegionInfo hri = RegionReplicaUtil.getRegionInfoForReplica(regionsToMerge[i], j);
      procs[procsIdx++] = env.getAssignmentManager().createAssignProcedure(hri, serverName);
    }
  }
  env.getMasterServices().getMasterProcedureExecutor().submitProcedures(procs);
}

private void undeployRegions(HbckInfo hi) throws IOException, InterruptedException {
  undeployRegionsForHbi(hi);
  // undeploy replicas of the region (but only if the method is invoked for the primary)
  if (hi.getReplicaId() != RegionInfo.DEFAULT_REPLICA_ID) {
    return;
  }
  int numReplicas = admin.getTableDescriptor(hi.getTableName()).getRegionReplication();
  for (int i = 1; i < numReplicas; i++) {
    if (hi.getPrimaryHRIForDeployedReplica() == null) continue;
    RegionInfo hri = RegionReplicaUtil.getRegionInfoForReplica(
        hi.getPrimaryHRIForDeployedReplica(), i);
    HbckInfo h = regionInfoMap.get(hri.getEncodedName());
    if (h != null) {
      undeployRegionsForHbi(h);
      //set skip checks; we undeployed it, and we don't want to evaluate this anymore
      //in consistency checks
      h.setSkipChecks(true);
    }
  }
}

public OperationStatus[] batchReplay(MutationReplay[] mutations, long replaySeqId)
    throws IOException {
  if (!RegionReplicaUtil.isDefaultReplica(getRegionInfo())
      && replaySeqId < lastReplayedOpenRegionSeqId) {
    // if it is a secondary replica we should ignore these entries silently
    // since they are coming out of order
    if (LOG.isTraceEnabled()) {
      LOG.trace(getRegionInfo().getEncodedName() + " : "
        + "Skipping " + mutations.length + " mutations with replaySeqId=" + replaySeqId
        + " which is < than lastReplayedOpenRegionSeqId=" + lastReplayedOpenRegionSeqId);
      for (MutationReplay mut : mutations) {
        LOG.trace(getRegionInfo().getEncodedName() + " : Skipping : " + mut.mutation);
      }
    }

    OperationStatus[] statuses = new OperationStatus[mutations.length];
    for (int i = 0; i < statuses.length; i++) {
      statuses[i] = OperationStatus.SUCCESS;
    }
    return statuses;
  }
  return batchMutate(new ReplayBatchOperation(this, mutations, replaySeqId));
}

/** Checks whether the given regionName is either equal to our region, or that
 * the regionName is the primary region to our corresponding range for the secondary replica.
 */
private void checkTargetRegion(byte[] encodedRegionName, String exceptionMsg, Object payload)
    throws WrongRegionException {
  if (Bytes.equals(this.getRegionInfo().getEncodedNameAsBytes(), encodedRegionName)) {
    return;
  }

  if (!RegionReplicaUtil.isDefaultReplica(this.getRegionInfo()) &&
      Bytes.equals(encodedRegionName,
        this.fs.getRegionInfoForFS().getEncodedNameAsBytes())) {
    return;
  }

  throw new WrongRegionException(exceptionMsg + payload
    + " targetted for region " + Bytes.toStringBinary(encodedRegionName)
    + " does not match this region: " + this.getRegionInfo());
}

private HRegion getBiggestMemStoreOfRegionReplica(SortedMap<Long, HRegion> regionsBySize,
    Set<HRegion> excludedRegions) {
  synchronized (regionsInQueue) {
    for (HRegion region : regionsBySize.values()) {
      if (excludedRegions.contains(region)) {
        continue;
      }

      if (RegionReplicaUtil.isDefaultReplica(region.getRegionInfo())) {
        continue;
      }

      return region;
    }
  }
  return null;
}

private boolean checkDuplicates(Collection<HRegion> onlineRegions3) throws Exception {
  ArrayList<Region> copyOfRegion = new ArrayList<Region>(onlineRegions3);
  for (Region region : copyOfRegion) {
    RegionInfo regionInfo = region.getRegionInfo();
    RegionInfo regionInfoForReplica =
        RegionReplicaUtil.getRegionInfoForDefaultReplica(regionInfo);
    int i = 0;
    for (Region actualRegion : onlineRegions3) {
      if (regionInfoForReplica.equals(
        RegionReplicaUtil.getRegionInfoForDefaultReplica(actualRegion.getRegionInfo()))) {
        i++;
        if (i > 1) {
          LOG.info("Duplicate found " + actualRegion.getRegionInfo() + " " +
              region.getRegionInfo());
          assertTrue(Bytes.equals(region.getRegionInfo().getStartKey(),
            actualRegion.getRegionInfo().getStartKey()));
          assertTrue(Bytes.equals(region.getRegionInfo().getEndKey(),
            actualRegion.getRegionInfo().getEndKey()));
          return true;
        }
      }
    }
  }
  return false;
}

/**
 * Adds a hbase:meta row for each of the specified new regions. Initial state for new regions
 * is CLOSED.
 * @param connection connection we're using
 * @param regionInfos region information list
 * @param regionReplication
 * @param ts desired timestamp
 * @throws IOException if problem connecting or updating meta
 */
public static void addRegionsToMeta(Connection connection, List<RegionInfo> regionInfos,
    int regionReplication, long ts) throws IOException {
  List<Put> puts = new ArrayList<>();
  for (RegionInfo regionInfo : regionInfos) {
    if (RegionReplicaUtil.isDefaultReplica(regionInfo)) {
      Put put = makePutFromRegionInfo(regionInfo, ts);
      // New regions are added with initial state of CLOSED.
      addRegionStateToPut(put, RegionState.State.CLOSED);
      // Add empty locations for region replicas so that number of replicas can be cached
      // whenever the primary region is looked up from meta
      for (int i = 1; i < regionReplication; i++) {
        addEmptyLocation(put, i);
      }
      puts.add(put);
    }
  }
  putsToMetaTable(connection, puts);
  LOG.info("Added {} regions to meta.", puts.size());
}

private void addToReplicaMapping(HRegionInfo hri) {
  HRegionInfo defaultReplica = RegionReplicaUtil.getRegionInfoForDefaultReplica(hri);
  Set<HRegionInfo> replicas =
      defaultReplicaToOtherReplicas.get(defaultReplica);
  if (replicas == null) {
    replicas = new HashSet<HRegionInfo>();
    defaultReplicaToOtherReplicas.put(defaultReplica, replicas);
  }
  replicas.add(hri);
}

private void removeFromReplicaMapping(HRegionInfo hri) {
  HRegionInfo defaultReplica = RegionReplicaUtil.getRegionInfoForDefaultReplica(hri);
  Set<HRegionInfo> replicas = defaultReplicaToOtherReplicas.get(defaultReplica);
  if (replicas != null) {
    replicas.remove(hri);
    if (replicas.isEmpty()) {
      defaultReplicaToOtherReplicas.remove(defaultReplica);
    }
  }
}

private void doMergingOfReplicas(HRegionInfo mergedHri, final HRegionInfo hri_a,
    final HRegionInfo hri_b) {
  // Close replicas for the original unmerged regions. create/assign new replicas
  // for the merged parent.
  List<HRegionInfo> unmergedRegions = new ArrayList<HRegionInfo>();
  unmergedRegions.add(hri_a);
  unmergedRegions.add(hri_b);
  Map<ServerName, List<HRegionInfo>> map = regionStates.getRegionAssignments(unmergedRegions);
  Collection<List<HRegionInfo>> c = map.values();
  for (List<HRegionInfo> l : c) {
    for (HRegionInfo h : l) {
      if (!RegionReplicaUtil.isDefaultReplica(h)) {
        LOG.debug("Unassigning un-merged replica " + h);
        unassign(h);
      }
    }
  }
  int numReplicas = 1;
  try {
    numReplicas = server.getTableDescriptors().get(mergedHri.getTable()).
        getRegionReplication();
  } catch (IOException e) {
    LOG.warn("Couldn't get the replication attribute of the table " + mergedHri.getTable() +
        " due to " + e.getMessage() + ". The assignment of replicas for the merged region " +
        "will not be done");
  }
  List<HRegionInfo> regions = new ArrayList<HRegionInfo>();
  for (int i = 1; i < numReplicas; i++) {
    regions.add(RegionReplicaUtil.getRegionInfoForReplica(mergedHri, i));
  }
  try {
    assign(regions);
  } catch (IOException ioe) {
    LOG.warn("Couldn't assign all replica(s) of region " + mergedHri + " because of " +
              ioe.getMessage());
  } catch (InterruptedException ie) {
    LOG.warn("Couldn't assign all replica(s) of region " + mergedHri+ " because of " +
              ie.getMessage());
  }
}

private void prepareDaughterReplicaForAssignment(HRegionInfo daughterHri, HRegionInfo parentHri,
    int replicaId, Map<HRegionInfo, ServerName> map) {
  HRegionInfo parentReplica = RegionReplicaUtil.getRegionInfoForReplica(parentHri, replicaId);
  HRegionInfo daughterReplica = RegionReplicaUtil.getRegionInfoForReplica(daughterHri,
      replicaId);
  LOG.debug("Created replica region for daughter " + daughterReplica);
  ServerName sn;
  if ((sn = regionStates.getRegionServerOfRegion(parentReplica)) != null) {
    map.put(daughterReplica, sn);
  } else {
    List<ServerName> servers = serverManager.getOnlineServersList();
    sn = servers.get((new Random(System.currentTimeMillis())).nextInt(servers.size()));
    map.put(daughterReplica, sn);
  }
}

private void tryAssignmentRepair(HbckInfo hbi, String msg) throws IOException,
  KeeperException, InterruptedException {
  // If we are trying to fix the errors
  if (shouldFixAssignments()) {
    errors.print(msg);
    undeployRegions(hbi);
    setShouldRerun();
    HRegionInfo hri = hbi.getHdfsHRI();
    if (hri == null) {
      hri = hbi.metaEntry;
    }
    HBaseFsckRepair.fixUnassigned(admin, hri);
    HBaseFsckRepair.waitUntilAssigned(admin, hri);

    // also assign replicas if needed (do it only when this call operates on a primary replica)
    if (hbi.getReplicaId() != HRegionInfo.DEFAULT_REPLICA_ID) return;
    int replicationCount = admin.getTableDescriptor(hri.getTable()).getRegionReplication();
    for (int i = 1; i < replicationCount; i++) {
      hri = RegionReplicaUtil.getRegionInfoForReplica(hri, i);
      HbckInfo h = regionInfoMap.get(hri.getEncodedName());
      if (h != null) {
        undeployRegions(h);
        //set skip checks; we undeploy & deploy it; we don't want to evaluate this hbi anymore
        //in consistency checks
        h.setSkipChecks(true);
      }
      HBaseFsckRepair.fixUnassigned(admin, hri);
      HBaseFsckRepair.waitUntilAssigned(admin, hri);
    }

  }
}

private void assignMetaReplica(int replicaId)
    throws IOException, KeeperException, InterruptedException {
  errors.reportError(ERROR_CODE.NO_META_REGION, "hbase:meta, replicaId " +
      replicaId +" is not found on any region.");
  if (shouldFixAssignments()) {
    errors.print("Trying to fix a problem with hbase:meta..");
    setShouldRerun();
    // try to fix it (treat it as unassigned region)
    HRegionInfo h = RegionReplicaUtil.getRegionInfoForReplica(
        HRegionInfo.FIRST_META_REGIONINFO, replicaId);
    HBaseFsckRepair.fixUnassigned(admin, h);
    HBaseFsckRepair.waitUntilAssigned(admin, h);
  }
}

public synchronized void addServer(HRegionInfo hri, ServerName server) {
  OnlineEntry rse = new OnlineEntry() ;
  rse.hri = hri;
  rse.hsa = server;
  this.deployedEntries.add(rse);
  this.deployedOn.add(server);
  // save the replicaId that we see deployed in the cluster
  this.deployedReplicaId = hri.getReplicaId();
  this.primaryHRIForDeployedReplica =
      RegionReplicaUtil.getRegionInfoForDefaultReplica(hri);
}

/**
 * Returns the regionInfo object to use for interacting with the file system.
 * @return An HRegionInfo object to interact with the filesystem
 */
public static HRegionInfo getRegionInfoForFs(HRegionInfo regionInfo) {
  if (regionInfo == null) {
    return null;
  }
  return RegionReplicaUtil.getRegionInfoForDefaultReplica(regionInfo);
}

@Test
public void testNeedsBalanceForColocatedReplicas() {
  // check for the case where there are two hosts and with one rack, and where
  // both the replicas are hosted on the same server
  List<HRegionInfo> regions = randomRegions(1);
  ServerName s1 = ServerName.valueOf("host1", 1000, 11111);
  ServerName s2 = ServerName.valueOf("host11", 1000, 11111);
  Map<ServerName, List<HRegionInfo>> map = new HashMap<ServerName, List<HRegionInfo>>();
  map.put(s1, regions);
  regions.add(RegionReplicaUtil.getRegionInfoForReplica(regions.get(0), 1));
  // until the step above s1 holds two replicas of a region
  regions = randomRegions(1);
  map.put(s2, regions);
  assertTrue(loadBalancer.needsBalance(new Cluster(map, null, null, null)));
  // check for the case where there are two hosts on the same rack and there are two racks
  // and both the replicas are on the same rack
  map.clear();
  regions = randomRegions(1);
  List<HRegionInfo> regionsOnS2 = new ArrayList<HRegionInfo>(1);
  regionsOnS2.add(RegionReplicaUtil.getRegionInfoForReplica(regions.get(0), 1));
  map.put(s1, regions);
  map.put(s2, regionsOnS2);
  // add another server so that the cluster has some host on another rack
  map.put(ServerName.valueOf("host2", 1000, 11111), randomRegions(1));
  assertTrue(loadBalancer.needsBalance(new Cluster(map, null, null,
      new ForTestRackManagerOne())));
}

public static void verifyReplicasCameOnline(TableName tableName, Admin admin,
    int regionReplication) throws IOException {
  List<HRegionInfo> regions = admin.getTableRegions(tableName);
  HashSet<HRegionInfo> set = new HashSet<HRegionInfo>();
  for (HRegionInfo hri : regions) {
    set.add(RegionReplicaUtil.getRegionInfoForDefaultReplica(hri));
    for (int i = 0; i < regionReplication; i++) {
      HRegionInfo replica = RegionReplicaUtil.getRegionInfoForReplica(hri, i);
      if (!regions.contains(replica)) {
        Assert.fail(replica + " is not contained in the list of online regions");
      }
    }
  }
  assert(set.size() == getSplitKeys().length + 1);
}

/**
 * 
 * @param zkw
 * @param replicaId
 * @return meta table regions and their locations.
 */
public List<Pair<HRegionInfo, ServerName>> getMetaRegionsAndLocations(ZooKeeperWatcher zkw,
    int replicaId) {
  ServerName serverName = getMetaRegionLocation(zkw, replicaId);
  List<Pair<HRegionInfo, ServerName>> list = new ArrayList<Pair<HRegionInfo, ServerName>>();
  list.add(new Pair<HRegionInfo, ServerName>(RegionReplicaUtil.getRegionInfoForReplica(
      HRegionInfo.FIRST_META_REGIONINFO, replicaId), serverName));
  return list;
}

/**
 * Convert a RegionInfo to a HRegionInfo
 *
 * @param proto the RegionInfo to convert
 * @return the converted HRegionInfho
 */
public static HRegionInfo convert(final RegionInfo proto) {
  if (proto == null) return null;
  TableName tableName =
      ProtobufUtil.toTableName(proto.getTableName());
  if (tableName.equals(TableName.META_TABLE_NAME)) {
    return RegionReplicaUtil.getRegionInfoForReplica(FIRST_META_REGIONINFO,
        proto.getReplicaId());
  }
  long regionId = proto.getRegionId();
  int replicaId = proto.hasReplicaId() ? proto.getReplicaId() : DEFAULT_REPLICA_ID;
  byte[] startKey = null;
  byte[] endKey = null;
  if (proto.hasStartKey()) {
    startKey = proto.getStartKey().toByteArray();
  }
  if (proto.hasEndKey()) {
    endKey = proto.getEndKey().toByteArray();
  }
  boolean split = false;
  if (proto.hasSplit()) {
    split = proto.getSplit();
  }
  HRegionInfo hri = new HRegionInfo(
      tableName,
      startKey,
      endKey, split, regionId, replicaId);
  if (proto.hasOffline()) {
    hri.setOffline(proto.getOffline());
  }
  return hri;
}

private void addToReplicaMapping(HRegionInfo hri) {
  HRegionInfo defaultReplica = RegionReplicaUtil.getRegionInfoForDefaultReplica(hri);
  Set<HRegionInfo> replicas =
      defaultReplicaToOtherReplicas.get(defaultReplica);
  if (replicas == null) {
    replicas = new HashSet<HRegionInfo>();
    defaultReplicaToOtherReplicas.put(defaultReplica, replicas);
  }
  replicas.add(hri);
}

private void removeFromReplicaMapping(HRegionInfo hri) {
  HRegionInfo defaultReplica = RegionReplicaUtil.getRegionInfoForDefaultReplica(hri);
  Set<HRegionInfo> replicas = defaultReplicaToOtherReplicas.get(defaultReplica);
  if (replicas != null) {
    replicas.remove(hri);
    if (replicas.isEmpty()) {
      defaultReplicaToOtherReplicas.remove(defaultReplica);
    }
  }
}

@Override
public synchronized Void call() throws IOException {
  errors.progress();
  try {
    BlockingInterface server = connection.getAdmin(rsinfo);

    // list all online regions from this region server
    List<HRegionInfo> regions = ProtobufUtil.getOnlineRegions(server);
    regions = filterRegions(regions);

    if (details) {
      errors.detail("RegionServer: " + rsinfo.getServerName() +
                       " number of regions: " + regions.size());
      for (HRegionInfo rinfo: regions) {
        errors.detail("  " + rinfo.getRegionNameAsString() +
                         " id: " + rinfo.getRegionId() +
                         " encoded_name: " + rinfo.getEncodedName() +
                         " start: " + Bytes.toStringBinary(rinfo.getStartKey()) +
                         " end: " + Bytes.toStringBinary(rinfo.getEndKey()));
      }
    }

    // check to see if the existence of this region matches the region in META
    for (HRegionInfo r:regions) {
      HbckInfo hbi = hbck.getOrCreateInfo(r.getEncodedName());
      if (!RegionReplicaUtil.isDefaultReplica(r)) hbi.setSkipChecks(true);
      hbi.addServer(r, rsinfo);
    }
  } catch (IOException e) {          // unable to connect to the region server.
    errors.reportError(ERROR_CODE.RS_CONNECT_FAILURE, "RegionServer: " + rsinfo.getServerName() +
      " Unable to fetch region information. " + e);
    throw e;
  }
  return null;
}

/**
 * Returns the regionInfo object to use for interacting with the file system.
 * @return An HRegionInfo object to interact with the filesystem
 */
public static HRegionInfo getRegionInfoForFs(HRegionInfo regionInfo) {
  if (regionInfo == null) {
    return null;
  }
  return RegionReplicaUtil.getRegionInfoForDefaultReplica(regionInfo);
}

@Test
public void testNeedsBalanceForColocatedReplicas() {
  // check for the case where there are two hosts and with one rack, and where
  // both the replicas are hosted on the same server
  List<HRegionInfo> regions = randomRegions(1);
  ServerName s1 = ServerName.valueOf("host1", 1000, 11111);
  ServerName s2 = ServerName.valueOf("host11", 1000, 11111);
  Map<ServerName, List<HRegionInfo>> map = new HashMap<ServerName, List<HRegionInfo>>();
  map.put(s1, regions);
  regions.add(RegionReplicaUtil.getRegionInfoForReplica(regions.get(0), 1));
  // until the step above s1 holds two replicas of a region
  regions = randomRegions(1);
  map.put(s2, regions);
  assertTrue(loadBalancer.needsBalance(new Cluster(map, null, null, null)));
  // check for the case where there are two hosts on the same rack and there are two racks
  // and both the replicas are on the same rack
  map.clear();
  regions = randomRegions(1);
  List<HRegionInfo> regionsOnS2 = new ArrayList<HRegionInfo>(1);
  regionsOnS2.add(RegionReplicaUtil.getRegionInfoForReplica(regions.get(0), 1));
  map.put(s1, regions);
  map.put(s2, regionsOnS2);
  // add another server so that the cluster has some host on another rack
  map.put(ServerName.valueOf("host2", 1000, 11111), randomRegions(1));
  assertTrue(loadBalancer.needsBalance(new Cluster(map, null, null,
      new ForTestRackManagerOne())));
}

public static void verifyReplicasCameOnline(TableName tableName, Admin admin,
    int regionReplication) throws IOException {
  List<HRegionInfo> regions = admin.getTableRegions(tableName);
  HashSet<HRegionInfo> set = new HashSet<HRegionInfo>();
  for (HRegionInfo hri : regions) {
    set.add(RegionReplicaUtil.getRegionInfoForDefaultReplica(hri));
    for (int i = 0; i < regionReplication; i++) {
      HRegionInfo replica = RegionReplicaUtil.getRegionInfoForReplica(hri, i);
      if (!regions.contains(replica)) {
        Assert.fail(replica + " is not contained in the list of online regions");
      }
    }
  }
  assert(set.size() == getSplitKeys().length + 1);
}

/**
 * Adds a hbase:meta row for each of the specified new regions.
 * @param connection connection we're using
 * @param regionInfos region information list
 * @throws IOException if problem connecting or updating meta
 */
public static void addRegionsToMeta(Connection connection,
                                    List<HRegionInfo> regionInfos)
        throws IOException {
    List<Put> puts = new ArrayList<Put>();
    for (HRegionInfo regionInfo : regionInfos) {
        if (RegionReplicaUtil.isDefaultReplica(regionInfo)) {
            puts.add(makePutFromRegionInfo(regionInfo));
        }
    }
    putsToMetaTable(connection, puts);
    LOG.info("Added " + puts.size());
}

/**
 * @author Wang Xiaoyi
 * Adds a hbase:meta row for each of the specified new regions.
 * @param connection connection we're using
 * @param regionInfos region information list
 * @param schema schema of table this region belongs to
 * @throws IOException
 */
public static void addRegionsToMeta(Connection connection,
                                    List<HRegionInfo> regionInfos, String schema)
        throws IOException{
    List<Put> puts = new ArrayList<Put>();
    for (HRegionInfo regionInfo : regionInfos) {
        if (RegionReplicaUtil.isDefaultReplica(regionInfo)) {
            puts.add(makePutFromRegionInfo(regionInfo, schema));
        }
    }
    putsToMetaTable(connection, puts);
    LOG.info("Added " + puts.size());
}

private UnassignProcedure[] createUnassignProcedures(final MasterProcedureEnv env,
    final int regionReplication) {
  final UnassignProcedure[] procs =
      new UnassignProcedure[regionsToMerge.length * regionReplication];
  int procsIdx = 0;
  for (int i = 0; i < regionsToMerge.length; ++i) {
    for (int j = 0; j < regionReplication; ++j) {
      final RegionInfo hri = RegionReplicaUtil.getRegionInfoForReplica(regionsToMerge[i], j);
      procs[procsIdx++] = env.getAssignmentManager().createUnassignProcedure(hri,null,true);
    }
  }
  return procs;
}