import org.apache.hadoop.hbase.ipc.RpcServerInterface; //导入依赖的package包/类
@Override
public void start(CoprocessorEnvironment env) {
  // if running at region
  if (env instanceof RegionCoprocessorEnvironment) {
    RegionCoprocessorEnvironment regionEnv = (RegionCoprocessorEnvironment)env;
    /* Getting the RpcServer from a RegionCE is wrong. There cannot be an expectation that Region
     is hosted inside a RegionServer. If you need RpcServer, then pass in a RegionServerCE.
     TODO: FIX.
     */
    RegionServerServices rss = ((HasRegionServerServices)regionEnv).getRegionServerServices();
    RpcServerInterface server = rss.getRpcServer();
    SecretManager<?> mgr = ((RpcServer)server).getSecretManager();
    if (mgr instanceof AuthenticationTokenSecretManager) {
      secretManager = (AuthenticationTokenSecretManager)mgr;
    }
  }
}
 

import org.apache.hadoop.hbase.ipc.RpcServerInterface; //导入依赖的package包/类
@Override
protected RpcServerInterface createRpcServer(Server server, Configuration conf,
    RpcSchedulerFactory rpcSchedulerFactory, InetSocketAddress bindAddress, String name)
    throws IOException {
  // RpcServer at HM by default enable ByteBufferPool iff HM having user table region in it
  boolean reservoirEnabled = conf.getBoolean(RESERVOIR_ENABLED_KEY,
      (LoadBalancer.isTablesOnMaster(conf) && !LoadBalancer.isSystemTablesOnlyOnMaster(conf)));
  try {
    return RpcServerFactory.createRpcServer(server, name, getServices(),
        bindAddress, // use final bindAddress for this server.
        conf, rpcSchedulerFactory.create(conf, this, server), reservoirEnabled);
  } catch (BindException be) {
    throw new IOException(be.getMessage() + ". To switch ports use the '"
        + HConstants.MASTER_PORT + "' configuration property.",
        be.getCause() != null ? be.getCause() : be);
  }
}
 

import org.apache.hadoop.hbase.ipc.RpcServerInterface; //导入依赖的package包/类
@Override
public void start(CoprocessorEnvironment env) {
  // if running at region
  if (env instanceof RegionCoprocessorEnvironment) {
    RegionCoprocessorEnvironment regionEnv =
        (RegionCoprocessorEnvironment)env;
    RpcServerInterface server = regionEnv.getRegionServerServices().getRpcServer();
    SecretManager<?> mgr = ((RpcServer)server).getSecretManager();
    if (mgr instanceof AuthenticationTokenSecretManager) {
      secretManager = (AuthenticationTokenSecretManager)mgr;
    }
  }
}
 

import org.apache.hadoop.hbase.ipc.RpcServerInterface; //导入依赖的package包/类
@Test
public void testMultiLimits() throws Exception {
  final TableName name = TableName.valueOf("testMultiLimits");
  Table t = TEST_UTIL.createTable(name, FAMILY);
  TEST_UTIL.loadTable(t, FAMILY, false);

  // Split the table to make sure that the chunking happens accross regions.
  try (final Admin admin = TEST_UTIL.getHBaseAdmin()) {
    admin.split(name);
    TEST_UTIL.waitFor(60000, new Waiter.Predicate<Exception>() {
      @Override
      public boolean evaluate() throws Exception {
        return admin.getTableRegions(name).size() > 1;
      }
    });
  }
  List<Get> gets = new ArrayList<>(MAX_SIZE);

  for (int i = 0; i < MAX_SIZE; i++) {
    gets.add(new Get(HBaseTestingUtility.ROWS[i]));
  }

  RpcServerInterface rpcServer = TEST_UTIL.getHBaseCluster().getRegionServer(0).getRpcServer();
  BaseSource s = rpcServer.getMetrics().getMetricsSource();
  long startingExceptions = METRICS_ASSERT.getCounter("exceptions", s);
  long startingMultiExceptions = METRICS_ASSERT.getCounter("exceptions.multiResponseTooLarge", s);

  Result[] results = t.get(gets);
  assertEquals(MAX_SIZE, results.length);

  // Cells from TEST_UTIL.loadTable have a length of 27.
  // Multiplying by less than that gives an easy lower bound on size.
  // However in reality each kv is being reported as much higher than that.
  METRICS_ASSERT.assertCounterGt("exceptions",
      startingExceptions + ((MAX_SIZE * 25) / MAX_SIZE), s);
  METRICS_ASSERT.assertCounterGt("exceptions.multiResponseTooLarge",
      startingMultiExceptions + ((MAX_SIZE * 25) / MAX_SIZE), s);
}
 

