Java 类org.apache.hadoop.hbase.ipc.PayloadCarryingRpcController 实例源码

/**
 * Sends an MERGE REGIONS RPC to the specified server to merge the specified
 * regions.
 * <p>
 * A region server could reject the close request because it either does not
 * have the specified region.
 * @param server server to merge regions
 * @param region_a region to merge
 * @param region_b region to merge
 * @param forcible true if do a compulsory merge, otherwise we will only merge
 *          two adjacent regions
 * @throws IOException
 */
public void sendRegionsMerge(ServerName server, HRegionInfo region_a,
    HRegionInfo region_b, boolean forcible) throws IOException {
  if (server == null)
    throw new NullPointerException("Passed server is null");
  if (region_a == null || region_b == null)
    throw new NullPointerException("Passed region is null");
  AdminService.BlockingInterface admin = getRsAdmin(server);
  if (admin == null) {
    throw new IOException("Attempting to send MERGE REGIONS RPC to server "
        + server.toString() + " for region "
        + region_a.getRegionNameAsString() + ","
        + region_b.getRegionNameAsString()
        + " failed because no RPC connection found to this server");
  }
  PayloadCarryingRpcController controller = newRpcController();
  ProtobufUtil.mergeRegions(controller, admin, region_a, region_b, forcible);
}

private void replayToServer(HRegionInfo regionInfo, List<Entry> entries)
    throws IOException, ServiceException {
  if (entries.isEmpty()) return;

  Entry[] entriesArray = new Entry[entries.size()];
  entriesArray = entries.toArray(entriesArray);
  AdminService.BlockingInterface remoteSvr = conn.getAdmin(getLocation().getServerName());

  Pair<AdminProtos.ReplicateWALEntryRequest, CellScanner> p =
      ReplicationProtbufUtil.buildReplicateWALEntryRequest(entriesArray);
  PayloadCarryingRpcController controller = rpcControllerFactory.newController(p.getSecond());
  try {
    remoteSvr.replay(controller, p.getFirst());
  } catch (ServiceException se) {
    throw ProtobufUtil.getRemoteException(se);
  }
}

/**
 * Replicate WAL entries on the region server.
 *
 * @param controller the RPC controller
 * @param request the request
 * @throws ServiceException
 */
@Override
@QosPriority(priority=HConstants.REPLICATION_QOS)
public ReplicateWALEntryResponse replicateWALEntry(final RpcController controller,
    final ReplicateWALEntryRequest request) throws ServiceException {
  try {
    checkOpen();
    if (regionServer.replicationSinkHandler != null) {
      requestCount.increment();
      List<WALEntry> entries = request.getEntryList();
      CellScanner cellScanner = ((PayloadCarryingRpcController)controller).cellScanner();
      regionServer.getRegionServerCoprocessorHost().preReplicateLogEntries(entries, cellScanner);
      regionServer.replicationSinkHandler.replicateLogEntries(entries, cellScanner);
      regionServer.getRegionServerCoprocessorHost().postReplicateLogEntries(entries, cellScanner);
      return ReplicateWALEntryResponse.newBuilder().build();
    } else {
      throw new ServiceException("Replication services are not initialized yet");
    }
  } catch (IOException ie) {
    throw new ServiceException(ie);
  }
}

/**
 * Test get of meta region fails properly if nothing to connect to.
 * @throws IOException
 * @throws InterruptedException
 * @throws KeeperException
 * @throws ServiceException
 */
@Test
public void testVerifyMetaRegionLocationFails()
throws IOException, InterruptedException, KeeperException, ServiceException {
  ClusterConnection connection = Mockito.mock(ClusterConnection.class);
  ServiceException connectException =
    new ServiceException(new ConnectException("Connection refused"));
  final AdminProtos.AdminService.BlockingInterface implementation =
    Mockito.mock(AdminProtos.AdminService.BlockingInterface.class);
  Mockito.when(implementation.getRegionInfo((RpcController)Mockito.any(),
    (GetRegionInfoRequest)Mockito.any())).thenThrow(connectException);
  Mockito.when(connection.getAdmin(Mockito.any(ServerName.class))).
    thenReturn(implementation);
      RpcControllerFactory controllerFactory = Mockito.mock(RpcControllerFactory.class);
      Mockito.when(controllerFactory.newController()).thenReturn(
        Mockito.mock(PayloadCarryingRpcController.class));
      Mockito.when(connection.getRpcControllerFactory()).thenReturn(controllerFactory);

  ServerName sn = ServerName.valueOf("example.com", 1234, System.currentTimeMillis());
  MetaTableLocator.setMetaLocation(this.watcher,
          sn,
          RegionState.State.OPENING);
  assertFalse(new MetaTableLocator().verifyMetaRegionLocation(connection, watcher, 100));
  MetaTableLocator.setMetaLocation(this.watcher, sn, RegionState.State.OPEN);
  assertFalse(new MetaTableLocator().verifyMetaRegionLocation(connection, watcher, 100));
}

