Java 类org.apache.hadoop.hbase.util.Pair 实例源码

/**
 * Create a protocol buffer bulk load request
 *
 * @param familyPaths
 * @param regionName
 * @param assignSeqNum
 * @return a bulk load request
 */
public static BulkLoadHFileRequest buildBulkLoadHFileRequest(
    final List<Pair<byte[], String>> familyPaths,
    final byte[] regionName, boolean assignSeqNum) {
  BulkLoadHFileRequest.Builder builder = BulkLoadHFileRequest.newBuilder();
  RegionSpecifier region = buildRegionSpecifier(
    RegionSpecifierType.REGION_NAME, regionName);
  builder.setRegion(region);
  FamilyPath.Builder familyPathBuilder = FamilyPath.newBuilder();
  for (Pair<byte[], String> familyPath: familyPaths) {
    familyPathBuilder.setFamily(ByteStringer.wrap(familyPath.getFirst()));
    familyPathBuilder.setPath(familyPath.getSecond());
    builder.addFamilyPath(familyPathBuilder.build());
  }
  builder.setAssignSeqNum(assignSeqNum);
  return builder.build();
}

/**
 * Modify a table, synchronous. Waiting logic similar to that of {@code admin.rb#alter_status}.
 */
@SuppressWarnings("serial")
public static void modifyTableSync(Admin admin, HTableDescriptor desc)
    throws IOException, InterruptedException {
  admin.modifyTable(desc.getTableName(), desc);
  Pair<Integer, Integer> status = new Pair<Integer, Integer>() {{
    setFirst(0);
    setSecond(0);
  }};
  int i = 0;
  do {
    status = admin.getAlterStatus(desc.getTableName());
    if (status.getSecond() != 0) {
      LOG.debug(status.getSecond() - status.getFirst() + "/" + status.getSecond()
        + " regions updated.");
      Thread.sleep(1 * 1000l);
    } else {
      LOG.debug("All regions updated.");
      break;
    }
  } while (status.getFirst() != 0 && i++ < 500);
  if (status.getFirst() != 0) {
    throw new IOException("Failed to update all regions even after 500 seconds.");
  }
}

protected Pair<Map<String, Integer>, Map<String, List<Integer>>> extractLabelsAndAuths(
    List<List<Cell>> labelDetails) {
  Map<String, Integer> labels = new HashMap<String, Integer>();
  Map<String, List<Integer>> userAuths = new HashMap<String, List<Integer>>();
  for (List<Cell> cells : labelDetails) {
    for (Cell cell : cells) {
      if (Bytes.equals(cell.getQualifierArray(), cell.getQualifierOffset(),
          cell.getQualifierLength(), LABEL_QUALIFIER, 0, LABEL_QUALIFIER.length)) {
        labels.put(
            Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()),
            Bytes.toInt(cell.getRowArray(), cell.getRowOffset()));
      } else {
        // These are user cells who has authorization for this label
        String user = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(),
            cell.getQualifierLength());
        List<Integer> auths = userAuths.get(user);
        if (auths == null) {
          auths = new ArrayList<Integer>();
          userAuths.put(user, auths);
        }
        auths.add(Bytes.toInt(cell.getRowArray(), cell.getRowOffset()));
      }
    }
  }
  return new Pair<Map<String, Integer>, Map<String, List<Integer>>>(labels, userAuths);
}

/**
 * Get the corresponding start keys and regions for an arbitrary range of
 * keys.
 * <p>
 * @param startKey Starting row in range, inclusive
 * @param endKey Ending row in range
 * @param includeEndKey true if endRow is inclusive, false if exclusive
 * @param reload true to reload information or false to use cached information
 * @return A pair of list of start keys and list of HRegionLocations that
 *         contain the specified range
 * @throws IOException if a remote or network exception occurs
 * @deprecated This is no longer a public API
 */
@Deprecated
private Pair<List<byte[]>, List<HRegionLocation>> getKeysAndRegionsInRange(
    final byte[] startKey, final byte[] endKey, final boolean includeEndKey,
    final boolean reload) throws IOException {
  final boolean endKeyIsEndOfTable = Bytes.equals(endKey,HConstants.EMPTY_END_ROW);
  if ((Bytes.compareTo(startKey, endKey) > 0) && !endKeyIsEndOfTable) {
    throw new IllegalArgumentException(
      "Invalid range: " + Bytes.toStringBinary(startKey) +
      " > " + Bytes.toStringBinary(endKey));
  }
  List<byte[]> keysInRange = new ArrayList<byte[]>();
  List<HRegionLocation> regionsInRange = new ArrayList<HRegionLocation>();
  byte[] currentKey = startKey;
  do {
    HRegionLocation regionLocation = getRegionLocation(currentKey, reload);
    keysInRange.add(currentKey);
    regionsInRange.add(regionLocation);
    currentKey = regionLocation.getRegionInfo().getEndKey();
  } while (!Bytes.equals(currentKey, HConstants.EMPTY_END_ROW)
      && (endKeyIsEndOfTable || Bytes.compareTo(currentKey, endKey) < 0
          || (includeEndKey && Bytes.compareTo(currentKey, endKey) == 0)));
  return new Pair<List<byte[]>, List<HRegionLocation>>(keysInRange,
      regionsInRange);
}

