import org.apache.hadoop.hbase.CellScannable; //导入依赖的package包/类
/**
 * Create a protocol buffer MultiRequest for row mutations that does not hold data.  Data/Cells
 * are carried outside of protobuf.  Return references to the Cells in <code>cells</code> param.
  * Does not propagate Action absolute position.  Does not set atomic action on the created
 * RegionAtomic.  Caller should do that if wanted.
 * @param regionName
 * @param rowMutations
 * @param cells Return in here a list of Cells as CellIterable.
 * @return a region mutation minus data
 * @throws IOException
 */
public static RegionAction.Builder buildNoDataRegionAction(final byte[] regionName,
    final RowMutations rowMutations, final List<CellScannable> cells,
    final RegionAction.Builder regionActionBuilder,
    final ClientProtos.Action.Builder actionBuilder,
    final MutationProto.Builder mutationBuilder)
throws IOException {
  for (Mutation mutation: rowMutations.getMutations()) {
    MutationType type = null;
    if (mutation instanceof Put) {
      type = MutationType.PUT;
    } else if (mutation instanceof Delete) {
      type = MutationType.DELETE;
    } else {
      throw new DoNotRetryIOException("RowMutations supports only put and delete, not " +
        mutation.getClass().getName());
    }
    mutationBuilder.clear();
    MutationProto mp = ProtobufUtil.toMutationNoData(type, mutation, mutationBuilder);
    cells.add(mutation);
    actionBuilder.clear();
    regionActionBuilder.addAction(actionBuilder.setMutation(mp).build());
  }
  return regionActionBuilder;
}
 

import org.apache.hadoop.hbase.CellScannable; //导入依赖的package包/类
@Test
public void testListOfCellScannerables() throws IOException {
  List<CellScannable> cells = new ArrayList<CellScannable>();
  final int count = 10;
  for (int i = 0; i < count; i++) {
    cells.add(createCell(i));
  }
  PayloadCarryingRpcController controller = new PayloadCarryingRpcController(cells);
  CellScanner cellScanner = controller.cellScanner();
  int index = 0;
  for (; cellScanner.advance(); index++) {
    Cell cell = cellScanner.current();
    byte [] indexBytes = Bytes.toBytes(index);
    assertTrue("" + index, Bytes.equals(indexBytes, 0, indexBytes.length, cell.getValueArray(),
      cell.getValueOffset(), cell.getValueLength()));
  }
  assertEquals(count, index);
}
 

import org.apache.hadoop.hbase.CellScannable; //导入依赖的package包/类
private ClientProtos.MultiRequest buildReq(Map<byte[], RegionRequest> actionsByRegion,
    List<CellScannable> cells, Map<Integer, Integer> rowMutationsIndexMap) throws IOException {
  ClientProtos.MultiRequest.Builder multiRequestBuilder = ClientProtos.MultiRequest.newBuilder();
  ClientProtos.RegionAction.Builder regionActionBuilder = ClientProtos.RegionAction.newBuilder();
  ClientProtos.Action.Builder actionBuilder = ClientProtos.Action.newBuilder();
  ClientProtos.MutationProto.Builder mutationBuilder = ClientProtos.MutationProto.newBuilder();
  for (Map.Entry<byte[], RegionRequest> entry : actionsByRegion.entrySet()) {
    long nonceGroup = conn.getNonceGenerator().getNonceGroup();
    // multiRequestBuilder will be populated with region actions.
    // rowMutationsIndexMap will be non-empty after the call if there is RowMutations in the
    // action list.
    RequestConverter.buildNoDataRegionActions(entry.getKey(),
      entry.getValue().actions, cells, multiRequestBuilder, regionActionBuilder, actionBuilder,
      mutationBuilder, nonceGroup, rowMutationsIndexMap);
  }
  return multiRequestBuilder.build();
}
 

import org.apache.hadoop.hbase.CellScannable; //导入依赖的package包/类
@Test
public void testListOfCellScannerables() throws IOException {
  final int count = 10;
  List<CellScannable> cells = new ArrayList<>(count);

  for (int i = 0; i < count; i++) {
    cells.add(createCell(i));
  }
  HBaseRpcController controller = new HBaseRpcControllerImpl(cells);
  CellScanner cellScanner = controller.cellScanner();
  int index = 0;
  for (; cellScanner.advance(); index++) {
    Cell cell = cellScanner.current();
    byte[] indexBytes = Bytes.toBytes(index);
    assertTrue("" + index, Bytes.equals(indexBytes, 0, indexBytes.length, cell.getValueArray(),
      cell.getValueOffset(), cell.getValueLength()));
  }
  assertEquals(count, index);
}
 