/**
 * Abort a procedure but does not block and wait for it be completely removed.
 * You can use Future.get(long, TimeUnit) to wait on the operation to complete.
 * It may throw ExecutionException if there was an error while executing the operation
 * or TimeoutException in case the wait timeout was not long enough to allow the
 * operation to complete.
 *
 * @param procId ID of the procedure to abort
 * @param mayInterruptIfRunning if the proc completed at least one step, should it be aborted?
 * @return true if aborted, false if procedure already completed or does not exist
 * @throws IOException
 */
@Override
public Future<Boolean> abortProcedureAsync(
    final long procId,
    final boolean mayInterruptIfRunning) throws IOException {
  Boolean abortProcResponse = executeCallable(
    new MasterCallable<AbortProcedureResponse>(getConnection()) {
      @Override
      public AbortProcedureResponse call(int callTimeout) throws ServiceException {
        PayloadCarryingRpcController controller = rpcControllerFactory.newController();
        controller.setCallTimeout(callTimeout);
        AbortProcedureRequest abortProcRequest =
            AbortProcedureRequest.newBuilder().setProcId(procId).build();
        return master.abortProcedure(controller, abortProcRequest);
      }
    }).getIsProcedureAborted();

  AbortProcedureFuture abortProcFuture =
      new AbortProcedureFuture(this, procId, abortProcResponse);
  return abortProcFuture;
}

static HTableDescriptor getTableDescriptor(final TableName tableName, HConnection connection,
    RpcRetryingCallerFactory rpcCallerFactory, final RpcControllerFactory rpcControllerFactory,
       int operationTimeout) throws TableNotFoundException, IOException {

    if (tableName == null) return null;
    HTableDescriptor htd = executeCallable(new MasterCallable<HTableDescriptor>(connection) {
      @Override
      public HTableDescriptor call(int callTimeout) throws ServiceException {
        PayloadCarryingRpcController controller = rpcControllerFactory.newController();
        controller.setCallTimeout(callTimeout);
        GetTableDescriptorsResponse htds;
        GetTableDescriptorsRequest req =
                RequestConverter.buildGetTableDescriptorsRequest(tableName);
        htds = master.getTableDescriptors(controller, req);

        if (!htds.getTableSchemaList().isEmpty()) {
          return HTableDescriptor.convert(htds.getTableSchemaList().get(0));
        }
        return null;
      }
    }, rpcCallerFactory, operationTimeout);
    if (htd != null) {
      return htd;
    }
    throw new TableNotFoundException(tableName.getNameAsString());
}

/**
 * Deletes the table but does not block and wait for it be completely removed.
 * You can use Future.get(long, TimeUnit) to wait on the operation to complete.
 * It may throw ExecutionException if there was an error while executing the operation
 * or TimeoutException in case the wait timeout was not long enough to allow the
 * operation to complete.
 *
 * @param desc table descriptor for table
 * @param tableName name of table to delete
 * @throws IOException if a remote or network exception occurs
 * @return the result of the async delete. You can use Future.get(long, TimeUnit)
 *    to wait on the operation to complete.
 */
// TODO: This should be called Async but it will break binary compatibility
private Future<Void> deleteTableAsyncV2(final TableName tableName) throws IOException {
  DeleteTableResponse response = executeCallable(
    new MasterCallable<DeleteTableResponse>(getConnection()) {
      @Override
      public DeleteTableResponse call(int callTimeout) throws ServiceException {
        PayloadCarryingRpcController controller = rpcControllerFactory.newController();
        controller.setCallTimeout(callTimeout);
        controller.setPriority(tableName);
        DeleteTableRequest req =
            RequestConverter.buildDeleteTableRequest(tableName, ng.getNonceGroup(),ng.newNonce());
        return master.deleteTable(controller,req);
      }
    });
  return new DeleteTableFuture(this, tableName, response);
}

/**
 * Enable the table but does not block and wait for it be completely enabled.
 * You can use Future.get(long, TimeUnit) to wait on the operation to complete.
 * It may throw ExecutionException if there was an error while executing the operation
 * or TimeoutException in case the wait timeout was not long enough to allow the
 * operation to complete.
 *
 * @param tableName name of table to delete
 * @throws IOException if a remote or network exception occurs
 * @return the result of the async enable. You can use Future.get(long, TimeUnit)
 *    to wait on the operation to complete.
 */
// TODO: This should be called Async but it will break binary compatibility
private Future<Void> enableTableAsyncV2(final TableName tableName) throws IOException {
  TableName.isLegalFullyQualifiedTableName(tableName.getName());
  EnableTableResponse response = executeCallable(
    new MasterCallable<EnableTableResponse>(getConnection()) {
      @Override
      public EnableTableResponse call(int callTimeout) throws ServiceException {
        PayloadCarryingRpcController controller = rpcControllerFactory.newController();
        controller.setCallTimeout(callTimeout);
        controller.setPriority(tableName);

        LOG.info("Started enable of " + tableName);
        EnableTableRequest req =
            RequestConverter.buildEnableTableRequest(tableName, ng.getNonceGroup(),ng.newNonce());
        return master.enableTable(controller,req);
      }
    });
  return new EnableTableFuture(this, tableName, response);
}

