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

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

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

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

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

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

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

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

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

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

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

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

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

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;
}

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

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

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

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

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

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;
  }
}

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;
}

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

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

/**
 * Tests the case where all replicas of a region throw an exception. It should not cause a hang
 * but the exception should propagate to the client
 */
@Test (timeout = 30000)
public void testExceptionsFromReplicasArePropagated() throws IOException {
  scan.setConsistency(Consistency.TIMELINE);

  // Mock a caller which calls the callable for ScannerCallableWithReplicas,
  // but throws an exception for the actual scanner calls via callWithRetries.
  rpcFactory = new MockRpcRetryingCallerFactory(conf);
  conf.set(RpcRetryingCallerFactory.CUSTOM_CALLER_CONF_KEY,
    MockRpcRetryingCallerFactory.class.getName());

  // mock 3 replica locations
  when(clusterConn.locateRegion((TableName)any(), (byte[])any(), anyBoolean(),
    anyBoolean(), anyInt())).thenReturn(new RegionLocations(null, null, null));

  try (MockClientScanner scanner = new MockClientScanner(conf, scan, TableName.valueOf(name.getMethodName()),
    clusterConn, rpcFactory, new RpcControllerFactory(conf), pool, Integer.MAX_VALUE)) {
    Iterator<Result> iter = scanner.iterator();
    while (iter.hasNext()) {
      iter.next();
    }
    fail("Should have failed with RetriesExhaustedException");
  } catch (RuntimeException expected) {
    assertThat(expected.getCause(), instanceOf(RetriesExhaustedException.class));
  }
}

/**
 * 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);
  // puts need to track errors globally due to how the APIs currently work.
  ap = new AsyncProcess(connection, configuration, pool, rpcCallerFactory, true, rpcControllerFactory);
  multiAp = this.connection.getAsyncProcess();

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

static RegionServerCallable<Result[]> getSmallScanCallable(
    final Scan sc, HConnection connection, TableName table,
    byte[] localStartKey, final int cacheNum, final RpcControllerFactory rpcControllerFactory) {
  sc.setStartRow(localStartKey);
  RegionServerCallable<Result[]> callable = new RegionServerCallable<Result[]>(
      connection, table, sc.getStartRow()) {
    public Result[] call(int callTimeout) throws IOException {
      ScanRequest request = RequestConverter.buildScanRequest(getLocation()
          .getRegionInfo().getRegionName(), sc, cacheNum, true);
      PayloadCarryingRpcController controller = rpcControllerFactory.newController();
      controller.setPriority(getTableName());
      controller.setCallTimeout(callTimeout);
      try {
        ScanResponse response = getStub().scan(controller, request);
        return ResponseConverter.getResults(controller.cellScanner(), response);
      } catch (ServiceException se) {
        throw ProtobufUtil.getRemoteException(se);
      }
    }
  };
  return callable;
}

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

/**
 * Create a sink for WAL log entries replay
 * @param conf
 * @param tableName
 * @param conn
 * @throws IOException
 */
public WALEditsReplaySink(Configuration conf, TableName tableName, HConnection conn)
    throws IOException {
  this.conf = conf;
  this.metrics = new MetricsWALEditsReplay();
  this.conn = conn;
  this.tableName = tableName;
  this.skipErrors = conf.getBoolean(HConstants.HREGION_EDITS_REPLAY_SKIP_ERRORS,
    HConstants.DEFAULT_HREGION_EDITS_REPLAY_SKIP_ERRORS);
  // a single replay operation time out and default is 60 seconds
  this.replayTimeout = conf.getInt("hbase.regionserver.logreplay.timeout", 60000);
  this.rpcControllerFactory = RpcControllerFactory.instantiate(conf);
}

public RegionReplicaFlushHandler(Server server, ClusterConnection connection,
    RpcRetryingCallerFactory rpcRetryingCallerFactory, RpcControllerFactory rpcControllerFactory,
    int operationTimeout, HRegion region) {
  super(server, EventType.RS_REGION_REPLICA_FLUSH);
  this.connection = connection;
  this.rpcRetryingCallerFactory = rpcRetryingCallerFactory;
  this.rpcControllerFactory = rpcControllerFactory;
  this.operationTimeout = operationTimeout;
  this.region = region;
}

