import org.apache.hadoop.hbase.ipc.RpcControllerFactory; //导入依赖的package包/类
public RegionReplicaSinkWriter(RegionReplicaOutputSink sink, ClusterConnection connection,
    ExecutorService pool, int operationTimeout) {
  this.sink = sink;
  this.connection = connection;
  this.operationTimeout = operationTimeout;
  this.rpcRetryingCallerFactory
    = RpcRetryingCallerFactory.instantiate(connection.getConfiguration());
  this.rpcControllerFactory = RpcControllerFactory.instantiate(connection.getConfiguration());
  this.pool = pool;

  int nonExistentTableCacheExpiryMs = connection.getConfiguration()
    .getInt("hbase.region.replica.replication.cache.disabledAndDroppedTables.expiryMs", 5000);
  // A cache for non existing tables that have a default expiry of 5 sec. This means that if the
  // table is created again with the same name, we might miss to replicate for that amount of
  // time. But this cache prevents overloading meta requests for every edit from a deleted file.
  disabledAndDroppedTables = CacheBuilder.newBuilder()
    .expireAfterWrite(nonExistentTableCacheExpiryMs, TimeUnit.MILLISECONDS)
    .initialCapacity(10)
    .maximumSize(1000)
    .build();
}
 

import org.apache.hadoop.hbase.ipc.RpcControllerFactory; //导入依赖的package包/类
private void replicateUsingCallable(ClusterConnection connection, Queue<Entry> entries)
    throws IOException, RuntimeException {
  Entry entry;
  while ((entry = entries.poll()) != null) {
    byte[] row = entry.getEdit().getCells().get(0).getRow();
    RegionLocations locations = connection.locateRegion(tableName, row, true, true);
    RegionReplicaReplayCallable callable = new RegionReplicaReplayCallable(connection,
      RpcControllerFactory.instantiate(connection.getConfiguration()),
      table.getName(), locations.getRegionLocation(1),
      locations.getRegionLocation(1).getRegionInfo(), row, Lists.newArrayList(entry),
      new AtomicLong());

    RpcRetryingCallerFactory factory = RpcRetryingCallerFactory.instantiate(
      connection.getConfiguration());
    factory.<ReplicateWALEntryResponse> newCaller().callWithRetries(callable, 10000);
  }
}
 

import org.apache.hadoop.hbase.ipc.RpcControllerFactory; //导入依赖的package包/类
/**
 * 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));
}
 

import org.apache.hadoop.hbase.ipc.RpcControllerFactory; //导入依赖的package包/类
/**
 * Get a Mocked {@link HConnection} that goes with the passed <code>conf</code>
 * configuration instance.  Minimally the mock will return
 * <code>conf</conf> when {@link ClusterConnection#getConfiguration()} is invoked.
 * Be sure to shutdown the connection when done by calling
 * {@link HConnectionManager#deleteConnection(Configuration)} else it
 * will stick around; this is probably not what you want.
 * @param conf configuration
 * @return HConnection object for <code>conf</code>
 * @throws ZooKeeperConnectionException
 */
public static ClusterConnection getMockedConnection(final Configuration conf)
throws ZooKeeperConnectionException {
  HConnectionKey connectionKey = new HConnectionKey(conf);
  synchronized (ConnectionManager.CONNECTION_INSTANCES) {
    HConnectionImplementation connection =
        ConnectionManager.CONNECTION_INSTANCES.get(connectionKey);
    if (connection == null) {
      connection = Mockito.mock(HConnectionImplementation.class);
      Mockito.when(connection.getConfiguration()).thenReturn(conf);
      Mockito.when(connection.getRpcControllerFactory()).thenReturn(
      Mockito.mock(RpcControllerFactory.class));
      // we need a real retrying caller
      RpcRetryingCallerFactory callerFactory = new RpcRetryingCallerFactory(conf);
      Mockito.when(connection.getRpcRetryingCallerFactory()).thenReturn(callerFactory);
      ConnectionManager.CONNECTION_INSTANCES.put(connectionKey, connection);
    }
    return connection;
  }
}
 