/**
 * Disable the table but does not block and wait for it be completely disabled.
 * You can use Future.get(long, TimeUnit) to wait on the operation to complete.
 * It may throw ExecutionException if there was an error while executing the operation
 * or TimeoutException in case the wait timeout was not long enough to allow the
 * operation to complete.
 *
 * @param tableName name of table to delete
 * @throws IOException if a remote or network exception occurs
 * @return the result of the async disable. You can use Future.get(long, TimeUnit)
 *    to wait on the operation to complete.
 */
// TODO: This should be called Async but it will break binary compatibility
private Future<Void> disableTableAsyncV2(final TableName tableName) throws IOException {
  TableName.isLegalFullyQualifiedTableName(tableName.getName());
  DisableTableResponse response = executeCallable(
    new MasterCallable<DisableTableResponse>(getConnection()) {
      @Override
      public DisableTableResponse call(int callTimeout) throws ServiceException {
        PayloadCarryingRpcController controller = rpcControllerFactory.newController();
        controller.setCallTimeout(callTimeout);
        controller.setPriority(tableName);

        LOG.info("Started disable of " + tableName);
        DisableTableRequest req =
            RequestConverter.buildDisableTableRequest(
              tableName, ng.getNonceGroup(), ng.newNonce());
        return master.disableTable(controller, req);
      }
    });
  return new DisableTableFuture(this, tableName, response);
}

/**
 * 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;
    }
  });
}

/**
 * Add a column to an existing table.
 * Asynchronous operation.
 *
 * @param tableName name of the table to add column to
 * @param column column descriptor of column to be added
 * @throws IOException if a remote or network exception occurs
 */
@Override
public void addColumn(final TableName tableName, final HColumnDescriptor column)
throws IOException {
  executeCallable(new MasterCallable<Void>(getConnection()) {
    @Override
    public Void call(int callTimeout) throws ServiceException {
      PayloadCarryingRpcController controller = rpcControllerFactory.newController();
      controller.setCallTimeout(callTimeout);
      controller.setPriority(tableName);
      AddColumnRequest req = RequestConverter.buildAddColumnRequest(
        tableName, column, ng.getNonceGroup(), ng.newNonce());
      master.addColumn(controller,req);
      return null;
    }
  });
}

/**
 * Delete a column from a table.
 * Asynchronous operation.
 *
 * @param tableName name of table
 * @param columnName name of column to be deleted
 * @throws IOException if a remote or network exception occurs
 */
@Override
public void deleteColumn(final TableName tableName, final byte [] columnName)
throws IOException {
  executeCallable(new MasterCallable<Void>(getConnection()) {
    @Override
    public Void call(int callTimeout) throws ServiceException {
      PayloadCarryingRpcController controller = rpcControllerFactory.newController();
      controller.setCallTimeout(callTimeout);
      controller.setPriority(tableName);
      DeleteColumnRequest req = RequestConverter.buildDeleteColumnRequest(
        tableName, columnName, ng.getNonceGroup(), ng.newNonce());
      master.deleteColumn(controller, req);
      return null;
    }
  });
}

/**
 * Modify an existing column family on a table.
 * Asynchronous operation.
 *
 * @param tableName name of table
 * @param descriptor new column descriptor to use
 * @throws IOException if a remote or network exception occurs
 */
@Override
public void modifyColumn(final TableName tableName, final HColumnDescriptor descriptor)
throws IOException {
  executeCallable(new MasterCallable<Void>(getConnection()) {
    @Override
    public Void call(int callTimeout) throws ServiceException {
      PayloadCarryingRpcController controller = rpcControllerFactory.newController();
      controller.setCallTimeout(callTimeout);
      controller.setPriority(tableName);
      ModifyColumnRequest req = RequestConverter.buildModifyColumnRequest(
        tableName, descriptor, ng.getNonceGroup(), ng.newNonce());
      master.modifyColumn(controller, req);
      return null;
    }
  });
}

/**
 * @param regionName
 *          Region name to assign.
 * @throws MasterNotRunningException
 * @throws ZooKeeperConnectionException
 * @throws IOException
 */
@Override
public void assign(final byte[] regionName) throws MasterNotRunningException,
    ZooKeeperConnectionException, IOException {
  final byte[] toBeAssigned = getRegionName(regionName);
  executeCallable(new MasterCallable<Void>(getConnection()) {
    @Override
    public Void call(int callTimeout) throws ServiceException {
      PayloadCarryingRpcController controller = rpcControllerFactory.newController();
      controller.setCallTimeout(callTimeout);
      // Hard to know the table name, at least check if meta
      if (isMetaRegion(regionName)) {
        controller.setPriority(TableName.META_TABLE_NAME);
      }

      AssignRegionRequest request =
        RequestConverter.buildAssignRegionRequest(toBeAssigned);
      master.assignRegion(controller,request);
      return null;
    }
  });
}

/**
 * Unassign a region from current hosting regionserver.  Region will then be
 * assigned to a regionserver chosen at random.  Region could be reassigned
 * back to the same server.  Use {@link #move(byte[], byte[])} if you want
 * to control the region movement.
 * @param regionName Region to unassign. Will clear any existing RegionPlan
 * if one found.
 * @param force If true, force unassign (Will remove region from
 * regions-in-transition too if present. If results in double assignment
 * use hbck -fix to resolve. To be used by experts).
 * @throws MasterNotRunningException
 * @throws ZooKeeperConnectionException
 * @throws IOException
 */