import org.apache.hadoop.hbase.ipc.RpcServerInterface; //导入依赖的package包/类
/**
 * Create a stubbed out RegionServerService, mainly for getting FS.
 * This version is used by TestTokenAuthentication
 */
public RegionServerServices createMockRegionServerService(RpcServerInterface rpc) throws IOException {
  final MockRegionServerServices rss = new MockRegionServerServices(getZooKeeperWatcher());
  rss.setFileSystem(getTestFileSystem());
  rss.setRpcServer(rpc);
  return rss;
}
 

import org.apache.hadoop.hbase.ipc.RpcServerInterface; //导入依赖的package包/类
@Before
public void setupBasicMocks() throws IOException, ServiceException {
  rs = Mockito.mock(HRegionServer.class);
  rpcServices = Mockito.mock(RSRpcServices.class);
  rpcServer = Mockito.mock(RpcServerInterface.class);
  Mockito.doReturn(HBaseConfiguration.create())
    .when(rs).getConfiguration();
  Mockito.doReturn(rpcServices).when(rs).getRSRpcServices();
  Mockito.doReturn(rpcServer).when(rs).getRpcServer();
  Mockito.doReturn(fakeResponse).when(rpcServices).getServerInfo(
    (RpcController)Mockito.any(), (GetServerInfoRequest)Mockito.any());
  // Fake ZKW
  ZooKeeperWatcher zkw = Mockito.mock(ZooKeeperWatcher.class);
  Mockito.doReturn("fakequorum").when(zkw).getQuorum();
  Mockito.doReturn(zkw).when(rs).getZooKeeper();

  // Fake CacheConfig
  LOG.warn("The " + HConstants.HFILE_BLOCK_CACHE_SIZE_KEY + " is set to 0");
  CacheConfig cacheConf = Mockito.mock(CacheConfig.class);
  Mockito.doReturn(null).when(cacheConf).getBlockCache();
  Mockito.doReturn(cacheConf).when(rs).getCacheConfig();
  
  // Fake MasterAddressTracker
  MasterAddressTracker mat = Mockito.mock(MasterAddressTracker.class);
  Mockito.doReturn(fakeMasterAddress).when(mat).getMasterAddress();
  Mockito.doReturn(mat).when(rs).getMasterAddressTracker();

  MetricsRegionServer rms = Mockito.mock(MetricsRegionServer.class);
  Mockito.doReturn(new MetricsRegionServerWrapperStub()).when(rms).getRegionServerWrapper();
  Mockito.doReturn(rms).when(rs).getRegionServerMetrics();

  MetricsHBaseServer ms = Mockito.mock(MetricsHBaseServer.class);
  Mockito.doReturn(new MetricsHBaseServerWrapperStub()).when(ms).getHBaseServerWrapper();
  Mockito.doReturn(ms).when(rpcServer).getMetrics();
}
 