import org.apache.hadoop.hbase.ipc.RpcControllerFactory; //导入依赖的package包/类
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());
}
 

import org.apache.hadoop.hbase.ipc.RpcControllerFactory; //导入依赖的package包/类
public RpcRetryingCallerWithReadReplicas(
    RpcControllerFactory rpcControllerFactory, TableName tableName,
    ClusterConnection cConnection, final Get get,
    ExecutorService pool, int retries, int callTimeout,
    int timeBeforeReplicas) {
  this.rpcControllerFactory = rpcControllerFactory;
  this.tableName = tableName;
  this.cConnection = cConnection;
  this.conf = cConnection.getConfiguration();
  this.get = get;
  this.pool = pool;
  this.retries = retries;
  this.callTimeout = callTimeout;
  this.timeBeforeReplicas = timeBeforeReplicas;
  this.rpcRetryingCallerFactory = new RpcRetryingCallerFactory(conf);
}
 

import org.apache.hadoop.hbase.ipc.RpcControllerFactory; //导入依赖的package包/类
BufferedMutatorImpl(ClusterConnection conn, RpcRetryingCallerFactory rpcCallerFactory,
    RpcControllerFactory rpcFactory, BufferedMutatorParams params) {
  if (conn == null || conn.isClosed()) {
    throw new IllegalArgumentException("Connection is null or closed.");
  }

  this.tableName = params.getTableName();
  this.connection = conn;
  this.conf = connection.getConfiguration();
  this.pool = params.getPool();
  this.listener = params.getListener();

  ConnectionConfiguration tableConf = new ConnectionConfiguration(conf);
  this.writeBufferSize = params.getWriteBufferSize() != BufferedMutatorParams.UNSET ?
      params.getWriteBufferSize() : tableConf.getWriteBufferSize();
  this.maxKeyValueSize = params.getMaxKeyValueSize() != BufferedMutatorParams.UNSET ?
      params.getMaxKeyValueSize() : tableConf.getMaxKeyValueSize();

  // puts need to track errors globally due to how the APIs currently work.
  ap = new AsyncProcess(connection, conf, pool, rpcCallerFactory, true, rpcFactory);
}
 

import org.apache.hadoop.hbase.ipc.RpcControllerFactory; //导入依赖的package包/类
/**
 * setup this HTable's parameter based on the passed configuration
 */
private void finishSetup() throws IOException {
  if (connConfiguration == null) {
    connConfiguration = new ConnectionConfiguration(configuration);
  }

  this.operationTimeout = tableName.isSystemTable() ?
      connConfiguration.getMetaOperationTimeout() : connConfiguration.getOperationTimeout();
  this.scannerCaching = connConfiguration.getScannerCaching();
  this.scannerMaxResultSize = connConfiguration.getScannerMaxResultSize();
  if (this.rpcCallerFactory == null) {
    this.rpcCallerFactory = connection.getNewRpcRetryingCallerFactory(configuration);
  }
  if (this.rpcControllerFactory == null) {
    this.rpcControllerFactory = RpcControllerFactory.instantiate(configuration);
  }

  // puts need to track errors globally due to how the APIs currently work.
  multiAp = this.connection.getAsyncProcess();

  this.closed = false;

  this.locator = new HRegionLocator(tableName, connection);
}
 

import org.apache.hadoop.hbase.ipc.RpcControllerFactory; //导入依赖的package包/类
BufferedMutatorImpl(ClusterConnection conn, RpcRetryingCallerFactory rpcCallerFactory,
    RpcControllerFactory rpcFactory, BufferedMutatorParams params) {
  if (conn == null || conn.isClosed()) {
    throw new IllegalArgumentException("Connection is null or closed.");
  }

  this.tableName = params.getTableName();
  this.connection = conn;
  this.conf = connection.getConfiguration();
  this.pool = params.getPool();
  this.listener = params.getListener();

  TableConfiguration tableConf = new TableConfiguration(conf);
  this.writeBufferSize = params.getWriteBufferSize() != BufferedMutatorParams.UNSET ?
      params.getWriteBufferSize() : tableConf.getWriteBufferSize();
  this.maxKeyValueSize = params.getMaxKeyValueSize() != BufferedMutatorParams.UNSET ?
      params.getMaxKeyValueSize() : tableConf.getMaxKeyValueSize();

  // puts need to track errors globally due to how the APIs currently work.
  ap = new AsyncProcess(connection, conf, pool, rpcCallerFactory, true, rpcFactory);
}
 