@Override
public void unassign(final byte [] regionName, final boolean force)
throws MasterNotRunningException, ZooKeeperConnectionException, IOException {
  final byte[] toBeUnassigned = getRegionName(regionName);
  executeCallable(new MasterCallable<Void>(getConnection()) {
    @Override
    public Void call(int callTimeout) throws ServiceException {
      PayloadCarryingRpcController controller = rpcControllerFactory.newController();
      controller.setCallTimeout(callTimeout);
      // Hard to know the table name, at least check if meta
      if (isMetaRegion(regionName)) {
        controller.setPriority(TableName.META_TABLE_NAME);
      }
      UnassignRegionRequest request =
        RequestConverter.buildUnassignRegionRequest(toBeUnassigned, force);
      master.unassignRegion(controller, request);
      return null;
    }
  });
}

/**
 * Offline specified region from master's in-memory state. It will not attempt to reassign the
 * region as in unassign. This API can be used when a region not served by any region server and
 * still online as per Master's in memory state. If this API is incorrectly used on active region
 * then master will loose track of that region.
 *
 * This is a special method that should be used by experts or hbck.
 *
 * @param regionName
 *          Region to offline.
 * @throws IOException
 */
@Override
public void offline(final byte [] regionName)
throws IOException {
  executeCallable(new MasterCallable<Void>(getConnection()) {
    @Override
    public Void call(int callTimeout) throws ServiceException {
      PayloadCarryingRpcController controller = rpcControllerFactory.newController();
      controller.setCallTimeout(callTimeout);
      // Hard to know the table name, at least check if meta
      if (isMetaRegion(regionName)) {
        controller.setPriority(TableName.META_TABLE_NAME);
      }
      master.offlineRegion(controller, RequestConverter.buildOfflineRegionRequest(regionName));
      return null;
    }
  });
}

/**
 * Turn the load balancer on or off.
 * @param on If true, enable balancer. If false, disable balancer.
 * @param synchronous If true, it waits until current balance() call, if outstanding, to return.
 * @return Previous balancer value
 */
@Override
public boolean setBalancerRunning(final boolean on, final boolean synchronous)
throws IOException {
  return executeCallable(new MasterCallable<Boolean>(getConnection()) {
    @Override
    public Boolean call(int callTimeout) throws ServiceException {
      PayloadCarryingRpcController controller = rpcControllerFactory.newController();
      controller.setCallTimeout(callTimeout);

      SetBalancerRunningRequest req =
          RequestConverter.buildSetBalancerRunningRequest(on, synchronous);
      return master.setBalancerRunning(controller, req).getPrevBalanceValue();
    }
  });
}

/**
 * Modify an existing table, more IRB friendly version.
 * Asynchronous operation.  This means that it may be a while before your
 * schema change is updated across all of the table.
 *
 * @param tableName name of table.
 * @param htd modified description of the table
 * @throws IOException if a remote or network exception occurs
 */
@Override
public void modifyTable(final TableName tableName, final HTableDescriptor htd)
throws IOException {
  if (!tableName.equals(htd.getTableName())) {
    throw new IllegalArgumentException("the specified table name '" + tableName +
      "' doesn't match with the HTD one: " + htd.getTableName());
  }

  executeCallable(new MasterCallable<Void>(getConnection()) {
    @Override
    public Void call(int callTimeout) throws ServiceException {
      PayloadCarryingRpcController controller = rpcControllerFactory.newController();
      controller.setCallTimeout(callTimeout);
      controller.setPriority(tableName);
      ModifyTableRequest request = RequestConverter.buildModifyTableRequest(
        tableName, htd, ng.getNonceGroup(), ng.newNonce());
      master.modifyTable(controller, request);
      return null;
    }
  });
}

/**
 * Stop the designated regionserver
 * @param hostnamePort Hostname and port delimited by a <code>:</code> as in
 * <code>example.org:1234</code>
 * @throws IOException if a remote or network exception occurs
 */
@Override
public synchronized void stopRegionServer(final String hostnamePort)
throws IOException {
  String hostname = Addressing.parseHostname(hostnamePort);
  int port = Addressing.parsePort(hostnamePort);
  AdminService.BlockingInterface admin =
    this.connection.getAdmin(ServerName.valueOf(hostname, port, 0));
  StopServerRequest request = RequestConverter.buildStopServerRequest(
    "Called by admin client " + this.connection.toString());
  PayloadCarryingRpcController controller = rpcControllerFactory.newController();

  controller.setPriority(HConstants.HIGH_QOS);
  try {
    // TODO: this does not do retries, it should. Set priority and timeout in controller
    admin.stopServer(controller, request);
  } catch (ServiceException se) {
    throw ProtobufUtil.getRemoteException(se);
  }
}

/**
 * Create a new namespace
 * @param descriptor descriptor which describes the new namespace
 * @throws IOException
 */