private void addCallsForOtherReplicas(
    ResultBoundedCompletionService<Pair<Result[], ScannerCallable>> cs, RegionLocations rl,
    int min, int max) {
  if (scan.getConsistency() == Consistency.STRONG) {
    return; // not scheduling on other replicas for strong consistency
  }
  for (int id = min; id <= max; id++) {
    if (currentScannerCallable.id == id) {
      continue; //this was already scheduled earlier
    }
    ScannerCallable s = currentScannerCallable.getScannerCallableForReplica(id);
    setStartRowForReplicaCallable(s);
    outstandingCallables.add(s);
    RetryingRPC retryingOnReplica = new RetryingRPC(s);
    cs.submit(retryingOnReplica, scannerTimeout, id);
  }
}

/**
 * Get the HRegionInfo from cache, if not there, from the hbase:meta table
 * @param  regionName
 * @return HRegionInfo for the region
 */
@SuppressWarnings("deprecation")
protected HRegionInfo getRegionInfo(final byte [] regionName) {
  String encodedName = HRegionInfo.encodeRegionName(regionName);
  RegionState regionState = getRegionState(encodedName);
  if (regionState != null) {
    return regionState.getRegion();
  }

  try {
    Pair<HRegionInfo, ServerName> p =
      MetaTableAccessor.getRegion(server.getConnection(), regionName);
    HRegionInfo hri = p == null ? null : p.getFirst();
    if (hri != null) {
      createRegionState(hri);
    }
    return hri;
  } catch (IOException e) {
    server.abort("Aborting because error occoured while reading "
      + Bytes.toStringBinary(regionName) + " from hbase:meta", e);
    return null;
  }
}

/**
 * Checks if the specified region has merge qualifiers, if so, try to clean
 * them
 * @param region
 * @return true if the specified region doesn't have merge qualifier now
 * @throws IOException
 */
public boolean cleanMergeQualifier(final HRegionInfo region)
    throws IOException {
  // Get merge regions if it is a merged region and already has merge
  // qualifier
  Pair<HRegionInfo, HRegionInfo> mergeRegions = MetaTableAccessor
      .getRegionsFromMergeQualifier(this.services.getConnection(),
        region.getRegionName());
  if (mergeRegions == null
      || (mergeRegions.getFirst() == null && mergeRegions.getSecond() == null)) {
    // It doesn't have merge qualifier, no need to clean
    return true;
  }
  // It shouldn't happen, we must insert/delete these two qualifiers together
  if (mergeRegions.getFirst() == null || mergeRegions.getSecond() == null) {
    LOG.error("Merged region " + region.getRegionNameAsString()
        + " has only one merge qualifier in META.");
    return false;
  }
  return cleanMergeRegion(region, mergeRegions.getFirst(),
      mergeRegions.getSecond());
}

/**
 * @param regionsInMeta
 * @return List of regions neither in transition nor assigned.
 * @throws IOException
 */
private Map<HRegionInfo, ServerName> regionsToAssignWithServerName(
    final List<Pair<HRegionInfo, ServerName>> regionsInMeta) throws IOException {
  Map<HRegionInfo, ServerName> regionsToAssign =
      new HashMap<HRegionInfo, ServerName>(regionsInMeta.size());
  RegionStates regionStates = this.assignmentManager.getRegionStates();
  for (Pair<HRegionInfo, ServerName> regionLocation : regionsInMeta) {
    HRegionInfo hri = regionLocation.getFirst();
    ServerName sn = regionLocation.getSecond();
    if (regionStates.isRegionOffline(hri)) {
      regionsToAssign.put(hri, sn);
    } else {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Skipping assign for the region " + hri + " during enable table "
            + hri.getTable() + " because its already in tranition or assigned.");
      }
    }
  }
  return regionsToAssign;
}

/**
 * @param regionsInMeta
 * @return List of regions neither in transition nor assigned.
 * @throws IOException
 */
private static Map<HRegionInfo, ServerName> regionsToAssignWithServerName(
    final MasterProcedureEnv env,
    final List<Pair<HRegionInfo, ServerName>> regionsInMeta) throws IOException {
  Map<HRegionInfo, ServerName> regionsToAssign =
      new HashMap<HRegionInfo, ServerName>(regionsInMeta.size());
  RegionStates regionStates = env.getMasterServices().getAssignmentManager().getRegionStates();
  for (Pair<HRegionInfo, ServerName> regionLocation : regionsInMeta) {
    HRegionInfo hri = regionLocation.getFirst();
    ServerName sn = regionLocation.getSecond();
    if (regionStates.isRegionOffline(hri)) {
      regionsToAssign.put(hri, sn);
    } else {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Skipping assign for the region " + hri + " during enable table "
            + hri.getTable() + " because its already in tranition or assigned.");
      }
    }
  }
  return regionsToAssign;
}

/**
 * Create up a map that is keyed by meta row name and whose value is the HRegionInfo and
 * ServerName to return for this row.
 * @return Map with faked hbase:meta content in it.
 */