import org.apache.hadoop.hbase.ipc.RpcControllerFactory; //导入依赖的package包/类
/**
 * setup this HTable's parameter based on the passed configuration
 */
private void finishSetup() throws IOException {
  if (tableConfiguration == null) {
    tableConfiguration = new TableConfiguration(configuration);
  }

  this.operationTimeout = tableName.isSystemTable() ?
      tableConfiguration.getMetaOperationTimeout() : tableConfiguration.getOperationTimeout();
  this.scannerCaching = tableConfiguration.getScannerCaching();

  if (this.rpcCallerFactory == null) {
    this.rpcCallerFactory = connection.getNewRpcRetryingCallerFactory(configuration);
  }
  if (this.rpcControllerFactory == null) {
    this.rpcControllerFactory = RpcControllerFactory.instantiate(configuration);
  }

  // puts need to track errors globally due to how the APIs currently work.
  multiAp = this.connection.getAsyncProcess();

  this.closed = false;
}
 

import org.apache.hadoop.hbase.ipc.RpcControllerFactory; //导入依赖的package包/类
/**
 * setup this HTable's parameter based on the passed configuration
 */
private void finishSetup() throws IOException {
  this.operationTimeout = tableName.isSystemTable() ?
    this.configuration.getInt(HConstants.HBASE_CLIENT_META_OPERATION_TIMEOUT,
      HConstants.DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT):
    this.configuration.getInt(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT,
      HConstants.DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT);
  this.writeBufferSize = this.configuration.getLong(
      "hbase.client.write.buffer", 2097152);
  this.clearBufferOnFail = true;
  this.autoFlush = true;
  this.currentWriteBufferSize = 0;
  this.scannerCaching = this.configuration.getInt(
      HConstants.HBASE_CLIENT_SCANNER_CACHING,
      HConstants.DEFAULT_HBASE_CLIENT_SCANNER_CACHING);

  this.rpcCallerFactory = RpcRetryingCallerFactory.instantiate(configuration);
  this.rpcControllerFactory = RpcControllerFactory.instantiate(configuration);
  ap = new AsyncProcess<Object>(connection, tableName, pool, null,
      configuration, rpcCallerFactory, rpcControllerFactory);

  this.maxKeyValueSize = this.configuration.getInt(
      "hbase.client.keyvalue.maxsize", -1);
  this.closed = false;
}
 

import org.apache.hadoop.hbase.ipc.RpcControllerFactory; //导入依赖的package包/类
static RegionServerCallable<Result[]> getSmallScanCallable(
    final Scan sc, HConnection connection, TableName table, byte[] localStartKey,
    final int cacheNum, final RpcControllerFactory rpcControllerFactory) throws IOException { 
  sc.setStartRow(localStartKey);
  RegionServerCallable<Result[]> callable = new RegionServerCallable<Result[]>(
      connection, table, sc.getStartRow()) {
    public Result[] call() throws IOException {
      ScanRequest request = RequestConverter.buildScanRequest(getLocation()
        .getRegionInfo().getRegionName(), sc, cacheNum, true);
      ScanResponse response = null;
      PayloadCarryingRpcController controller = rpcControllerFactory.newController();
      try {
        controller.setPriority(getTableName());
        response = getStub().scan(controller, request);
        return ResponseConverter.getResults(controller.cellScanner(),
            response);
      } catch (ServiceException se) {
        throw ProtobufUtil.getRemoteException(se);
      }
    }
  };
  return callable;
}
 