@Override
public void createNamespace(final NamespaceDescriptor descriptor) throws IOException {
  executeCallable(new MasterCallable<Void>(getConnection()) {
    @Override
    public Void call(int callTimeout) throws Exception {
      PayloadCarryingRpcController controller = rpcControllerFactory.newController();
      controller.setCallTimeout(callTimeout);
      // TODO: set priority based on NS?
      master.createNamespace(controller,
        CreateNamespaceRequest.newBuilder()
          .setNamespaceDescriptor(ProtobufUtil
            .toProtoNamespaceDescriptor(descriptor)).build()
      );
      return null;
    }
  });
}

/**
 * List available namespace descriptors
 * @return List of descriptors
 * @throws IOException
 */
@Override
public NamespaceDescriptor[] listNamespaceDescriptors() throws IOException {
  return
      executeCallable(new MasterCallable<NamespaceDescriptor[]>(getConnection()) {
        @Override
        public NamespaceDescriptor[] call(int callTimeout) throws Exception {
          PayloadCarryingRpcController controller = rpcControllerFactory.newController();
          controller.setCallTimeout(callTimeout);
          List<HBaseProtos.NamespaceDescriptor> list =
              master.listNamespaceDescriptors(controller,
                ListNamespaceDescriptorsRequest.newBuilder().build())
              .getNamespaceDescriptorList();
          NamespaceDescriptor[] res = new NamespaceDescriptor[list.size()];
          for(int i = 0; i < list.size(); i++) {
            res[i] = ProtobufUtil.toNamespaceDescriptor(list.get(i));
          }
          return res;
        }
      });
}

/**
 * List procedures
 * @return procedure list
 * @throws IOException
 */
@Override
public ProcedureInfo[] listProcedures() throws IOException {
  return
      executeCallable(new MasterCallable<ProcedureInfo[]>(getConnection()) {
        @Override
        public ProcedureInfo[] call(int callTimeout) throws Exception {
          PayloadCarryingRpcController controller = rpcControllerFactory.newController();
          controller.setCallTimeout(callTimeout);
          List<ProcedureProtos.Procedure> procList = master.listProcedures(
            controller, ListProceduresRequest.newBuilder().build()).getProcedureList();
          ProcedureInfo[] procInfoList = new ProcedureInfo[procList.size()];
          for (int i = 0; i < procList.size(); i++) {
            procInfoList[i] = ProcedureInfo.convert(procList.get(i));
          }
          return procInfoList;
        }
      });
}

/**
 * Get list of table descriptors by namespace
 * @param name namespace name
 * @return A descriptor
 * @throws IOException
 */
@Override
public HTableDescriptor[] listTableDescriptorsByNamespace(final String name) throws IOException {
  return
      executeCallable(new MasterCallable<HTableDescriptor[]>(getConnection()) {
        @Override
        public HTableDescriptor[] call(int callTimeout) throws Exception {
          PayloadCarryingRpcController controller = rpcControllerFactory.newController();
          controller.setCallTimeout(callTimeout);
          List<TableSchema> list =
              master.listTableDescriptorsByNamespace(controller,
                ListTableDescriptorsByNamespaceRequest.newBuilder().setNamespaceName(name)
                .build()).getTableSchemaList();
          HTableDescriptor[] res = new HTableDescriptor[list.size()];
          for(int i=0; i < list.size(); i++) {

            res[i] = HTableDescriptor.convert(list.get(i));
          }
          return res;
        }
      });
}

/**
 * Get list of table names by namespace
 * @param name namespace name
 * @return The list of table names in the namespace
 * @throws IOException
 */
@Override
public TableName[] listTableNamesByNamespace(final String name) throws IOException {
  return
      executeCallable(new MasterCallable<TableName[]>(getConnection()) {
        @Override
        public TableName[] call(int callTimeout) throws Exception {
          PayloadCarryingRpcController controller = rpcControllerFactory.newController();
          controller.setCallTimeout(callTimeout);
          List<HBaseProtos.TableName> tableNames =
            master.listTableNamesByNamespace(controller, ListTableNamesByNamespaceRequest.
              newBuilder().setNamespaceName(name).build())
              .getTableNameList();
          TableName[] result = new TableName[tableNames.size()];
          for (int i = 0; i < tableNames.size(); i++) {
            result[i] = ProtobufUtil.toTableName(tableNames.get(i));
          }
          return result;
        }
      });
}

/**
 * {@inheritDoc}
 */
@Override
public CompactionState getCompactionStateForRegion(final byte[] regionName)
throws IOException {
  try {
    Pair<HRegionInfo, ServerName> regionServerPair = getRegion(regionName);
    if (regionServerPair == null) {
      throw new IllegalArgumentException("Invalid region: " + Bytes.toStringBinary(regionName));
    }
    if (regionServerPair.getSecond() == null) {
      throw new NoServerForRegionException(Bytes.toStringBinary(regionName));
    }
    ServerName sn = regionServerPair.getSecond();
    AdminService.BlockingInterface admin = this.connection.getAdmin(sn);
    GetRegionInfoRequest request = RequestConverter.buildGetRegionInfoRequest(
      regionServerPair.getFirst().getRegionName(), true);
    PayloadCarryingRpcController controller = rpcControllerFactory.newController();
    // TODO: this does not do retries, it should. Set priority and timeout in controller
    GetRegionInfoResponse response = admin.getRegionInfo(controller, request);
    return response.getCompactionState();
  } catch (ServiceException se) {
    throw ProtobufUtil.getRemoteException(se);
  }
}

