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

@Override
public RpcScheduler create(Configuration conf, PriorityFunction priority, Abortable server) {
  int handlerCount = conf.getInt(HConstants.REGION_SERVER_HANDLER_COUNT,
HConstants.DEFAULT_REGION_SERVER_HANDLER_COUNT);

  return new SimpleRpcScheduler(
    conf,
    handlerCount,
    conf.getInt(HConstants.REGION_SERVER_HIGH_PRIORITY_HANDLER_COUNT,
      HConstants.DEFAULT_REGION_SERVER_HIGH_PRIORITY_HANDLER_COUNT),
    conf.getInt(HConstants.REGION_SERVER_REPLICATION_HANDLER_COUNT,
        HConstants.DEFAULT_REGION_SERVER_REPLICATION_HANDLER_COUNT),
    priority,
    server,
    HConstants.QOS_THRESHOLD);
}

@Override
public RpcScheduler create(Configuration conf, PriorityFunction priority, Abortable server) {
  int handlerCount = conf.getInt(HConstants.REGION_SERVER_HANDLER_COUNT,
HConstants.DEFAULT_REGION_SERVER_HANDLER_COUNT);

  return new SimpleRpcScheduler(
    conf,
    handlerCount,
    conf.getInt(HConstants.REGION_SERVER_HIGH_PRIORITY_HANDLER_COUNT,
      HConstants.DEFAULT_REGION_SERVER_HIGH_PRIORITY_HANDLER_COUNT),
    conf.getInt(HConstants.REGION_SERVER_REPLICATION_HANDLER_COUNT,
        HConstants.DEFAULT_REGION_SERVER_REPLICATION_HANDLER_COUNT),
    priority,
    server,
    HConstants.QOS_THRESHOLD);
}

@Override
public RpcScheduler create(Configuration conf, RegionServerServices server) {
  int handlerCount = conf.getInt(HConstants.REGION_SERVER_HANDLER_COUNT,
      HConstants.DEFAULT_REGION_SERVER_HANDLER_COUNT);
  return new SimpleRpcScheduler(
      conf,
      handlerCount,
      conf.getInt(HConstants.REGION_SERVER_META_HANDLER_COUNT,
          HConstants.DEFAULT_REGION_SERVER_META_HANDLER_COUNT),
      conf.getInt(HConstants.REGION_SERVER_REPLICATION_HANDLER_COUNT,
          HConstants.DEFAULT_REGION_SERVER_REPLICATION_HANDLER_COUNT),
      server,
      HConstants.QOS_THRESHOLD);
}

@Override
public RpcScheduler create(Configuration conf, PriorityFunction priority, Abortable server) {
  int handlerCount = conf.getInt(HConstants.REGION_SERVER_HANDLER_COUNT,
      HConstants.DEFAULT_REGION_SERVER_HANDLER_COUNT);
  return new SimpleRpcScheduler(
    conf,
    handlerCount,
    conf.getInt(HConstants.REGION_SERVER_HIGH_PRIORITY_HANDLER_COUNT,
      HConstants.DEFAULT_REGION_SERVER_HIGH_PRIORITY_HANDLER_COUNT),
    conf.getInt(HConstants.REGION_SERVER_REPLICATION_HANDLER_COUNT,
        HConstants.DEFAULT_REGION_SERVER_REPLICATION_HANDLER_COUNT),
    priority,
    server,
    HConstants.QOS_THRESHOLD);
}

@Test
public void testRWQ() {
  // Set some configs just to see how it changes the scheduler. Can't assert the settings had
  // an effect. Just eyeball the log.
  this.conf.setDouble(RWQueueRpcExecutor.CALL_QUEUE_READ_SHARE_CONF_KEY, 0.5);
  this.conf.setDouble(RpcExecutor.CALL_QUEUE_HANDLER_FACTOR_CONF_KEY, 0.5);
  this.conf.setDouble(RWQueueRpcExecutor.CALL_QUEUE_SCAN_SHARE_CONF_KEY, 0.5);
  RpcSchedulerFactory factory = new SimpleRpcSchedulerFactory();
  RpcScheduler rpcScheduler = factory.create(this.conf, null, null);
  assertTrue(rpcScheduler.getClass().equals(SimpleRpcScheduler.class));
}

@Override
public RpcScheduler create(Configuration conf, PriorityFunction priority) {
  int handlerCount = conf.getInt(HConstants.REGION_SERVER_HANDLER_COUNT,
      HConstants.DEFAULT_REGION_SERVER_HANDLER_COUNT);
  return new SimpleRpcScheduler(
      conf,
      handlerCount,
      conf.getInt(HConstants.REGION_SERVER_META_HANDLER_COUNT,
          HConstants.DEFAULT_REGION_SERVER_META_HANDLER_COUNT),
      conf.getInt(HConstants.REGION_SERVER_REPLICATION_HANDLER_COUNT,
          HConstants.DEFAULT_REGION_SERVER_REPLICATION_HANDLER_COUNT),
      priority,
      HConstants.QOS_THRESHOLD);
}

@Test
public void testCallQueueTooBigExceptionDoesntTriggerPffe() throws Exception {
  Admin admin = TEST_UTIL.getAdmin();

  final String tableName = name.getMethodName();
  HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(Bytes
    .toBytes(tableName)));
  desc.addFamily(new HColumnDescriptor(FAMILY));
  admin.createTable(desc, Bytes.toBytes("aaaa"), Bytes.toBytes("zzzz"), 3);

  Configuration conf = TEST_UTIL.getConfiguration();
  conf.setLong(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, 100);
  conf.setInt(HConstants.HBASE_CLIENT_PAUSE, 500);
  conf.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 1);

  conf.setBoolean(HConstants.HBASE_CLIENT_FAST_FAIL_MODE_ENABLED, true);
  conf.setLong(HConstants.HBASE_CLIENT_FAST_FAIL_THREASHOLD_MS, 0);
  conf.setClass(HConstants.HBASE_CLIENT_FAST_FAIL_INTERCEPTOR_IMPL,
    CallQueueTooBigPffeInterceptor.class, PreemptiveFastFailInterceptor.class);

  final Connection connection = ConnectionFactory.createConnection(conf);

  //Set max call queues size to 0
  SimpleRpcScheduler srs = (SimpleRpcScheduler)
    TEST_UTIL.getHBaseCluster().getRegionServer(0).getRpcServer().getScheduler();
  Configuration newConf = HBaseConfiguration.create(TEST_UTIL.getConfiguration());
  newConf.setInt("hbase.ipc.server.max.callqueue.length", 0);
  srs.onConfigurationChange(newConf);

  try (Table table = connection.getTable(TableName.valueOf(tableName))) {
    Get get = new Get(new byte[1]);
    table.get(get);
  } catch (Throwable ex) {
  }

  assertEquals("We should have not entered PFFE mode on CQTBE, but we did;"
    + " number of times this mode should have been entered:", 0,
    CallQueueTooBigPffeInterceptor.numCallQueueTooBig.get());

  newConf = HBaseConfiguration.create(TEST_UTIL.getConfiguration());
  newConf.setInt("hbase.ipc.server.max.callqueue.length", 250);
  srs.onConfigurationChange(newConf);
}