import org.apache.hadoop.hbase.ipc.RpcControllerFactory; //导入依赖的package包/类
private void replicateUsingCallable(ClusterConnection connection, Queue<Entry> entries)
    throws IOException, RuntimeException {
  Entry entry;
  while ((entry = entries.poll()) != null) {
    byte[] row = CellUtil.cloneRow(entry.getEdit().getCells().get(0));
    RegionLocations locations = connection.locateRegion(tableName, row, true, true);
    RegionReplicaReplayCallable callable = new RegionReplicaReplayCallable(connection,
      RpcControllerFactory.instantiate(connection.getConfiguration()),
      table.getName(), locations.getRegionLocation(1),
      locations.getRegionLocation(1).getRegionInfo(), row, Lists.newArrayList(entry),
      new AtomicLong());

    RpcRetryingCallerFactory factory = RpcRetryingCallerFactory.instantiate(
      connection.getConfiguration());
    factory.<ReplicateWALEntryResponse> newCaller().callWithRetries(callable, 10000);
  }
}
 

import org.apache.hadoop.hbase.ipc.RpcControllerFactory; //导入依赖的package包/类
/**
 * 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())).
    thenReturn(implementation);
      RpcControllerFactory controllerFactory = Mockito.mock(RpcControllerFactory.class);
      Mockito.when(controllerFactory.newController()).thenReturn(
        Mockito.mock(HBaseRpcController.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));
}
 

import org.apache.hadoop.hbase.ipc.RpcControllerFactory; //导入依赖的package包/类
static TableDescriptor getTableDescriptor(final TableName tableName, Connection connection,
    RpcRetryingCallerFactory rpcCallerFactory, final RpcControllerFactory rpcControllerFactory,
    int operationTimeout, int rpcTimeout) throws IOException {
  if (tableName == null) return null;
  TableDescriptor td =
      executeCallable(new MasterCallable<TableDescriptor>(connection, rpcControllerFactory) {
    @Override
    protected TableDescriptor rpcCall() throws Exception {
      GetTableDescriptorsRequest req =
          RequestConverter.buildGetTableDescriptorsRequest(tableName);
      GetTableDescriptorsResponse htds = master.getTableDescriptors(getRpcController(), req);
      if (!htds.getTableSchemaList().isEmpty()) {
        return ProtobufUtil.toTableDescriptor(htds.getTableSchemaList().get(0));
      }
      return null;
    }
  }, rpcCallerFactory, operationTimeout, rpcTimeout);
  if (td != null) {
    return td;
  }
  throw new TableNotFoundException(tableName.getNameAsString());
}
 

import org.apache.hadoop.hbase.ipc.RpcControllerFactory; //导入依赖的package包/类
public RpcRetryingCallerWithReadReplicas(
    RpcControllerFactory rpcControllerFactory, TableName tableName,
    ClusterConnection cConnection, final Get get,
    ExecutorService pool, int retries, int operationTimeout, int rpcTimeout,
    int timeBeforeReplicas) {
  this.rpcControllerFactory = rpcControllerFactory;
  this.tableName = tableName;
  this.cConnection = cConnection;
  this.conf = cConnection.getConfiguration();
  this.get = get;
  this.pool = pool;
  this.retries = retries;
  this.operationTimeout = operationTimeout;
  this.rpcTimeout = rpcTimeout;
  this.timeBeforeReplicas = timeBeforeReplicas;
  this.rpcRetryingCallerFactory = new RpcRetryingCallerFactory(conf);
}
 

import org.apache.hadoop.hbase.ipc.RpcControllerFactory; //导入依赖的package包/类
public AsyncConnectionImpl(Configuration conf, AsyncRegistry registry, String clusterId,
    User user) {
  this.conf = conf;
  this.user = user;
  this.connConf = new AsyncConnectionConfiguration(conf);
  this.registry = registry;
  this.rpcClient = RpcClientFactory.createClient(conf, clusterId);
  this.rpcControllerFactory = RpcControllerFactory.instantiate(conf);
  this.hostnameCanChange = conf.getBoolean(RESOLVE_HOSTNAME_ON_FAIL_KEY, true);
  this.rpcTimeout =
    (int) Math.min(Integer.MAX_VALUE, TimeUnit.NANOSECONDS.toMillis(connConf.getRpcTimeoutNs()));
  this.locator = new AsyncRegionLocator(this, RETRY_TIMER);
  this.callerFactory = new AsyncRpcRetryingCallerFactory(this, RETRY_TIMER);
  if (conf.getBoolean(CLIENT_NONCES_ENABLED_KEY, true)) {
    nonceGenerator = PerClientRandomNonceGenerator.get();
  } else {
    nonceGenerator = NO_NONCE_GENERATOR;
  }
}
 

import org.apache.hadoop.hbase.ipc.RpcControllerFactory; //导入依赖的package包/类
public FlushWorker(Configuration conf, ClusterConnection conn, HRegionLocation addr,
    HTableMultiplexer htableMultiplexer, int perRegionServerBufferQueueSize,
    ExecutorService pool, ScheduledExecutorService executor) {
  this.addr = addr;
  this.multiplexer = htableMultiplexer;
  this.queue = new LinkedBlockingQueue<>(perRegionServerBufferQueueSize);
  RpcRetryingCallerFactory rpcCallerFactory = RpcRetryingCallerFactory.instantiate(conf);
  RpcControllerFactory rpcControllerFactory = RpcControllerFactory.instantiate(conf);
  this.writeRpcTimeout = conf.getInt(HConstants.HBASE_RPC_WRITE_TIMEOUT_KEY,
      conf.getInt(HConstants.HBASE_RPC_TIMEOUT_KEY,
          HConstants.DEFAULT_HBASE_RPC_TIMEOUT));
  this.operationTimeout = conf.getInt(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT,
      HConstants.DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT);
  this.ap = new AsyncProcess(conn, conf, rpcCallerFactory, false, rpcControllerFactory);
  this.executor = executor;
  this.maxRetryInQueue = conf.getInt(TABLE_MULTIPLEXER_MAX_RETRIES_IN_QUEUE, 10000);
  this.pool = pool;
}
 

import org.apache.hadoop.hbase.ipc.RpcControllerFactory; //导入依赖的package包/类
/**
 * Executes an RPC to the HBase master to fetch its view on the Region sizes.
 */