static SortedMap<byte [], Pair<HRegionInfo, ServerName>> makeMeta(final byte [] tableName,
    final int regionCount, final long namespaceSpan, final int serverCount) {
  // I need a comparator for meta rows so we sort properly.
  SortedMap<byte [], Pair<HRegionInfo, ServerName>> meta =
    new ConcurrentSkipListMap<byte[], Pair<HRegionInfo,ServerName>>(new MetaRowsComparator());
  HRegionInfo [] hris = makeHRegionInfos(tableName, regionCount, namespaceSpan);
  ServerName [] serverNames = makeServerNames(serverCount);
  int per = regionCount / serverCount;
  int count = 0;
  for (HRegionInfo hri: hris) {
    Pair<HRegionInfo, ServerName> p =
      new Pair<HRegionInfo, ServerName>(hri, serverNames[count++ / per]);
    meta.put(hri.getRegionName(), p);
  }
  return meta;
}

private void runTest(Table hTable, int cqStart, int expectedSize) throws IOException {
  // [0, 2, ?, ?, ?, ?, 0, 0, 0, 1]
  byte[] fuzzyKey = new byte[10];
  ByteBuffer buf = ByteBuffer.wrap(fuzzyKey);
  buf.clear();
  buf.putShort((short) 2);
  for (int i = 0; i < 4; i++)
    buf.put((byte)63);
  buf.putInt((short)1);

  byte[] mask = new byte[] {0 , 0, 1, 1, 1, 1, 0, 0, 0, 0};

  Pair<byte[], byte[]> pair = new Pair<byte[], byte[]>(fuzzyKey, mask);
  FuzzyRowFilter fuzzyRowFilter = new FuzzyRowFilter(Lists.newArrayList(pair));
  ColumnRangeFilter columnRangeFilter = new ColumnRangeFilter(Bytes.toBytes(cqStart), true
          , Bytes.toBytes(4), true);
  //regular test
  runScanner(hTable, expectedSize, fuzzyRowFilter, columnRangeFilter);
  //reverse filter order test
  runScanner(hTable, expectedSize, columnRangeFilter, fuzzyRowFilter);
}

@Override
public ReturnCode filterKeyValue(Cell c) {
  final int startIndex = lastFoundIndex >= 0 ? lastFoundIndex : 0;
  final int size = fuzzyKeysData.size();
  for (int i = startIndex; i < size + startIndex; i++) {
    final int index = i % size;
    Pair<byte[], byte[]> fuzzyData = fuzzyKeysData.get(index);
    SatisfiesCode satisfiesCode =
        satisfies(isReversed(), c.getRowArray(), c.getRowOffset(), c.getRowLength(),
          fuzzyData.getFirst(), fuzzyData.getSecond());
    if (satisfiesCode == SatisfiesCode.YES) {
      lastFoundIndex = index;
      return ReturnCode.INCLUDE;
    }
  }
  // NOT FOUND -> seek next using hint
  lastFoundIndex = -1;

  return ReturnCode.SEEK_NEXT_USING_HINT;

}

private Pair<Integer, String> execWithRetries(String hostname, ServiceType service, String... cmd)
    throws IOException {
  RetryCounter retryCounter = retryCounterFactory.create();
  while (true) {
    try {
      return exec(hostname, service, cmd);
    } catch (IOException e) {
      retryOrThrow(retryCounter, e, hostname, cmd);
    }
    try {
      retryCounter.sleepUntilNextRetry();
    } catch (InterruptedException ex) {
      // ignore
      LOG.warn("Sleep Interrupted:" + ex);
    }
  }
}

/**
 * Execute the given command on the host using SSH
 * @return pair of exit code and command output
 * @throws IOException if something goes wrong.
 */
private Pair<Integer, String> exec(String hostname, ServiceType service, String... cmd)
  throws IOException {
  LOG.info("Executing remote command: " + StringUtils.join(cmd, " ") + " , hostname:" + hostname);

  RemoteShell shell = new RemoteShell(hostname, getServiceUser(service), cmd);
  try {
    shell.execute();
  } catch (Shell.ExitCodeException ex) {
    // capture the stdout of the process as well.
    String output = shell.getOutput();
    // add output for the ExitCodeException.
    throw new Shell.ExitCodeException(ex.getExitCode(), "stderr: " + ex.getMessage()
      + ", stdout: " + output);
  }

  LOG.info("Executed remote command, exit code:" + shell.getExitCode()
      + " , output:" + shell.getOutput());

  return new Pair<Integer, String>(shell.getExitCode(), shell.getOutput());
}

@VisibleForTesting
List<ReplicationPeer> listReplicationPeers() {
  Map<String, ReplicationPeerConfig> peers = listPeerConfigs();
  if (peers == null || peers.size() <= 0) {
    return null;
  }
  List<ReplicationPeer> listOfPeers = new ArrayList<ReplicationPeer>(peers.size());
  for (Entry<String, ReplicationPeerConfig> peerEntry : peers.entrySet()) {
    String peerId = peerEntry.getKey();
    try {
      Pair<ReplicationPeerConfig, Configuration> pair = this.replicationPeers.getPeerConf(peerId);
      Configuration peerConf = pair.getSecond();
      ReplicationPeer peer = new ReplicationPeerZKImpl(peerConf, peerId, pair.getFirst(),
          parseTableCFsFromConfig(this.getPeerTableCFs(peerId)));
      listOfPeers.add(peer);
    } catch (ReplicationException e) {
      LOG.warn("Failed to get valid replication peers. "
          + "Error connecting to peer cluster with peerId=" + peerId + ". Error message="
          + e.getMessage());
      LOG.debug("Failure details to get valid replication peers.", e);
      continue;
    }
  }
  return listOfPeers;
}