@Test
public void testFallbackToDefaultRpcControllerFactory() {
  Configuration conf = new Configuration(UTIL.getConfiguration());
  conf.set(RpcControllerFactory.CUSTOM_CONTROLLER_CONF_KEY, "foo.bar.Baz");

  // Should not fail
  RpcControllerFactory factory = RpcControllerFactory.instantiate(conf);
  assertNotNull(factory);
  assertEquals(factory.getClass(), RpcControllerFactory.class);
}

/**
 *
 * @param connection
 * @param tableName
 * @param scan
 * @param scanMetrics
 * @param id the replicaId
 */
public ScannerCallable (ClusterConnection connection, TableName tableName, Scan scan,
    ScanMetrics scanMetrics, RpcControllerFactory rpcControllerFactory, int id) {
  super(connection, tableName, scan.getStartRow());
  this.id = id;
  this.cConnection = connection;
  this.scan = scan;
  this.scanMetrics = scanMetrics;
  Configuration conf = connection.getConfiguration();
  logScannerActivity = conf.getBoolean(LOG_SCANNER_ACTIVITY, false);
  logCutOffLatency = conf.getInt(LOG_SCANNER_LATENCY_CUTOFF, 1000);
  this.controllerFactory = rpcControllerFactory;
}

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

/**
 * constructor
 * @param conf Configuration object
 * @param managed If true, does not do full shutdown on close; i.e. cleanup of connection
 * to zk and shutdown of all services; we just close down the resources this connection was
 * responsible for and decrement usage counters.  It is up to the caller to do the full
 * cleanup.  It is set when we want have connection sharing going on -- reuse of zk connection,
 * and cached region locations, established regionserver connections, etc.  When connections
 * are shared, we have reference counting going on and will only do full cleanup when no more
 * users of an HConnectionImplementation instance.
 */
HConnectionImplementation(Configuration conf, boolean managed,
    ExecutorService pool, User user) throws IOException {
  this(conf);
  this.user = user;
  this.batchPool = pool;
  this.managed = managed;
  this.registry = setupRegistry();
  retrieveClusterId();

  this.rpcClient = RpcClientFactory.createClient(this.conf, this.clusterId, this.metrics);
  this.rpcControllerFactory = RpcControllerFactory.instantiate(conf);

  // Do we publish the status?
  boolean shouldListen = conf.getBoolean(HConstants.STATUS_PUBLISHED,
      HConstants.STATUS_PUBLISHED_DEFAULT);
  Class<? extends ClusterStatusListener.Listener> listenerClass =
      conf.getClass(ClusterStatusListener.STATUS_LISTENER_CLASS,
          ClusterStatusListener.DEFAULT_STATUS_LISTENER_CLASS,
          ClusterStatusListener.Listener.class);
  if (shouldListen) {
    if (listenerClass == null) {
      LOG.warn(HConstants.STATUS_PUBLISHED + " is true, but " +
          ClusterStatusListener.STATUS_LISTENER_CLASS + " is not set - not listening status");
    } else {
      clusterStatusListener = new ClusterStatusListener(
          new ClusterStatusListener.DeadServerHandler() {
            @Override
            public void newDead(ServerName sn) {
              clearCaches(sn);
              rpcClient.cancelConnections(sn);
            }
          }, conf, listenerClass);
    }
  }
}

/**
 * @deprecated use
 *             {@link #ReversedScannerCallable(ClusterConnection, TableName, Scan, ScanMetrics, byte[], RpcControllerFactory )}
 */
@Deprecated
public ReversedScannerCallable(ClusterConnection connection, TableName tableName,
    Scan scan, ScanMetrics scanMetrics, byte[] locateStartRow) {
  this(connection, tableName, scan, scanMetrics, locateStartRow, RpcControllerFactory
      .instantiate(connection.getConfiguration()));
}

/**
 * Creates an object to access a HBase table.
 * Used by HBase internally.  DO NOT USE. See {@link ConnectionFactory} class comment for how to
 * get a {@link Table} instance (use {@link Table} instead of {@link HTable}).
 * @param tableName Name of the table.
 * @param connection HConnection to be used.
 * @param pool ExecutorService to be used.
 * @throws IOException if a remote or network exception occurs
 */