import org.apache.hadoop.hbase.ipc.RpcServerInterface; //导入依赖的package包/类
public org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService.BlockingInterface
    getClient(ServerName serverName) throws IOException {
  // client is trying to reach off-server, so we can't do anything special
  if (!this.serverName.equals(serverName)) {
    return delegate.getClient(serverName);
  }
  // the client is attempting to write to the same regionserver, we can short-circuit to our
  // local regionserver
  final BlockingService blocking = ClientService.newReflectiveBlockingService(this.server);
  final RpcServerInterface rpc = this.server.getRpcServer();

  final MonitoredRPCHandler status =
      TaskMonitor.get().createRPCStatus(Thread.currentThread().getName());
  status.pause("Setting up server-local call");

  final long timestamp = EnvironmentEdgeManager.currentTimeMillis();
  BlockingRpcChannel channel = new BlockingRpcChannel() {

    @Override
    public Message callBlockingMethod(MethodDescriptor method, RpcController controller,
        Message request, Message responsePrototype) throws ServiceException {
      try {
        // we never need a cell-scanner - everything is already fully formed
        return rpc.call(blocking, method, request, null, timestamp, status).getFirst();
      } catch (IOException e) {
        throw new ServiceException(e);
      }
    }
  };
  return ClientService.newBlockingStub(channel);
}
 

import org.apache.hadoop.hbase.ipc.RpcServerInterface; //导入依赖的package包/类
protected RpcServerInterface createRpcServer(Server server, Configuration conf,
    RpcSchedulerFactory rpcSchedulerFactory, InetSocketAddress bindAddress, String name)
    throws IOException {
  boolean reservoirEnabled = conf.getBoolean(RESERVOIR_ENABLED_KEY, true);
  try {
    return RpcServerFactory.createRpcServer(server, name, getServices(),
        bindAddress, // use final bindAddress for this server.
        conf, rpcSchedulerFactory.create(conf, this, server), reservoirEnabled);
  } catch (BindException be) {
    throw new IOException(be.getMessage() + ". To switch ports use the '"
        + HConstants.REGIONSERVER_PORT + "' configuration property.",
        be.getCause() != null ? be.getCause() : be);
  }
}
 

import org.apache.hadoop.hbase.ipc.RpcServerInterface; //导入依赖的package包/类
/**
 * Sets up a RPC Server and a Client. Does a RPC checks the result. If an exception is thrown from
 * the stub, this function will throw root cause of that exception.
 */
private void callRpcService(User clientUser) throws Exception {
  SecurityInfo securityInfoMock = Mockito.mock(SecurityInfo.class);
  Mockito.when(securityInfoMock.getServerPrincipal())
      .thenReturn(HBaseKerberosUtils.KRB_PRINCIPAL);
  SecurityInfo.addInfo("TestProtobufRpcProto", securityInfoMock);

  InetSocketAddress isa = new InetSocketAddress(HOST, 0);

  RpcServerInterface rpcServer = RpcServerFactory.createRpcServer(null, "AbstractTestSecureIPC",
      Lists.newArrayList(new RpcServer.BlockingServiceAndInterface((BlockingService) SERVICE, null)), isa,
      serverConf, new FifoRpcScheduler(serverConf, 1));
  rpcServer.start();
  try (RpcClient rpcClient = RpcClientFactory.createClient(clientConf,
    HConstants.DEFAULT_CLUSTER_ID.toString())) {
    BlockingInterface stub = newBlockingStub(rpcClient, rpcServer.getListenerAddress(),
      clientUser);
    TestThread th1 = new TestThread(stub);
    final Throwable exception[] = new Throwable[1];
    Collections.synchronizedList(new ArrayList<Throwable>());
    Thread.UncaughtExceptionHandler exceptionHandler = new Thread.UncaughtExceptionHandler() {
      @Override
      public void uncaughtException(Thread th, Throwable ex) {
        exception[0] = ex;
      }
    };
    th1.setUncaughtExceptionHandler(exceptionHandler);
    th1.start();
    th1.join();
    if (exception[0] != null) {
      // throw root cause.
      while (exception[0].getCause() != null) {
        exception[0] = exception[0].getCause();
      }
      throw (Exception) exception[0];
    }
  } finally {
    rpcServer.stop();
  }
}
 