/**
 * Get the status of alter command - indicates how many regions have received
 * the updated schema Asynchronous operation.
 *
 * @param tableName TableName instance
 * @return Pair indicating the number of regions updated Pair.getFirst() is the
 *         regions that are yet to be updated Pair.getSecond() is the total number
 *         of regions of the table
 * @throws IOException
 *           if a remote or network exception occurs
 */
@Override
public Pair<Integer, Integer> getAlterStatus(final TableName tableName)
throws IOException {
  return executeCallable(new MasterCallable<Pair<Integer, Integer>>(getConnection()) {
    @Override
    public Pair<Integer, Integer> call(int callTimeout) throws ServiceException {
      PayloadCarryingRpcController controller = rpcControllerFactory.newController();
      controller.setCallTimeout(callTimeout);
      controller.setPriority(tableName);

      GetSchemaAlterStatusRequest req = RequestConverter
          .buildGetSchemaAlterStatusRequest(tableName);
      GetSchemaAlterStatusResponse ret = master.getSchemaAlterStatus(controller, req);
      Pair<Integer, Integer> pair = new Pair<Integer, Integer>(Integer.valueOf(ret
          .getYetToUpdateRegions()), Integer.valueOf(ret.getTotalRegions()));
      return pair;
    }
  });
}

@Override
public Pair<ZKProcedureCoordinatorRpcs, List<ZKProcedureMemberRpcs>> start(
    ZooKeeperWatcher watcher, String operationName,
    ProcedureCoordinator coordinator, String controllerName,
    ProcedureMember member, List<String> expected) throws Exception {
  // start the controller
  ZKProcedureCoordinatorRpcs controller = new ZKProcedureCoordinatorRpcs(
      watcher, operationName, CONTROLLER_NODE_NAME);
  controller.start(coordinator);

  // make a cohort controller for each expected node

  List<ZKProcedureMemberRpcs> cohortControllers = new ArrayList<ZKProcedureMemberRpcs>();
  for (String nodeName : expected) {
    ZKProcedureMemberRpcs cc = new ZKProcedureMemberRpcs(watcher, operationName);
    cc.start(nodeName, member);
    cohortControllers.add(cc);
  }
  return new Pair<ZKProcedureCoordinatorRpcs, List<ZKProcedureMemberRpcs>>(
      controller, cohortControllers);
}

private Pair<Path, Path> splitStoreFile(final byte[] family, final StoreFile sf)
    throws IOException {
  if (LOG.isDebugEnabled()) {
      LOG.debug("Splitting started for store file: " + sf.getPath() + " for region: " +
                this.parent);
  }
  HRegionFileSystem fs = this.parent.getRegionFileSystem();
  String familyName = Bytes.toString(family);

  Path path_a =
      fs.splitStoreFile(this.hri_a, familyName, sf, this.splitrow, false,
        this.parent.getSplitPolicy());
  Path path_b =
      fs.splitStoreFile(this.hri_b, familyName, sf, this.splitrow, true,
        this.parent.getSplitPolicy());
  if (LOG.isDebugEnabled()) {
      LOG.debug("Splitting complete for store file: " + sf.getPath() + " for region: " +
                this.parent);
  }
  return new Pair<Path,Path>(path_a, path_b);
}

@Test
public void testMaxSend() {
  ClusterStatusPublisher csp = new ClusterStatusPublisher() {
    @Override
    protected List<Pair<ServerName, Long>> getDeadServers(long since) {
      List<Pair<ServerName, Long>> res = new ArrayList<Pair<ServerName, Long>>();
      switch ((int) EnvironmentEdgeManager.currentTime()) {
        case 2:
          res.add(new Pair<ServerName, Long>(ServerName.valueOf("hn", 10, 10), 1L));
          break;
        case 1000:
          break;
      }

      return res;
    }
  };

  mee.setValue(2);
  for (int i = 0; i < ClusterStatusPublisher.NB_SEND; i++) {
    Assert.assertEquals("i=" + i, 1, csp.generateDeadServersListToSend().size());
  }
  mee.setValue(1000);
  Assert.assertTrue(csp.generateDeadServersListToSend().isEmpty());
}

private void waitAndVerifyRegionNum(HMaster master, TableName tablename,
    int expectedRegionNum) throws Exception {
  List<Pair<HRegionInfo, ServerName>> tableRegionsInMeta;
  List<HRegionInfo> tableRegionsInMaster;
  long timeout = System.currentTimeMillis() + waitTime;
  while (System.currentTimeMillis() < timeout) {
    tableRegionsInMeta = MetaTableAccessor.getTableRegionsAndLocations(master.getZooKeeper(),
      master.getConnection(), tablename);
    tableRegionsInMaster = master.getAssignmentManager().getRegionStates()
        .getRegionsOfTable(tablename);
    if (tableRegionsInMeta.size() == expectedRegionNum
        && tableRegionsInMaster.size() == expectedRegionNum) {
      break;
    }
    Thread.sleep(250);
  }

  tableRegionsInMeta = MetaTableAccessor.getTableRegionsAndLocations(master.getZooKeeper(),
    master.getConnection(), tablename);
  LOG.info("Regions after merge:" + Joiner.on(',').join(tableRegionsInMeta));
  assertEquals(expectedRegionNum, tableRegionsInMeta.size());
}