public static GetSpaceQuotaRegionSizesResponse getMasterRegionSizes(
    Connection conn, RpcControllerFactory factory, RpcRetryingCallerFactory rpcCaller,
    int timeout) throws IOException {
  MasterCallable<GetSpaceQuotaRegionSizesResponse> callable =
      new MasterCallable<GetSpaceQuotaRegionSizesResponse>(conn, factory) {
    @Override
    protected GetSpaceQuotaRegionSizesResponse rpcCall() throws Exception {
      return master.getSpaceQuotaRegionSizes(
          getRpcController(), RequestConverter.buildGetSpaceQuotaRegionSizesRequest());
    }
  };
  RpcRetryingCaller<GetSpaceQuotaRegionSizesResponse> caller = rpcCaller.newCaller();
  try {
    return caller.callWithoutRetries(callable, timeout);
  } finally {
    callable.close();
  }
}
 

import org.apache.hadoop.hbase.ipc.RpcControllerFactory; //导入依赖的package包/类
/**
 * Executes an RPC tot he HBase master to fetch its view on space quotas.
 */
public static GetQuotaStatesResponse getMasterQuotaStates(
    Connection conn, RpcControllerFactory factory, RpcRetryingCallerFactory rpcCaller,
    int timeout) throws IOException {
  MasterCallable<GetQuotaStatesResponse> callable =
      new MasterCallable<GetQuotaStatesResponse>(conn, factory) {
    @Override
    protected GetQuotaStatesResponse rpcCall() throws Exception {
      return master.getQuotaStates(
          getRpcController(), RequestConverter.buildGetQuotaStatesRequest());
    }
  };
  RpcRetryingCaller<GetQuotaStatesResponse> caller = rpcCaller.newCaller();
  try {
    return caller.callWithoutRetries(callable, timeout);
  } finally {
    callable.close();
  }
}