/**
 * Take a snapshot without waiting for the server to complete that snapshot (asynchronous)
 * <p>
 * Only a single snapshot should be taken at a time, or results may be undefined.
 * @param snapshot snapshot to take
 * @return response from the server indicating the max time to wait for the snapshot
 * @throws IOException if the snapshot did not succeed or we lose contact with the master.
 * @throws SnapshotCreationException if snapshot creation failed
 * @throws IllegalArgumentException if the snapshot request is formatted incorrectly
 */
@Override
public SnapshotResponse takeSnapshotAsync(SnapshotDescription snapshot) throws IOException,
    SnapshotCreationException {
  ClientSnapshotDescriptionUtils.assertSnapshotRequestIsValid(snapshot);
  final SnapshotRequest request = SnapshotRequest.newBuilder().setSnapshot(snapshot)
      .build();
  // run the snapshot on the master
  return executeCallable(new MasterCallable<SnapshotResponse>(getConnection()) {
    @Override
    public SnapshotResponse call(int callTimeout) throws ServiceException {
      PayloadCarryingRpcController controller = rpcControllerFactory.newController();
      controller.setCallTimeout(callTimeout);
      return master.snapshot(controller, request);
    }
  });
}

/**
 * Check the current state of the specified procedure.
 * <p>
 * There are three possible states:
 * <ol>
 * <li>running - returns <tt>false</tt></li>
 * <li>finished - returns <tt>true</tt></li>
 * <li>finished with error - throws the exception that caused the procedure to fail</li>
 * </ol>
 * <p>
 *
 * @param signature The signature that uniquely identifies a procedure
 * @param instance The instance name of the procedure
 * @param props Property/Value pairs of properties passing to the procedure
 * @return true if the specified procedure is finished successfully, false if it is still running
 * @throws IOException if the specified procedure finished with error
 */
@Override
public boolean isProcedureFinished(String signature, String instance, Map<String, String> props)
    throws IOException {
  final ProcedureDescription.Builder builder = ProcedureDescription.newBuilder();
  builder.setSignature(signature).setInstance(instance);
  for (Entry<String, String> entry : props.entrySet()) {
    NameStringPair pair = NameStringPair.newBuilder().setName(entry.getKey())
        .setValue(entry.getValue()).build();
    builder.addConfiguration(pair);
  }
  final ProcedureDescription desc = builder.build();
  return executeCallable(
      new MasterCallable<IsProcedureDoneResponse>(getConnection()) {
        @Override
        public IsProcedureDoneResponse call(int callTimeout) throws ServiceException {
          PayloadCarryingRpcController controller = rpcControllerFactory.newController();
          controller.setCallTimeout(callTimeout);
          return master.isProcedureDone(controller, IsProcedureDoneRequest
              .newBuilder().setProcedure(desc).build());
        }
      }).getDone();
}

/**
 * Execute Restore/Clone snapshot and wait for the server to complete (asynchronous)
 * <p>
 * Only a single snapshot should be restored at a time, or results may be undefined.
 * @param snapshot snapshot to restore
 * @return response from the server indicating the max time to wait for the snapshot
 * @throws IOException if a remote or network exception occurs
 * @throws RestoreSnapshotException if snapshot failed to be restored
 * @throws IllegalArgumentException if the restore request is formatted incorrectly
 */
private RestoreSnapshotResponse internalRestoreSnapshotAsync(final SnapshotDescription snapshot)
    throws IOException, RestoreSnapshotException {
  ClientSnapshotDescriptionUtils.assertSnapshotRequestIsValid(snapshot);

  final RestoreSnapshotRequest request = RestoreSnapshotRequest.newBuilder().setSnapshot(snapshot)
      .build();

  // run the snapshot restore on the master
  return executeCallable(new MasterCallable<RestoreSnapshotResponse>(getConnection()) {
    @Override
    public RestoreSnapshotResponse call(int callTimeout) throws ServiceException {
      PayloadCarryingRpcController controller = rpcControllerFactory.newController();
      controller.setCallTimeout(callTimeout);
      return master.restoreSnapshot(controller, request);
    }
  });
}

/**
 * Delete an existing snapshot.
 * @param snapshotName name of the snapshot
 * @throws IOException if a remote or network exception occurs
 */
@Override
public void deleteSnapshot(final String snapshotName) throws IOException {
  // make sure the snapshot is possibly valid
  TableName.isLegalFullyQualifiedTableName(Bytes.toBytes(snapshotName));
  // do the delete
  executeCallable(new MasterCallable<Void>(getConnection()) {
    @Override
    public Void call(int callTimeout) throws ServiceException {
      PayloadCarryingRpcController controller = rpcControllerFactory.newController();
      controller.setCallTimeout(callTimeout);
      master.deleteSnapshot(controller,
        DeleteSnapshotRequest.newBuilder().
          setSnapshot(SnapshotDescription.newBuilder().setName(snapshotName).build()).build()
      );
      return null;
    }
  });
}