@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()));
}

/**
 * @param pbBytes A pb serialized {@link FuzzyRowFilter} instance
 * @return An instance of {@link FuzzyRowFilter} made from <code>bytes</code>
 * @throws DeserializationException
 * @see #toByteArray
 */
public static FuzzyRowFilter parseFrom(final byte[] pbBytes) throws DeserializationException {
  FilterProtos.FuzzyRowFilter proto;
  try {
    proto = FilterProtos.FuzzyRowFilter.parseFrom(pbBytes);
  } catch (InvalidProtocolBufferException e) {
    throw new DeserializationException(e);
  }
  int count = proto.getFuzzyKeysDataCount();
  ArrayList<Pair<byte[], byte[]>> fuzzyKeysData = new ArrayList<Pair<byte[], byte[]>>(count);
  for (int i = 0; i < count; ++i) {
    BytesBytesPair current = proto.getFuzzyKeysData(i);
    byte[] keyBytes = current.getFirst().toByteArray();
    byte[] keyMeta = current.getSecond().toByteArray();
    fuzzyKeysData.add(new Pair<byte[], byte[]>(keyBytes, keyMeta));
  }
  return new FuzzyRowFilter(fuzzyKeysData);
}

/**
 * Get the full path of the HFile referenced by the back reference
 *
 * @param rootDir root hbase directory
 * @param linkRefPath Link Back Reference path
 * @return full path of the referenced hfile
 */
public static Path getHFileFromBackReference(final Path rootDir, final Path linkRefPath) {
  Pair<TableName, String> p = parseBackReferenceName(linkRefPath.getName());
  TableName linkTableName = p.getFirst();
  String linkRegionName = p.getSecond();

  String hfileName = getBackReferenceFileName(linkRefPath.getParent());
  Path familyPath = linkRefPath.getParent().getParent();
  Path regionPath = familyPath.getParent();
  Path tablePath = regionPath.getParent();

  String linkName = createHFileLinkName(FSUtils.getTableName(tablePath),
          regionPath.getName(), hfileName);
  Path linkTableDir = FSUtils.getTableDir(rootDir, linkTableName);
  Path regionDir = HRegion.getRegionDir(linkTableDir, linkRegionName);
  return new Path(new Path(regionDir, familyPath.getName()), linkName);
}

/**
 * If the input is a region name, it is returned as is. If it's an
 * encoded region name, the corresponding region is found from meta
 * and its region name is returned. If we can't find any region in
 * meta matching the input as either region name or encoded region
 * name, the input is returned as is. We don't throw unknown
 * region exception.
 */
private byte[] getRegionName(
    final byte[] regionNameOrEncodedRegionName) throws IOException {
  if (Bytes.equals(regionNameOrEncodedRegionName,
      HRegionInfo.FIRST_META_REGIONINFO.getRegionName())
        || Bytes.equals(regionNameOrEncodedRegionName,
          HRegionInfo.FIRST_META_REGIONINFO.getEncodedNameAsBytes())) {
    return HRegionInfo.FIRST_META_REGIONINFO.getRegionName();
  }
  byte[] tmp = regionNameOrEncodedRegionName;
  Pair<HRegionInfo, ServerName> regionServerPair = getRegion(regionNameOrEncodedRegionName);
  if (regionServerPair != null && regionServerPair.getFirst() != null) {
    tmp = regionServerPair.getFirst().getRegionName();
  }
  return tmp;
}

private PairOfSameType<HRegionInfo> requestMergeRegion(
    HMaster master, TableName tablename,
    int regionAnum, int regionBnum) throws Exception {
  List<Pair<HRegionInfo, ServerName>> tableRegions = MetaTableAccessor
      .getTableRegionsAndLocations(master.getZooKeeper(),
        master.getConnection(), tablename);
  HRegionInfo regionA = tableRegions.get(regionAnum).getFirst();
  HRegionInfo regionB = tableRegions.get(regionBnum).getFirst();
  TEST_UTIL.getHBaseAdmin().mergeRegions(
    regionA.getEncodedNameAsBytes(),
    regionB.getEncodedNameAsBytes(), false);
  return new PairOfSameType<HRegionInfo>(regionA, regionB);
}

/**
 * Reads a varInt value stored in an array.
 *
 * @param input
 *          Input array where the varInt is available
 * @param offset
 *          Offset in the input array where varInt is available
 * @return A pair of integers in which first value is the actual decoded varInt value and second
 *         value as number of bytes taken by this varInt for it's storage in the input array.
 * @throws IOException
 */