import org.apache.hadoop.hbase.CellScannable; //导入依赖的package包/类
@Override
public ScanResponse scan(RpcController controller, ScanRequest request)
    throws ServiceException {
  ScanResponse.Builder builder = ScanResponse.newBuilder();
  try {
    if (request.hasScan()) {
      byte[] regionName = request.getRegion().getValue().toByteArray();
      builder.setScannerId(openScanner(regionName, null));
      builder.setMoreResults(true);
    }
    else {
      long scannerId = request.getScannerId();
      Result result = next(scannerId);
      if (result != null) {
        builder.addCellsPerResult(result.size());
        List<CellScannable> results = new ArrayList<CellScannable>(1);
        results.add(result);
        ((PayloadCarryingRpcController) controller).setCellScanner(CellUtil
            .createCellScanner(results));
        builder.setMoreResults(true);
      }
      else {
        builder.setMoreResults(false);
        close(scannerId);
      }
    }
  } catch (IOException ie) {
    throw new ServiceException(ie);
  }
  return builder.build();
}
 

import org.apache.hadoop.hbase.CellScannable; //导入依赖的package包/类
@Override
public ScanResponse scan(RpcController controller, ScanRequest request)
    throws ServiceException {
  ScanResponse.Builder builder = ScanResponse.newBuilder();
  try {
    if (request.hasScan()) {
      byte[] regionName = request.getRegion().getValue().toByteArray();
      builder.setScannerId(openScanner(regionName, null));
      builder.setMoreResults(true);
    }
    else {
      long scannerId = request.getScannerId();
      Result result = next(scannerId);
      if (result != null) {
        builder.addCellsPerResult(result.size());
        List<CellScannable> results = new ArrayList<>(1);
        results.add(result);
        ((HBaseRpcController) controller).setCellScanner(CellUtil
            .createCellScanner(results));
        builder.setMoreResults(true);
      }
      else {
        builder.setMoreResults(false);
        close(scannerId);
      }
    }
  } catch (IOException ie) {
    throw new ServiceException(ie);
  }
  return builder.build();
}
 

import org.apache.hadoop.hbase.CellScannable; //导入依赖的package包/类
@Override
public PayloadCarryingRpcController newController(final List<CellScannable> cellIterables) {
  return new CountingRpcController(super.newController(cellIterables));
}
 

import org.apache.hadoop.hbase.CellScannable; //导入依赖的package包/类
public static void main(String[] args) throws IOException, SecurityException,
    NoSuchMethodException, InterruptedException {
  if (args.length != 2) {
    System.out.println("Usage: TestAsyncIPC <CYCLES> <CELLS_PER_CYCLE>");
    return;
  }
  // ((Log4JLogger)HBaseServer.LOG).getLogger().setLevel(Level.INFO);
  // ((Log4JLogger)HBaseClient.LOG).getLogger().setLevel(Level.INFO);
  int cycles = Integer.parseInt(args[0]);
  int cellcount = Integer.parseInt(args[1]);
  Configuration conf = HBaseConfiguration.create();
  TestRpcServer rpcServer = new TestRpcServer();
  MethodDescriptor md = SERVICE.getDescriptorForType().findMethodByName("echo");
  EchoRequestProto param = EchoRequestProto.newBuilder().setMessage("hello").build();
  AsyncRpcClient client = new AsyncRpcClient(conf);
  KeyValue kv = BIG_CELL;
  Put p = new Put(CellUtil.cloneRow(kv));
  for (int i = 0; i < cellcount; i++) {
    p.add(kv);
  }
  RowMutations rm = new RowMutations(CellUtil.cloneRow(kv));
  rm.add(p);
  try {
    rpcServer.start();
    InetSocketAddress address = rpcServer.getListenerAddress();
    if (address == null) {
      throw new IOException("Listener channel is closed");
    }
    long startTime = System.currentTimeMillis();
    User user = User.getCurrent();
    for (int i = 0; i < cycles; i++) {
      List<CellScannable> cells = new ArrayList<CellScannable>();
      // Message param = RequestConverter.buildMultiRequest(HConstants.EMPTY_BYTE_ARRAY, rm);
      ClientProtos.RegionAction.Builder builder =
          RequestConverter.buildNoDataRegionAction(HConstants.EMPTY_BYTE_ARRAY, rm, cells,
            RegionAction.newBuilder(), ClientProtos.Action.newBuilder(),
            MutationProto.newBuilder());
      builder.setRegion(RegionSpecifier
          .newBuilder()
          .setType(RegionSpecifierType.REGION_NAME)
          .setValue(
            ByteString.copyFrom(HRegionInfo.FIRST_META_REGIONINFO.getEncodedNameAsBytes())));
      if (i % 100000 == 0) {
        LOG.info("" + i);
        // Uncomment this for a thread dump every so often.
        // ReflectionUtils.printThreadInfo(new PrintWriter(System.out),
        // "Thread dump " + Thread.currentThread().getName());
      }
      PayloadCarryingRpcController pcrc =
          new PayloadCarryingRpcController(CellUtil.createCellScanner(cells));
      // Pair<Message, CellScanner> response =
      client.call(pcrc, md, builder.build(), param, user, address,
          new MetricsConnection.CallStats());
      /*
       * int count = 0; while (p.getSecond().advance()) { count++; } assertEquals(cells.size(),
       * count);
       */
    }
    LOG.info("Cycled " + cycles + " time(s) with " + cellcount + " cell(s) in "
        + (System.currentTimeMillis() - startTime) + "ms");
  } finally {
    client.close();
    rpcServer.stop();
  }
}
 