import org.apache.hadoop.hbase.ipc.RpcServerInterface; //导入依赖的package包/类
@Test
public void testMultiLimits() throws Exception {
  final TableName tableName = TableName.valueOf(name.getMethodName());
  Table t = TEST_UTIL.createTable(tableName, FAMILY);
  TEST_UTIL.loadTable(t, FAMILY, false);

  // Split the table to make sure that the chunking happens accross regions.
  try (final Admin admin = TEST_UTIL.getAdmin()) {
    admin.split(tableName);
    TEST_UTIL.waitFor(60000, new Waiter.Predicate<Exception>() {
      @Override
      public boolean evaluate() throws Exception {
        return admin.getTableRegions(tableName).size() > 1;
      }
    });
  }
  List<Get> gets = new ArrayList<>(MAX_SIZE);

  for (int i = 0; i < MAX_SIZE; i++) {
    gets.add(new Get(HBaseTestingUtility.ROWS[i]));
  }

  RpcServerInterface rpcServer = TEST_UTIL.getHBaseCluster().getRegionServer(0).getRpcServer();
  BaseSource s = rpcServer.getMetrics().getMetricsSource();
  long startingExceptions = METRICS_ASSERT.getCounter("exceptions", s);
  long startingMultiExceptions = METRICS_ASSERT.getCounter("exceptions.multiResponseTooLarge", s);

  Result[] results = t.get(gets);
  assertEquals(MAX_SIZE, results.length);

  // Cells from TEST_UTIL.loadTable have a length of 27.
  // Multiplying by less than that gives an easy lower bound on size.
  // However in reality each kv is being reported as much higher than that.
  METRICS_ASSERT.assertCounterGt("exceptions",
      startingExceptions + ((MAX_SIZE * 25) / MAX_SIZE), s);
  METRICS_ASSERT.assertCounterGt("exceptions.multiResponseTooLarge",
      startingMultiExceptions + ((MAX_SIZE * 25) / MAX_SIZE), s);
}
 

import org.apache.hadoop.hbase.ipc.RpcServerInterface; //导入依赖的package包/类
/**
 * Create a stubbed out RegionServerService, mainly for getting FS.
 * This version is used by TestTokenAuthentication
 */
public RegionServerServices createMockRegionServerService(RpcServerInterface rpc) throws
    IOException {
  final MockRegionServerServices rss = new MockRegionServerServices(getZooKeeperWatcher());
  rss.setFileSystem(getTestFileSystem());
  rss.setRpcServer(rpc);
  return rss;
}
 

import org.apache.hadoop.hbase.ipc.RpcServerInterface; //导入依赖的package包/类
@Before
public void setupBasicMocks() throws IOException, ServiceException {
  rs = Mockito.mock(HRegionServer.class);
  rpcServices = Mockito.mock(RSRpcServices.class);
  rpcServer = Mockito.mock(RpcServerInterface.class);
  Mockito.doReturn(HBaseConfiguration.create())
    .when(rs).getConfiguration();
  Mockito.doReturn(rpcServices).when(rs).getRSRpcServices();
  Mockito.doReturn(rpcServer).when(rs).getRpcServer();
  Mockito.doReturn(fakeResponse).when(rpcServices).getServerInfo(
    (RpcController)Mockito.any(), (GetServerInfoRequest)Mockito.any());
  // Fake ZKW
  ZKWatcher zkw = Mockito.mock(ZKWatcher.class);
  Mockito.doReturn("fakequorum").when(zkw).getQuorum();
  Mockito.doReturn(zkw).when(rs).getZooKeeper();

  // Fake CacheConfig
  LOG.warn("The " + HConstants.HFILE_BLOCK_CACHE_SIZE_KEY + " is set to 0");
  CacheConfig cacheConf = Mockito.mock(CacheConfig.class);
  Mockito.doReturn(null).when(cacheConf).getBlockCache();
  Mockito.doReturn(cacheConf).when(rs).getCacheConfig();

  // Fake MasterAddressTracker
  MasterAddressTracker mat = Mockito.mock(MasterAddressTracker.class);
  Mockito.doReturn(fakeMasterAddress).when(mat).getMasterAddress();
  Mockito.doReturn(mat).when(rs).getMasterAddressTracker();

  MetricsRegionServer rms = Mockito.mock(MetricsRegionServer.class);
  Mockito.doReturn(new MetricsRegionServerWrapperStub()).when(rms).getRegionServerWrapper();
  Mockito.doReturn(rms).when(rs).getRegionServerMetrics();

  MetricsHBaseServer ms = Mockito.mock(MetricsHBaseServer.class);
  Mockito.doReturn(new MetricsHBaseServerWrapperStub()).when(ms).getHBaseServerWrapper();
  Mockito.doReturn(ms).when(rpcServer).getMetrics();
}
 