@Override
public long getLastMajorCompactionTimestampForRegion(final byte[] regionName) throws IOException {
  return executeCallable(new MasterCallable<Long>(getConnection()) {
    @Override
    public Long call(int callTimeout) throws ServiceException {
      PayloadCarryingRpcController controller = rpcControllerFactory.newController();
      controller.setCallTimeout(callTimeout);
      MajorCompactionTimestampForRegionRequest req =
          MajorCompactionTimestampForRegionRequest
              .newBuilder()
              .setRegion(
                RequestConverter
                    .buildRegionSpecifier(RegionSpecifierType.REGION_NAME, regionName)).build();
      return master.getLastMajorCompactionTimestampForRegion(controller, req)
          .getCompactionTimestamp();
    }
  });
}

@Override
public List<SecurityCapability> getSecurityCapabilities() throws IOException {
  try {
    return executeCallable(new MasterCallable<List<SecurityCapability>>(getConnection()) {
      @Override
      public List<SecurityCapability> call(int callTimeout) throws ServiceException {
        PayloadCarryingRpcController controller = rpcControllerFactory.newController();
        controller.setCallTimeout(callTimeout);
        SecurityCapabilitiesRequest req = SecurityCapabilitiesRequest.newBuilder().build();
        return ProtobufUtil.toSecurityCapabilityList(
          master.getSecurityCapabilities(controller, req).getCapabilitiesList());
      }
    });
  } catch (IOException e) {
    if (e instanceof RemoteException) {
      e = ((RemoteException)e).unwrapRemoteException();
    }
    throw e;
  }
}

/**
* {@inheritDoc}
* @deprecated Use reversed scan instead.
*/
@Override
@Deprecated
public Result getRowOrBefore(final byte[] row, final byte[] family)
    throws IOException {
  RegionServerCallable<Result> callable = new RegionServerCallable<Result>(this.connection,
      tableName, row) {
    @Override
   public Result call(int callTimeout) throws IOException {
      PayloadCarryingRpcController controller = rpcControllerFactory.newController();
      controller.setPriority(tableName);
      controller.setCallTimeout(callTimeout);
      ClientProtos.GetRequest request = RequestConverter.buildGetRowOrBeforeRequest(
          getLocation().getRegionInfo().getRegionName(), row, family);
      try {
        ClientProtos.GetResponse response = getStub().get(controller, request);
        if (!response.hasResult()) return null;
        return ProtobufUtil.toResult(response.getResult());
      } catch (ServiceException se) {
        throw ProtobufUtil.getRemoteException(se);
      }
    }
  };
  return rpcCallerFactory.<Result>newCaller().callWithRetries(callable, this.operationTimeout);
}

/**
 * {@inheritDoc}
 */
@Override
public boolean checkAndPut(final byte [] row,
    final byte [] family, final byte [] qualifier, final byte [] value,
    final Put put)
throws IOException {
  RegionServerCallable<Boolean> callable =
    new RegionServerCallable<Boolean>(connection, getName(), row) {
      @Override
      public Boolean call(int callTimeout) throws IOException {
        PayloadCarryingRpcController controller = rpcControllerFactory.newController();
        controller.setPriority(tableName);
        controller.setCallTimeout(callTimeout);
        try {
          MutateRequest request = RequestConverter.buildMutateRequest(
              getLocation().getRegionInfo().getRegionName(), row, family, qualifier,
              new BinaryComparator(value), CompareType.EQUAL, put);
          MutateResponse response = getStub().mutate(controller, request);
          return Boolean.valueOf(response.getProcessed());
        } catch (ServiceException se) {
          throw ProtobufUtil.getRemoteException(se);
        }
      }
    };
  return rpcCallerFactory.<Boolean> newCaller().callWithRetries(callable, this.operationTimeout);
}

/**
 * {@inheritDoc}
 */
@Override
public boolean checkAndDelete(final byte [] row,
    final byte [] family, final byte [] qualifier, final byte [] value,
    final Delete delete)
throws IOException {
  RegionServerCallable<Boolean> callable =
    new RegionServerCallable<Boolean>(connection, getName(), row) {
      @Override
      public Boolean call(int callTimeout) throws IOException {
        PayloadCarryingRpcController controller = rpcControllerFactory.newController();
        controller.setPriority(tableName);
        controller.setCallTimeout(callTimeout);
        try {
          MutateRequest request = RequestConverter.buildMutateRequest(
            getLocation().getRegionInfo().getRegionName(), row, family, qualifier,
              new BinaryComparator(value), CompareType.EQUAL, delete);
          MutateResponse response = getStub().mutate(controller, request);
          return Boolean.valueOf(response.getProcessed());
        } catch (ServiceException se) {
          throw ProtobufUtil.getRemoteException(se);
        }
      }
    };
  return rpcCallerFactory.<Boolean> newCaller().callWithRetries(callable, this.operationTimeout);
}

/**
 * A helper to replicate a list of WAL entries using admin protocol.
 *
 * @param admin
 * @param entries
 * @throws java.io.IOException
 */