import org.apache.hadoop.hbase.CellScannable; //导入依赖的package包/类
public static void main(String[] args) throws IOException, SecurityException,
    NoSuchMethodException, InterruptedException {
  if (args.length != 2) {
    System.out.println("Usage: TestIPC <CYCLES> <CELLS_PER_CYCLE>");
    return;
  }
  // ((Log4JLogger)HBaseServer.LOG).getLogger().setLevel(Level.INFO);
  // ((Log4JLogger)HBaseClient.LOG).getLogger().setLevel(Level.INFO);
  int cycles = Integer.parseInt(args[0]);
  int cellcount = Integer.parseInt(args[1]);
  Configuration conf = HBaseConfiguration.create();
  TestRpcServer rpcServer = new TestRpcServer();
  MethodDescriptor md = SERVICE.getDescriptorForType().findMethodByName("echo");
  EchoRequestProto param = EchoRequestProto.newBuilder().setMessage("hello").build();
  RpcClientImpl client = new RpcClientImpl(conf, HConstants.CLUSTER_ID_DEFAULT);
  KeyValue kv = BIG_CELL;
  Put p = new Put(CellUtil.cloneRow(kv));
  for (int i = 0; i < cellcount; i++) {
    p.add(kv);
  }
  RowMutations rm = new RowMutations(CellUtil.cloneRow(kv));
  rm.add(p);
  try {
    rpcServer.start();
    long startTime = System.currentTimeMillis();
    User user = User.getCurrent();
    InetSocketAddress address = rpcServer.getListenerAddress();
    if (address == null) {
      throw new IOException("Listener channel is closed");
    }
    for (int i = 0; i < cycles; i++) {
      List<CellScannable> cells = new ArrayList<CellScannable>();
      // Message param = RequestConverter.buildMultiRequest(HConstants.EMPTY_BYTE_ARRAY, rm);
      ClientProtos.RegionAction.Builder builder =
          RequestConverter.buildNoDataRegionAction(HConstants.EMPTY_BYTE_ARRAY, rm, cells,
            RegionAction.newBuilder(), ClientProtos.Action.newBuilder(),
            MutationProto.newBuilder());
      builder.setRegion(RegionSpecifier
          .newBuilder()
          .setType(RegionSpecifierType.REGION_NAME)
          .setValue(
            ByteString.copyFrom(HRegionInfo.FIRST_META_REGIONINFO.getEncodedNameAsBytes())));
      if (i % 100000 == 0) {
        LOG.info("" + i);
        // Uncomment this for a thread dump every so often.
        // ReflectionUtils.printThreadInfo(new PrintWriter(System.out),
        // "Thread dump " + Thread.currentThread().getName());
      }
      PayloadCarryingRpcController pcrc =
          new PayloadCarryingRpcController(CellUtil.createCellScanner(cells));
      // Pair<Message, CellScanner> response =
      client.call(pcrc, md, builder.build(), param, user, address,
          new MetricsConnection.CallStats());
      /*
       * int count = 0; while (p.getSecond().advance()) { count++; } assertEquals(cells.size(),
       * count);
       */
    }
    LOG.info("Cycled " + cycles + " time(s) with " + cellcount + " cell(s) in "
        + (System.currentTimeMillis() - startTime) + "ms");
  } finally {
    client.close();
    rpcServer.stop();
  }
}
 