import org.apache.hadoop.hbase.ipc.RpcServerInterface; //导入依赖的package包/类
/**
 * Create a stubbed out RegionServerService, mainly for getting FS. 
 * This version is used by TestTokenAuthentication
 */
public RegionServerServices createMockRegionServerService(RpcServerInterface rpc) throws IOException {
  final MockRegionServerServices rss = new MockRegionServerServices(getZooKeeperWatcher());
  rss.setFileSystem(getTestFileSystem());
  rss.setRpcServer(rpc);
  return rss;
}
 

import org.apache.hadoop.hbase.ipc.RpcServerInterface; //导入依赖的package包/类
@Override public RpcServerInterface getRpcServer() {
  return rpcServices.rpcServer;
}
 

import org.apache.hadoop.hbase.ipc.RpcServerInterface; //导入依赖的package包/类
private void callRpcService(Class<? extends RpcClient> rpcImplClass, User clientUser,
                            Configuration clientConf, boolean allowInsecureFallback)
    throws Exception {
  Configuration clientConfCopy = new Configuration(clientConf);
  clientConfCopy.set(RpcClientFactory.CUSTOM_RPC_CLIENT_IMPL_CONF_KEY, rpcImplClass.getName());

  Configuration conf = getSecuredConfiguration();
  conf.setBoolean(RpcServer.FALLBACK_TO_INSECURE_CLIENT_AUTH, allowInsecureFallback);

  SecurityInfo securityInfoMock = Mockito.mock(SecurityInfo.class);
  Mockito.when(securityInfoMock.getServerPrincipal())
      .thenReturn(HBaseKerberosUtils.KRB_PRINCIPAL);
  SecurityInfo.addInfo("TestProtobufRpcProto", securityInfoMock);

  InetSocketAddress isa = new InetSocketAddress(HOST, 0);

  RpcServerInterface rpcServer =
      new RpcServer(null, "AbstractTestSecureIPC",
          Lists.newArrayList(new RpcServer.BlockingServiceAndInterface(SERVICE, null)), isa,
          conf, new FifoRpcScheduler(conf, 1));
  rpcServer.start();
  try (RpcClient rpcClient = RpcClientFactory.createClient(clientConf,
      HConstants.DEFAULT_CLUSTER_ID.toString())) {
    InetSocketAddress address = rpcServer.getListenerAddress();
    if (address == null) {
      throw new IOException("Listener channel is closed");
    }
    BlockingRpcChannel channel =
        rpcClient.createBlockingRpcChannel(

          ServerName.valueOf(address.getHostName(), address.getPort(),
          System.currentTimeMillis()), clientUser, 5000);
    TestRpcServiceProtos.TestProtobufRpcProto.BlockingInterface stub =
        TestRpcServiceProtos.TestProtobufRpcProto.newBlockingStub(channel);
    List<String> results = new ArrayList<String>();
    TestThread th1 = new TestThread(stub, results);
    th1.start();
    th1.join();

  } finally {
    rpcServer.stop();
  }
}
 