public static Pair<Integer, Integer> readRawVarint32(byte[] input, int offset) throws IOException {
  int newOffset = offset;
  byte tmp = input[newOffset++];
  if (tmp >= 0) {
    return new Pair<Integer, Integer>((int) tmp, newOffset - offset);
  }
  int result = tmp & 0x7f;
  tmp = input[newOffset++];
  if (tmp >= 0) {
    result |= tmp << 7;
  } else {
    result |= (tmp & 0x7f) << 7;
    tmp = input[newOffset++];
    if (tmp >= 0) {
      result |= tmp << 14;
    } else {
      result |= (tmp & 0x7f) << 14;
      tmp = input[newOffset++];
      if (tmp >= 0) {
        result |= tmp << 21;
      } else {
        result |= (tmp & 0x7f) << 21;
        tmp = input[newOffset++];
        result |= tmp << 28;
        if (tmp < 0) {
          // Discard upper 32 bits.
          for (int i = 0; i < 5; i++) {
            tmp = input[newOffset++];
            if (tmp >= 0) {
              return new Pair<Integer, Integer>(result, newOffset - offset);
            }
          }
          throw new IOException("Malformed varint");
        }
      }
    }
  }
  return new Pair<Integer, Integer>(result, newOffset - offset);
}

@Override
public Pair<Result[], ScannerCallable> call(int callTimeout) throws IOException {
  // since the retries is done within the ResultBoundedCompletionService,
  // we don't invoke callWithRetries here
  if (cancelled) {
    return null;
  }
  Result[] res = this.caller.callWithoutRetries(this.callable, callTimeout);
  return new Pair<Result[], ScannerCallable>(res, this.callable);
}

/**
 * Gets the region info and assignment for the specified region.
 * @param connection connection we're using
 * @param regionName Region to lookup.
 * @return Location and HRegionInfo for <code>regionName</code>
 * @throws IOException
 * @deprecated use {@link #getRegionLocation(Connection, byte[])} instead
 */
@Deprecated
public static Pair<HRegionInfo, ServerName> getRegion(Connection connection, byte [] regionName)
  throws IOException {
  HRegionLocation location = getRegionLocation(connection, regionName);
  return location == null
    ? null
    : new Pair<HRegionInfo, ServerName>(location.getRegionInfo(), location.getServerName());
}

static List<HRegionInfo> getListOfHRegionInfos(final List<Pair<HRegionInfo, ServerName>> pairs) {
  if (pairs == null || pairs.isEmpty()) return null;
  List<HRegionInfo> result = new ArrayList<HRegionInfo>(pairs.size());
  for (Pair<HRegionInfo, ServerName> pair: pairs) {
    result.add(pair.getFirst());
  }
  return result;
}

static ScanResponse doMetaScanResponse(final SortedMap<byte [], Pair<HRegionInfo, ServerName>> meta,
    final AtomicLong sequenceids, final ScanRequest request) {
  ScanResponse.Builder builder = ScanResponse.newBuilder();
  int max = request.getNumberOfRows();
  int count = 0;
  Map<byte [], Pair<HRegionInfo, ServerName>> tail =
    request.hasScan()? meta.tailMap(request.getScan().getStartRow().toByteArray()): meta;
    ClientProtos.Result.Builder resultBuilder = ClientProtos.Result.newBuilder();
  for (Map.Entry<byte [], Pair<HRegionInfo, ServerName>> e: tail.entrySet()) {
    // Can be 0 on open of a scanner -- i.e. rpc to setup scannerid only.
    if (max <= 0) break;
    if (++count > max) break;
    HRegionInfo hri = e.getValue().getFirst();
    ByteString row = ByteStringer.wrap(hri.getRegionName());
    resultBuilder.clear();
    resultBuilder.addCell(getRegionInfo(row, hri));
    resultBuilder.addCell(getServer(row, e.getValue().getSecond()));
    resultBuilder.addCell(getStartCode(row));
    builder.addResults(resultBuilder.build());
    // Set more to false if we are on the last region in table.
    if (hri.getEndKey().length <= 0) builder.setMoreResults(false);
    else builder.setMoreResults(true);
  }
  // If no scannerid, set one.
  builder.setScannerId(request.hasScannerId()?
    request.getScannerId(): sequenceids.incrementAndGet());
  return builder.build();
}

@Override
public boolean flush() throws IOException {
  String curLoc = null;
  int curSize = 0;
  List<Pair<HRegionLocation, Entry>> curQueue = null;
  synchronized (this.serverToBufferQueueMap) {
    for (String locationKey : this.serverToBufferQueueMap.keySet()) {
      curQueue = this.serverToBufferQueueMap.get(locationKey);
      if (!curQueue.isEmpty()) {
        curSize = curQueue.size();
        curLoc = locationKey;
        break;
      }
    }
    if (curSize > 0) {
      this.serverToBufferQueueMap.remove(curLoc);
    }
  }

  if (curSize > 0) {
    this.processWorkItems(curLoc, curQueue);
    // We should already have control of the monitor; ensure this is the case.
    synchronized(controller.dataAvailable) {
      controller.dataAvailable.notifyAll();
    }
    return true;
  }
  return false;
}