public static void replicateWALEntry(final AdminService.BlockingInterface admin,
    final Entry[] entries) throws IOException {
  Pair<AdminProtos.ReplicateWALEntryRequest, CellScanner> p =
    buildReplicateWALEntryRequest(entries, null);
  PayloadCarryingRpcController controller = new PayloadCarryingRpcController(p.getSecond());
  try {
    admin.replicateWALEntry(controller, p.getFirst());
  } catch (ServiceException se) {
    throw ProtobufUtil.getRemoteException(se);
  }
}

/**
 * Sends an CLOSE RPC to the specified server to close the specified region.
 * <p>
 * A region server could reject the close request because it either does not
 * have the specified region or the region is being split.
 * @param server server to open a region
 * @param region region to open
 * @param versionOfClosingNode
 *   the version of znode to compare when RS transitions the znode from
 *   CLOSING state.
 * @param dest - if the region is moved to another server, the destination server. null otherwise.
 * @return true if server acknowledged close, false if not
 * @throws IOException
 */
public boolean sendRegionClose(ServerName server, HRegionInfo region,
  int versionOfClosingNode, ServerName dest, boolean transitionInZK) throws IOException {
  if (server == null) throw new NullPointerException("Passed server is null");
  AdminService.BlockingInterface admin = getRsAdmin(server);
  if (admin == null) {
    throw new IOException("Attempting to send CLOSE RPC to server " +
      server.toString() + " for region " +
      region.getRegionNameAsString() +
      " failed because no RPC connection found to this server");
  }
  PayloadCarryingRpcController controller = newRpcController();
  return ProtobufUtil.closeRegion(controller, admin, server, region.getRegionName(),
    versionOfClosingNode, dest, transitionInZK);
}

/**
 * Sends a WARMUP RPC to the specified server to warmup the specified region.
 * <p>
 * A region server could reject the close request because it either does not
 * have the specified region or the region is being split.
 * @param server server to warmup a region
 * @param region region to  warmup
 */
public void sendRegionWarmup(ServerName server,
    HRegionInfo region) {
  if (server == null) return;
  try {
    AdminService.BlockingInterface admin = getRsAdmin(server);
    PayloadCarryingRpcController controller = newRpcController();
    ProtobufUtil.warmupRegion(controller, admin, region);
  } catch (IOException e) {
    LOG.error("Received exception in RPC for warmup server:" +
      server + "region: " + region +
      "exception: " + e);
  }
}

/**
 * Contacts a region server and waits up to timeout ms
 * to close the region.  This bypasses the active hmaster.
 */
public static void closeRegionSilentlyAndWait(ClusterConnection connection,
  ServerName server, HRegionInfo region, long timeout) throws IOException, InterruptedException {
  AdminService.BlockingInterface rs = connection.getAdmin(server);
  PayloadCarryingRpcController controller = connection.getRpcControllerFactory().newController();
  try {
    ProtobufUtil.closeRegion(controller, rs, server, region.getRegionName(), false);
  } catch (IOException e) {
    LOG.warn("Exception when closing region: " + region.getRegionNameAsString(), e);
  }
  long expiration = timeout + System.currentTimeMillis();
  while (System.currentTimeMillis() < expiration) {
    try {
      HRegionInfo rsRegion =
        ProtobufUtil.getRegionInfo(controller, rs, region.getRegionName());
      if (rsRegion == null) return;
    } catch (IOException ioe) {
      if (ioe instanceof NotServingRegionException) // no need to retry again
        return;
      LOG.warn("Exception when retrieving regioninfo from: "
        + region.getRegionNameAsString(), ioe);
    }
    Thread.sleep(1000);
  }
  throw new IOException("Region " + region + " failed to close within"
      + " timeout " + timeout);
}

/**
 * Check if a region server is reachable and has the expected start code
 */
public boolean isServerReachable(ServerName server) {
  if (server == null) throw new NullPointerException("Passed server is null");


  RetryCounter retryCounter = pingRetryCounterFactory.create();
  while (retryCounter.shouldRetry()) {
    synchronized (this.onlineServers) {
      if (this.deadservers.isDeadServer(server)) {
        return false;
      }
    }
    try {
      PayloadCarryingRpcController controller = newRpcController();
      AdminService.BlockingInterface admin = getRsAdmin(server);
      if (admin != null) {
        ServerInfo info = ProtobufUtil.getServerInfo(controller, admin);
        return info != null && info.hasServerName()
          && server.getStartcode() == info.getServerName().getStartCode();
      }
    } catch (IOException ioe) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Couldn't reach " + server + ", try=" + retryCounter.getAttemptTimes() + " of "
            + retryCounter.getMaxAttempts(), ioe);
      }
      try {
        retryCounter.sleepUntilNextRetry();
      } catch(InterruptedException ie) {
        Thread.currentThread().interrupt();
        break;
      }
    }
  }
  return false;
}

private void addResult(final MutateResponse.Builder builder,
    final Result result, final PayloadCarryingRpcController rpcc) {
  if (result == null) return;
  if (isClientCellBlockSupport()) {
    builder.setResult(ProtobufUtil.toResultNoData(result));
    rpcc.setCellScanner(result.cellScanner());
  } else {
    ClientProtos.Result pbr = ProtobufUtil.toResult(result);
    builder.setResult(pbr);
  }
}