import org.apache.hadoop.hbase.ipc.RpcServerInterface; //导入依赖的package包/类
@Test
public void testBlockMultiLimits() throws Exception {
  final TableName name = TableName.valueOf("testBlockMultiLimits");
  HTableDescriptor desc = new HTableDescriptor(name);
  HColumnDescriptor hcd = new HColumnDescriptor(FAMILY);
  hcd.setDataBlockEncoding(DataBlockEncoding.FAST_DIFF);
  desc.addFamily(hcd);
  TEST_UTIL.getHBaseAdmin().createTable(desc);
  Table t = TEST_UTIL.getConnection().getTable(name);

  final HRegionServer regionServer = TEST_UTIL.getHBaseCluster().getRegionServer(0);
  RpcServerInterface rpcServer = regionServer.getRpcServer();
  BaseSource s = rpcServer.getMetrics().getMetricsSource();
  long startingExceptions = METRICS_ASSERT.getCounter("exceptions", s);
  long startingMultiExceptions = METRICS_ASSERT.getCounter("exceptions.multiResponseTooLarge", s);

  byte[] row = Bytes.toBytes("TEST");
  byte[][] cols = new byte[][]{
      Bytes.toBytes("0"), // Get this
      Bytes.toBytes("1"), // Buffer
      Bytes.toBytes("2"), // Buffer
      Bytes.toBytes("3"), // Get This
      Bytes.toBytes("4"), // Buffer
      Bytes.toBytes("5"), // Buffer
  };

  // Set the value size so that one result will be less than the MAX_SIE
  // however the block being reference will be larger than MAX_SIZE.
  // This should cause the regionserver to try and send a result immediately.
  byte[] value = new byte[MAX_SIZE - 100];
  ThreadLocalRandom.current().nextBytes(value);

  for (byte[] col:cols) {
    Put p = new Put(row);
    p.addImmutable(FAMILY, col, value);
    t.put(p);
  }

  // Make sure that a flush happens
  try (final Admin admin = TEST_UTIL.getHBaseAdmin()) {
    admin.flush(name);
    TEST_UTIL.waitFor(60000, new Waiter.Predicate<Exception>() {
      @Override
      public boolean evaluate() throws Exception {
        return regionServer.getOnlineRegions(name).get(0).getMaxFlushedSeqId() > 3;
      }
    });
  }

  List<Get> gets = new ArrayList<>(2);
  Get g0 = new Get(row);
  g0.addColumn(FAMILY, cols[0]);
  gets.add(g0);

  Get g2 = new Get(row);
  g2.addColumn(FAMILY, cols[3]);
  gets.add(g2);

  Result[] results = t.get(gets);
  assertEquals(2, results.length);
  METRICS_ASSERT.assertCounterGt("exceptions", startingExceptions, s);
  METRICS_ASSERT.assertCounterGt("exceptions.multiResponseTooLarge",
      startingMultiExceptions, s);
}
 

import org.apache.hadoop.hbase.ipc.RpcServerInterface; //导入依赖的package包/类
@Override
public RpcServerInterface getRpcServer() {
  // TODO Auto-generated method stub
  return null;
}
 

import org.apache.hadoop.hbase.ipc.RpcServerInterface; //导入依赖的package包/类
@Override
public RpcServerInterface getRpcServer() {
  return rpcServer;
}
 

import org.apache.hadoop.hbase.ipc.RpcServerInterface; //导入依赖的package包/类
public void setRpcServer(RpcServerInterface rpc) {
  this.rpcServer = rpc;
}
 

import org.apache.hadoop.hbase.ipc.RpcServerInterface; //导入依赖的package包/类
@Override
public RpcServerInterface getRpcServer() {
    return rpcServices.rpcServer;
}