import org.apache.hadoop.hbase.CellScannable; //导入依赖的package包/类
public PayloadCarryingRpcController(final List<CellScannable> cellIterables) {
  this.cellScanner = cellIterables == null? null: CellUtil.createCellScanner(cellIterables);
}
 

import org.apache.hadoop.hbase.CellScannable; //导入依赖的package包/类
public PayloadCarryingRpcController newController(final List<CellScannable> cellIterables) {
  return new PayloadCarryingRpcController(cellIterables);
}
 

import org.apache.hadoop.hbase.CellScannable; //导入依赖的package包/类
public PayloadCarryingRpcController newController(final List<CellScannable> cellIterables) {
  return new CountingRpcController(super.newController(cellIterables));
}
 

import org.apache.hadoop.hbase.CellScannable; //导入依赖的package包/类
private void processServerShutdownHandler(AssignmentManager am, boolean splitRegion)
    throws IOException, ServiceException {
  // Make sure our new AM gets callbacks; once registered, can't unregister.
  // Thats ok because we make a new zk watcher for each test.
  this.watcher.registerListenerFirst(am);

  // Need to set up a fake scan of meta for the servershutdown handler
  // Make an RS Interface implementation.  Make it so a scanner can go against it.
  ClientProtos.ClientService.BlockingInterface implementation =
    Mockito.mock(ClientProtos.ClientService.BlockingInterface.class);
  // Get a meta row result that has region up on SERVERNAME_A

  Result r;
  if (splitRegion) {
    r = MetaMockingUtil.getMetaTableRowResultAsSplitRegion(REGIONINFO, SERVERNAME_A);
  } else {
    r = MetaMockingUtil.getMetaTableRowResult(REGIONINFO, SERVERNAME_A);
  }

  final ScanResponse.Builder builder = ScanResponse.newBuilder();
  builder.setMoreResults(true);
  builder.addCellsPerResult(r.size());
  final List<CellScannable> cellScannables = new ArrayList<CellScannable>(1);
  cellScannables.add(r);
  Mockito.when(implementation.scan(
    (RpcController)Mockito.any(), (ScanRequest)Mockito.any())).
    thenAnswer(new Answer<ScanResponse>() {
        @Override
        public ScanResponse answer(InvocationOnMock invocation) throws Throwable {
          PayloadCarryingRpcController controller = (PayloadCarryingRpcController) invocation
              .getArguments()[0];
          if (controller != null) {
            controller.setCellScanner(CellUtil.createCellScanner(cellScannables));
          }
          return builder.build();
        }
    });

  // Get a connection w/ mocked up common methods.
  ClusterConnection connection =
    HConnectionTestingUtility.getMockedConnectionAndDecorate(HTU.getConfiguration(),
      null, implementation, SERVERNAME_B, REGIONINFO);
  // These mocks were done up when all connections were managed.  World is different now we
  // moved to unmanaged connections.  It messes up the intercepts done in these tests.
  // Just mark connections as marked and then down in MetaTableAccessor, it will go the path
  // that picks up the above mocked up 'implementation' so 'scans' of meta return the expected
  // result.  Redo in new realm of unmanaged connections.
  Mockito.when(connection.isManaged()).thenReturn(true);
  try {
    // Make it so we can get a catalogtracker from servermanager.. .needed
    // down in guts of server shutdown handler.
    Mockito.when(this.server.getConnection()).thenReturn(connection);

    // Now make a server shutdown handler instance and invoke process.
    // Have it that SERVERNAME_A died.
    DeadServer deadServers = new DeadServer();
    deadServers.add(SERVERNAME_A);
    // I need a services instance that will return the AM
    MasterFileSystem fs = Mockito.mock(MasterFileSystem.class);
    Mockito.doNothing().when(fs).setLogRecoveryMode();
    Mockito.when(fs.getLogRecoveryMode()).thenReturn(RecoveryMode.LOG_REPLAY);
    MasterServices services = Mockito.mock(MasterServices.class);
    Mockito.when(services.getAssignmentManager()).thenReturn(am);
    Mockito.when(services.getServerManager()).thenReturn(this.serverManager);
    Mockito.when(services.getZooKeeper()).thenReturn(this.watcher);
    Mockito.when(services.getMasterFileSystem()).thenReturn(fs);
    Mockito.when(services.getConnection()).thenReturn(connection);
    ServerShutdownHandler handler = new ServerShutdownHandler(this.server,
        services, deadServers, SERVERNAME_A, false);
    am.failoverCleanupDone.set(true);
    handler.process();
    // The region in r will have been assigned.  It'll be up in zk as unassigned.
  } finally {
    if (connection != null) connection.close();
  }
}
 