FakeServer(final Configuration c, final SortedMap<byte [], Pair<HRegionInfo, ServerName>> meta,
    final AtomicLong sequenceids) {
  this.meta = meta;
  this.sequenceids = sequenceids;

  // Pause to simulate the server taking time applying the edits.  This will drive up the
  // number of threads used over in client.
  this.multiPause = c.getLong("hbase.test.multi.pause.when.done", 0);
  this.tooManyMultiRequests = c.getInt("hbase.test.multi.too.many", 3);
}

private Pair<Map<ServerName, List<HRegionInfo>>, List<HRegionInfo>>
segregateRegionsAndAssignRegionsWithFavoredNodes(List<HRegionInfo> regions,
    List<ServerName> availableServers) {
  Map<ServerName, List<HRegionInfo>> assignmentMapForFavoredNodes =
      new HashMap<ServerName, List<HRegionInfo>>(regions.size() / 2);
  List<HRegionInfo> regionsWithNoFavoredNodes = new ArrayList<HRegionInfo>(regions.size()/2);
  for (HRegionInfo region : regions) {
    List<ServerName> favoredNodes = globalFavoredNodesAssignmentPlan.getFavoredNodes(region);
    ServerName primaryHost = null;
    ServerName secondaryHost = null;
    ServerName tertiaryHost = null;
    if (favoredNodes != null) {
      for (ServerName s : favoredNodes) {
        ServerName serverWithLegitStartCode = availableServersContains(availableServers, s);
        if (serverWithLegitStartCode != null) {
          FavoredNodesPlan.Position position =
              FavoredNodesPlan.getFavoredServerPosition(favoredNodes, s);
          if (Position.PRIMARY.equals(position)) {
            primaryHost = serverWithLegitStartCode;
          } else if (Position.SECONDARY.equals(position)) {
            secondaryHost = serverWithLegitStartCode;
          } else if (Position.TERTIARY.equals(position)) {
            tertiaryHost = serverWithLegitStartCode;
          }
        }
      }
      assignRegionToAvailableFavoredNode(assignmentMapForFavoredNodes, region,
            primaryHost, secondaryHost, tertiaryHost);
    }
    if (primaryHost == null && secondaryHost == null && tertiaryHost == null) {
      //all favored nodes unavailable
      regionsWithNoFavoredNodes.add(region);
    }
  }
  return new Pair<Map<ServerName, List<HRegionInfo>>, List<HRegionInfo>>(
      assignmentMapForFavoredNodes, regionsWithNoFavoredNodes);
}

/**
 * Extract all the servers dead since a given time, and sort them.
 * @param ts the time, 0 for all
 * @return a sorted array list, by death time, lowest values first.
 */
public synchronized List<Pair<ServerName, Long>> copyDeadServersSince(long ts){
  List<Pair<ServerName, Long>> res =  new ArrayList<Pair<ServerName, Long>>(size());

  for (Map.Entry<ServerName, Long> entry:deadServers.entrySet()){
    if (entry.getValue() >= ts){
      res.add(new Pair<ServerName, Long>(entry.getKey(), entry.getValue()));
    }
  }

  Collections.sort(res, ServerNameDeathDateComparator);
  return res;
}

/**
 * Create a protocol buffer UpdateFavoredNodesRequest to update a list of favorednode mappings
 * @param updateRegionInfos
 * @return a protocol buffer UpdateFavoredNodesRequest
 */
public static UpdateFavoredNodesRequest buildUpdateFavoredNodesRequest(
    final List<Pair<HRegionInfo, List<ServerName>>> updateRegionInfos) {
  UpdateFavoredNodesRequest.Builder ubuilder = UpdateFavoredNodesRequest.newBuilder();
  for (Pair<HRegionInfo, List<ServerName>> pair : updateRegionInfos) {
    RegionUpdateInfo.Builder builder = RegionUpdateInfo.newBuilder();
    builder.setRegion(HRegionInfo.convert(pair.getFirst()));
    for (ServerName server : pair.getSecond()) {
      builder.addFavoredNodes(ProtobufUtil.toServerName(server));
    }
    ubuilder.addUpdateInfo(builder.build());
  }
  return ubuilder.build();
}

/**
 * Return the region and current deployment for the region containing
 * the given row. If the region cannot be found, returns null. If it
 * is found, but not currently deployed, the second element of the pair
 * may be null.
 */
@VisibleForTesting // Used by TestMaster.
Pair<HRegionInfo, ServerName> getTableRegionForRow(
    final TableName tableName, final byte [] rowKey)
throws IOException {
  final AtomicReference<Pair<HRegionInfo, ServerName>> result =
    new AtomicReference<Pair<HRegionInfo, ServerName>>(null);

  MetaScannerVisitor visitor =
    new MetaScannerVisitorBase() {
      @Override
      public boolean processRow(Result data) throws IOException {
        if (data == null || data.size() <= 0) {
          return true;
        }
        Pair<HRegionInfo, ServerName> pair = HRegionInfo.getHRegionInfoAndServerName(data);
        if (pair == null) {
          return false;
        }
        if (!pair.getFirst().getTable().equals(tableName)) {
          return false;
        }
        result.set(pair);
        return true;
      }
  };

  MetaScanner.metaScan(clusterConnection, visitor, tableName, rowKey, 1);
  return result.get();
}

/**
 * Create the dead server to send. A dead server is sent NB_SEND times. We send at max
 * MAX_SERVER_PER_MESSAGE at a time. if there are too many dead servers, we send the newly
 * dead first.
 */