@InterfaceAudience.Private
public HTable(TableName tableName, final ClusterConnection connection,
    final ConnectionConfiguration tableConfig,
    final RpcRetryingCallerFactory rpcCallerFactory,
    final RpcControllerFactory rpcControllerFactory,
    final ExecutorService pool) throws IOException {
  if (connection == null || connection.isClosed()) {
    throw new IllegalArgumentException("Connection is null or closed.");
  }
  this.tableName = tableName;
  this.cleanupConnectionOnClose = false;
  this.connection = connection;
  this.configuration = connection.getConfiguration();
  this.connConfiguration = tableConfig;
  this.pool = pool;
  if (pool == null) {
    this.pool = getDefaultExecutor(this.configuration);
    this.cleanupPoolOnClose = true;
  } else {
    this.cleanupPoolOnClose = false;
  }

  this.rpcCallerFactory = rpcCallerFactory;
  this.rpcControllerFactory = rpcControllerFactory;

  this.finishSetup();
}

public FlushRegionCallable(ClusterConnection connection,
    RpcControllerFactory rpcControllerFactory, TableName tableName, byte[] regionName,
    byte[] regionStartKey, boolean writeFlushWalMarker) {
  super(connection, rpcControllerFactory, tableName, regionStartKey);
  this.regionName = regionName;
  this.writeFlushWalMarker = writeFlushWalMarker;
}

public ScannerCallableWithReplicas getCallable(ClusterConnection connection, TableName table,
    Scan scan, ScanMetrics scanMetrics, byte[] localStartKey, int cacheNum,
    RpcControllerFactory controllerFactory, ExecutorService pool, int primaryOperationTimeout,
    int retries, int scannerTimeout, Configuration conf, RpcRetryingCaller<Result[]> caller) {
  scan.setStartRow(localStartKey);
  SmallScannerCallable s = new SmallScannerCallable(
    connection, table, scan, scanMetrics, controllerFactory, cacheNum, 0);
  ScannerCallableWithReplicas scannerCallableWithReplicas =
      new ScannerCallableWithReplicas(table, connection,
          s, pool, primaryOperationTimeout, scan, retries,
          scannerTimeout, cacheNum, conf, caller);
  return scannerCallableWithReplicas;
}

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.ap = new AsyncProcess(conn, conf, pool, rpcCallerFactory, false, rpcControllerFactory);
  this.executor = executor;
  this.maxRetryInQueue = conf.getInt(TABLE_MULTIPLEXER_MAX_RETRIES_IN_QUEUE, 10000);
}

public RegionAdminServiceCallable(ClusterConnection connection,
    RpcControllerFactory rpcControllerFactory, HRegionLocation location,
    TableName tableName, byte[] row, int replicaId) {
  this.connection = connection;
  this.rpcControllerFactory = rpcControllerFactory;
  this.location = location;
  this.tableName = tableName;
  this.row = row;
  this.replicaId = replicaId;
}

public MyAsyncProcess(ClusterConnection hc, Configuration conf, boolean useGlobalErrors,
    @SuppressWarnings("unused") boolean dummy) {
  super(hc, conf, new ThreadPoolExecutor(1, 20, 60, TimeUnit.SECONDS,
          new SynchronousQueue<Runnable>(), new CountingThreadFactory(new AtomicInteger())) {
    @Override
    public void execute(Runnable command) {
      throw new RejectedExecutionException("test under failure");
    }
  },
      new RpcRetryingCallerFactory(conf), useGlobalErrors, new RpcControllerFactory(conf));
}

@Before
@SuppressWarnings("deprecation")
public void setup() throws IOException {
  clusterConn = Mockito.mock(ClusterConnection.class);
  rpcFactory = Mockito.mock(RpcRetryingCallerFactory.class);
  controllerFactory = Mockito.mock(RpcControllerFactory.class);
  pool = Executors.newSingleThreadExecutor();
  scan = new Scan();
  conf = new Configuration();
  Mockito.when(clusterConn.getConfiguration()).thenReturn(conf);
}