import org.apache.hadoop.hbase.CellScannable; //导入依赖的package包/类
/**
 * Create an {@link AssignmentManagerWithExtrasForTesting} that has mocked
 * {@link CatalogTracker} etc.
 * @param server
 * @param manager
 * @return An AssignmentManagerWithExtras with mock connections, etc.
 * @throws IOException
 * @throws KeeperException
 */
private AssignmentManagerWithExtrasForTesting setUpMockedAssignmentManager(final Server server,
    final ServerManager manager) throws IOException, KeeperException,
      ServiceException, CoordinatedStateException {
  // Make an RS Interface implementation. Make it so a scanner can go against
  // it and a get to return the single region, REGIONINFO, this test is
  // messing with. Needed when "new master" joins cluster. AM will try and
  // rebuild its list of user regions and it will also get the HRI that goes
  // with an encoded name by doing a Get on hbase:meta
  ClientProtos.ClientService.BlockingInterface ri =
    Mockito.mock(ClientProtos.ClientService.BlockingInterface.class);
  // Get a meta row result that has region up on SERVERNAME_A for REGIONINFO
  Result r = MetaMockingUtil.getMetaTableRowResult(REGIONINFO, SERVERNAME_A);
  final ScanResponse.Builder builder = ScanResponse.newBuilder();
  builder.setMoreResults(true);
  builder.addCellsPerResult(r.size());
  final List<CellScannable> rows = new ArrayList<CellScannable>(1);
  rows.add(r);
  Answer<ScanResponse> ans = new Answer<ClientProtos.ScanResponse>() {
    @Override
    public ScanResponse answer(InvocationOnMock invocation) throws Throwable {
      PayloadCarryingRpcController controller = (PayloadCarryingRpcController) invocation
          .getArguments()[0];
      if (controller != null) {
        controller.setCellScanner(CellUtil.createCellScanner(rows));
      }
      return builder.build();
    }
  };
  if (enabling) {
    Mockito.when(ri.scan((RpcController) Mockito.any(), (ScanRequest) Mockito.any()))
        .thenAnswer(ans).thenAnswer(ans).thenAnswer(ans).thenAnswer(ans).thenAnswer(ans)
        .thenReturn(ScanResponse.newBuilder().setMoreResults(false).build());
  } else {
    Mockito.when(ri.scan((RpcController) Mockito.any(), (ScanRequest) Mockito.any())).thenAnswer(
        ans);
  }
  // If a get, return the above result too for REGIONINFO
  GetResponse.Builder getBuilder = GetResponse.newBuilder();
  getBuilder.setResult(ProtobufUtil.toResult(r));
  Mockito.when(ri.get((RpcController)Mockito.any(), (GetRequest) Mockito.any())).
    thenReturn(getBuilder.build());
  // Get a connection w/ mocked up common methods.
  ClusterConnection connection = (ClusterConnection)HConnectionTestingUtility.
    getMockedConnectionAndDecorate(HTU.getConfiguration(), null,
      ri, SERVERNAME_B, REGIONINFO);
  // These mocks were done up when all connections were managed.  World is different now we
  // moved to unmanaged connections.  It messes up the intercepts done in these tests.
  // Just mark connections as marked and then down in MetaTableAccessor, it will go the path
  // that picks up the above mocked up 'implementation' so 'scans' of meta return the expected
  // result.  Redo in new realm of unmanaged connections.
  Mockito.when(connection.isManaged()).thenReturn(true);
  // Make it so we can get the connection from our mocked catalogtracker
  // Create and startup an executor. Used by AM handling zk callbacks.
  ExecutorService executor = startupMasterExecutor("mockedAMExecutor");
  this.balancer = LoadBalancerFactory.getLoadBalancer(server.getConfiguration());
  AssignmentManagerWithExtrasForTesting am = new AssignmentManagerWithExtrasForTesting(
    server, connection, manager, this.balancer, executor, new NullTableLockManager());
  return am;
}
 