protected List<ServerName> generateDeadServersListToSend() {
  // We're getting the message sent since last time, and add them to the list
  long since = EnvironmentEdgeManager.currentTime() - messagePeriod * 2;
  for (Pair<ServerName, Long> dead : getDeadServers(since)) {
    lastSent.putIfAbsent(dead.getFirst(), 0);
  }

  // We're sending the new deads first.
  List<Map.Entry<ServerName, Integer>> entries = new ArrayList<Map.Entry<ServerName, Integer>>();
  entries.addAll(lastSent.entrySet());
  Collections.sort(entries, new Comparator<Map.Entry<ServerName, Integer>>() {
    @Override
    public int compare(Map.Entry<ServerName, Integer> o1, Map.Entry<ServerName, Integer> o2) {
      return o1.getValue().compareTo(o2.getValue());
    }
  });

  // With a limit of MAX_SERVER_PER_MESSAGE
  int max = entries.size() > MAX_SERVER_PER_MESSAGE ? MAX_SERVER_PER_MESSAGE : entries.size();
  List<ServerName> res = new ArrayList<ServerName>(max);

  for (int i = 0; i < max; i++) {
    Map.Entry<ServerName, Integer> toSend = entries.get(i);
    if (toSend.getValue() >= (NB_SEND - 1)) {
      lastSent.remove(toSend.getKey());
    } else {
      lastSent.replace(toSend.getKey(), toSend.getValue(), toSend.getValue() + 1);
    }

    res.add(toSend.getKey());
  }

  return res;
}

/**
 * Get the servers which died since a given timestamp.
 * protected because it can be subclassed by the tests.
 */
protected List<Pair<ServerName, Long>> getDeadServers(long since) {
  if (master.getServerManager() == null) {
    return Collections.emptyList();
  }

  return master.getServerManager().getDeadServers().copyDeadServersSince(since);
}

/**
 * It removes recovering regions under /hbase/recovering-regions/[encoded region name] so that the
 * region server hosting the region can allow reads to the recovered region
 * @param serverNames servers which are just recovered
 * @param isMetaRecovery whether current recovery is for the meta region on
 *          <code>serverNames<code>
 */
private void removeRecoveringRegions(final Set<ServerName> serverNames, Boolean isMetaRecovery) {
  if (!isLogReplaying()) {
    // the function is only used in WALEdit direct replay mode
    return;
  }
  if (serverNames == null || serverNames.isEmpty()) return;

  Set<String> recoveredServerNameSet = new HashSet<String>();
  for (ServerName tmpServerName : serverNames) {
    recoveredServerNameSet.add(tmpServerName.getServerName());
  }

  this.recoveringRegionLock.lock();
  try {
    ((BaseCoordinatedStateManager) server.getCoordinatedStateManager())
        .getSplitLogManagerCoordination().removeRecoveringRegions(recoveredServerNameSet,
          isMetaRecovery);
  } catch (IOException e) {
    LOG.warn("removeRecoveringRegions got exception. Will retry", e);
    if (serverNames != null && !serverNames.isEmpty()) {
      this.failedRecoveringRegionDeletions.add(new Pair<Set<ServerName>, Boolean>(serverNames,
          isMetaRecovery));
    }
  } finally {
    this.recoveringRegionLock.unlock();
  }
}

/**
 * @param regionName Name of a region.
 * @return a pair of HRegionInfo and ServerName if <code>regionName</code> is
 *  a verified region name (we call {@link
 *  MetaTableAccessor#getRegion(HConnection, byte[])}
 *  else null.
 * Throw IllegalArgumentException if <code>regionName</code> is null.
 * @throws IOException
 */
Pair<HRegionInfo, ServerName> getRegion(final byte[] regionName) throws IOException {
  if (regionName == null) {
    throw new IllegalArgumentException("Pass a table name or region name");
  }
  Pair<HRegionInfo, ServerName> pair =
    MetaTableAccessor.getRegion(connection, regionName);
  if (pair == null) {
    final AtomicReference<Pair<HRegionInfo, ServerName>> result =
      new AtomicReference<Pair<HRegionInfo, ServerName>>(null);
    final String encodedName = Bytes.toString(regionName);
    MetaScannerVisitor visitor = new MetaScannerVisitorBase() {
      @Override
      public boolean processRow(Result data) throws IOException {
        HRegionInfo info = HRegionInfo.getHRegionInfo(data);
        if (info == null) {
          LOG.warn("No serialized HRegionInfo in " + data);
          return true;
        }
        RegionLocations rl = MetaTableAccessor.getRegionLocations(data);
        boolean matched = false;
        ServerName sn = null;
        for (HRegionLocation h : rl.getRegionLocations()) {
          if (h != null && encodedName.equals(h.getRegionInfo().getEncodedName())) {
            sn = h.getServerName();
            info = h.getRegionInfo();
            matched = true;
          }
        }
        if (!matched) return true;
        result.set(new Pair<HRegionInfo, ServerName>(info, sn));
        return false; // found the region, stop
      }
    };

    MetaScanner.metaScan(connection, visitor, null);
    pair = result.get();
  }
  return pair;
}