import org.apache.hadoop.hbase.CellScannable; //导入依赖的package包/类
public static void main(String[] args)
throws IOException, SecurityException, NoSuchMethodException, InterruptedException {
  if (args.length != 2) {
    System.out.println("Usage: TestIPC <CYCLES> <CELLS_PER_CYCLE>");
    return;
  }
  // ((Log4JLogger)HBaseServer.LOG).getLogger().setLevel(Level.INFO);
  // ((Log4JLogger)HBaseClient.LOG).getLogger().setLevel(Level.INFO);
  int cycles = Integer.parseInt(args[0]);
  int cellcount = Integer.parseInt(args[1]);
  Configuration conf = HBaseConfiguration.create();
  TestRpcServer rpcServer = new TestRpcServer();
  MethodDescriptor md = SERVICE.getDescriptorForType().findMethodByName("echo");
  EchoRequestProto param = EchoRequestProto.newBuilder().setMessage("hello").build();
  RpcClientImpl client = new RpcClientImpl(conf, HConstants.CLUSTER_ID_DEFAULT);
  KeyValue kv = KeyValueUtil.ensureKeyValue(BIG_CELL);
  Put p = new Put(kv.getRow());
  for (int i = 0; i < cellcount; i++) {
    p.add(kv);
  }
  RowMutations rm = new RowMutations(kv.getRow());
  rm.add(p);
  try {
    rpcServer.start();
    InetSocketAddress address = rpcServer.getListenerAddress();
    long startTime = System.currentTimeMillis();
    User user = User.getCurrent();
    for (int i = 0; i < cycles; i++) {
      List<CellScannable> cells = new ArrayList<CellScannable>();
      // Message param = RequestConverter.buildMultiRequest(HConstants.EMPTY_BYTE_ARRAY, rm);
      ClientProtos.RegionAction.Builder builder = RequestConverter.buildNoDataRegionAction(
        HConstants.EMPTY_BYTE_ARRAY, rm, cells,
        RegionAction.newBuilder(),
        ClientProtos.Action.newBuilder(),
        MutationProto.newBuilder());
      builder.setRegion(RegionSpecifier.newBuilder().setType(RegionSpecifierType.REGION_NAME).
        setValue(ByteString.copyFrom(HRegionInfo.FIRST_META_REGIONINFO.getEncodedNameAsBytes())));
      if (i % 100000 == 0) {
        LOG.info("" + i);
        // Uncomment this for a thread dump every so often.
        // ReflectionUtils.printThreadInfo(new PrintWriter(System.out),
        //  "Thread dump " + Thread.currentThread().getName());
      }
      CellScanner cellScanner = CellUtil.createCellScanner(cells);
      Pair<Message, CellScanner> response =
          client.call(null, md, builder.build(), cellScanner, param, user, address, 0);
      /*
      int count = 0;
      while (p.getSecond().advance()) {
        count++;
      }
      assertEquals(cells.size(), count);*/
    }
    LOG.info("Cycled " + cycles + " time(s) with " + cellcount + " cell(s) in " +
       (System.currentTimeMillis() - startTime) + "ms");
  } finally {
    client.close();
    rpcServer.stop();
  }
}
 

import org.apache.hadoop.hbase.CellScannable; //导入依赖的package包/类
private void processServerShutdownHandler(CatalogTracker ct, AssignmentManager am, boolean splitRegion)
    throws IOException, ServiceException {
  // Make sure our new AM gets callbacks; once registered, can't unregister.
  // Thats ok because we make a new zk watcher for each test.
  this.watcher.registerListenerFirst(am);

  // Need to set up a fake scan of meta for the servershutdown handler
  // Make an RS Interface implementation.  Make it so a scanner can go against it.
  ClientProtos.ClientService.BlockingInterface implementation =
    Mockito.mock(ClientProtos.ClientService.BlockingInterface.class);
  // Get a meta row result that has region up on SERVERNAME_A

  Result r;
  if (splitRegion) {
    r = MetaMockingUtil.getMetaTableRowResultAsSplitRegion(REGIONINFO, SERVERNAME_A);
  } else {
    r = MetaMockingUtil.getMetaTableRowResult(REGIONINFO, SERVERNAME_A);
  }

  final ScanResponse.Builder builder = ScanResponse.newBuilder();
  builder.setMoreResults(true);
  builder.addCellsPerResult(r.size());
  final List<CellScannable> cellScannables = new ArrayList<CellScannable>(1);
  cellScannables.add(r);
  Mockito.when(implementation.scan(
    (RpcController)Mockito.any(), (ScanRequest)Mockito.any())).
    thenAnswer(new Answer<ScanResponse>() {
        @Override
        public ScanResponse answer(InvocationOnMock invocation) throws Throwable {
          PayloadCarryingRpcController controller = (PayloadCarryingRpcController) invocation
              .getArguments()[0];
          if (controller != null) {
            controller.setCellScanner(CellUtil.createCellScanner(cellScannables));
          }
          return builder.build();
        }
    });

  // Get a connection w/ mocked up common methods.
  HConnection connection =
    HConnectionTestingUtility.getMockedConnectionAndDecorate(HTU.getConfiguration(),
      null, implementation, SERVERNAME_B, REGIONINFO);

  // Make it so we can get a catalogtracker from servermanager.. .needed
  // down in guts of server shutdown handler.
  Mockito.when(ct.getConnection()).thenReturn(connection);
  Mockito.when(this.server.getCatalogTracker()).thenReturn(ct);

  // Now make a server shutdown handler instance and invoke process.
  // Have it that SERVERNAME_A died.
  DeadServer deadServers = new DeadServer();
  deadServers.add(SERVERNAME_A);
  // I need a services instance that will return the AM
  MasterServices services = Mockito.mock(MasterServices.class);
  Mockito.when(services.getAssignmentManager()).thenReturn(am);
  Mockito.when(services.getServerManager()).thenReturn(this.serverManager);
  Mockito.when(services.getZooKeeper()).thenReturn(this.watcher);
  ServerShutdownHandler handler = new ServerShutdownHandler(this.server,
    services, deadServers, SERVERNAME_A, false);
  am.failoverCleanupDone.set(true);
  handler.process();
  // The region in r will have been assigned.  It'll be up in zk as unassigned.
}
 

import org.apache.hadoop.hbase.CellScannable; //导入依赖的package包/类
/**
 * Create an {@link AssignmentManagerWithExtrasForTesting} that has mocked
 * {@link CatalogTracker} etc.
 * @param server
 * @param manager
 * @return An AssignmentManagerWithExtras with mock connections, etc.
 * @throws IOException
 * @throws KeeperException
 */
private AssignmentManagerWithExtrasForTesting setUpMockedAssignmentManager(final Server server,
    final ServerManager manager) throws IOException, KeeperException, ServiceException {
  // We need a mocked catalog tracker. Its used by our AM instance.
  CatalogTracker ct = Mockito.mock(CatalogTracker.class);
  // Make an RS Interface implementation. Make it so a scanner can go against
  // it and a get to return the single region, REGIONINFO, this test is
  // messing with. Needed when "new master" joins cluster. AM will try and
  // rebuild its list of user regions and it will also get the HRI that goes
  // with an encoded name by doing a Get on hbase:meta
  ClientProtos.ClientService.BlockingInterface ri =
    Mockito.mock(ClientProtos.ClientService.BlockingInterface.class);
  // Get a meta row result that has region up on SERVERNAME_A for REGIONINFO
  Result r = MetaMockingUtil.getMetaTableRowResult(REGIONINFO, SERVERNAME_A);
  final ScanResponse.Builder builder = ScanResponse.newBuilder();
  builder.setMoreResults(true);
  builder.addCellsPerResult(r.size());
  final List<CellScannable> rows = new ArrayList<CellScannable>(1);
  rows.add(r);
  Answer<ScanResponse> ans = new Answer<ClientProtos.ScanResponse>() {
    @Override
    public ScanResponse answer(InvocationOnMock invocation) throws Throwable {
      PayloadCarryingRpcController controller = (PayloadCarryingRpcController) invocation
          .getArguments()[0];
      if (controller != null) {
        controller.setCellScanner(CellUtil.createCellScanner(rows));
      }
      return builder.build();
    }
  };
  if (enabling) {
    Mockito.when(ri.scan((RpcController) Mockito.any(), (ScanRequest) Mockito.any()))
        .thenAnswer(ans).thenAnswer(ans).thenAnswer(ans).thenAnswer(ans).thenAnswer(ans)
        .thenReturn(ScanResponse.newBuilder().setMoreResults(false).build());
  } else {
    Mockito.when(ri.scan((RpcController) Mockito.any(), (ScanRequest) Mockito.any())).thenAnswer(
        ans);
  }
  // If a get, return the above result too for REGIONINFO
  GetResponse.Builder getBuilder = GetResponse.newBuilder();
  getBuilder.setResult(ProtobufUtil.toResult(r));
  Mockito.when(ri.get((RpcController)Mockito.any(), (GetRequest) Mockito.any())).
    thenReturn(getBuilder.build());
  // Get a connection w/ mocked up common methods.
  HConnection connection = HConnectionTestingUtility.
    getMockedConnectionAndDecorate(HTU.getConfiguration(), null,
      ri, SERVERNAME_B, REGIONINFO);
  // Make it so we can get the connection from our mocked catalogtracker
  Mockito.when(ct.getConnection()).thenReturn(connection);
  // Create and startup an executor. Used by AM handling zk callbacks.
  ExecutorService executor = startupMasterExecutor("mockedAMExecutor");
  this.balancer = LoadBalancerFactory.getLoadBalancer(server.getConfiguration());
  AssignmentManagerWithExtrasForTesting am = new AssignmentManagerWithExtrasForTesting(
    server, manager, ct, this.balancer, executor, new NullTableLockManager());
  return am;
}
 

import org.apache.hadoop.hbase.CellScannable; //导入依赖的package包/类
public static void main(String[] args)
throws IOException, SecurityException, NoSuchMethodException, InterruptedException {
  if (args.length != 2) {
    System.out.println("Usage: TestIPC <CYCLES> <CELLS_PER_CYCLE>");
    return;
  }
  // ((Log4JLogger)HBaseServer.LOG).getLogger().setLevel(Level.INFO);
  // ((Log4JLogger)HBaseClient.LOG).getLogger().setLevel(Level.INFO);
  int cycles = Integer.parseInt(args[0]);
  int cellcount = Integer.parseInt(args[1]);
  Configuration conf = HBaseConfiguration.create();
  TestRpcServer rpcServer = new TestRpcServer();
  MethodDescriptor md = SERVICE.getDescriptorForType().findMethodByName("echo");
  EchoRequestProto param = EchoRequestProto.newBuilder().setMessage("hello").build();
  RpcClient client = new RpcClient(conf, HConstants.CLUSTER_ID_DEFAULT);
  KeyValue kv = KeyValueUtil.ensureKeyValue(BIG_CELL);
  Put p = new Put(kv.getRow());
  for (int i = 0; i < cellcount; i++) {
    p.add(kv);
  }
  RowMutations rm = new RowMutations(kv.getRow());
  rm.add(p);
  try {
    rpcServer.start();
    InetSocketAddress address = rpcServer.getListenerAddress();
    long startTime = System.currentTimeMillis();
    User user = User.getCurrent();
    for (int i = 0; i < cycles; i++) {
      List<CellScannable> cells = new ArrayList<CellScannable>();
      // Message param = RequestConverter.buildMultiRequest(HConstants.EMPTY_BYTE_ARRAY, rm);
      ClientProtos.RegionAction.Builder builder = RequestConverter.buildNoDataRegionAction(
        HConstants.EMPTY_BYTE_ARRAY, rm, cells,
        RegionAction.newBuilder(),
        ClientProtos.Action.newBuilder(),
        MutationProto.newBuilder());
      builder.setRegion(RegionSpecifier.newBuilder().setType(RegionSpecifierType.REGION_NAME).
        setValue(ByteString.copyFrom(HRegionInfo.FIRST_META_REGIONINFO.getEncodedNameAsBytes())));
      if (i % 100000 == 0) {
        LOG.info("" + i);
        // Uncomment this for a thread dump every so often.
        // ReflectionUtils.printThreadInfo(new PrintWriter(System.out),
        //  "Thread dump " + Thread.currentThread().getName());
      }
      CellScanner cellScanner = CellUtil.createCellScanner(cells);
      Pair<Message, CellScanner> response =
        client.call(md, builder.build(), cellScanner, param, user, address, 0);
      /*
      int count = 0;
      while (p.getSecond().advance()) {
        count++;
      }
      assertEquals(cells.size(), count);*/
    }
    LOG.info("Cycled " + cycles + " time(s) with " + cellcount + " cell(s) in " +
       (System.currentTimeMillis() - startTime) + "ms");
  } finally {
    client.stop();
    rpcServer.stop();
  }
}
 

import org.apache.hadoop.hbase.CellScannable; //导入依赖的package包/类
@Override
public HBaseRpcController newController(final List<CellScannable> cellIterables) {
  return new CountingRpcController(super.newController(cellIterables));
}