Java 类org.apache.hadoop.hbase.protobuf.ProtobufUtil 实例源码

/**
 * @param pbBytes A pb serialized {@link ValueFilter} instance
 * @return An instance of {@link ValueFilter} made from <code>bytes</code>
 * @throws DeserializationException
 * @see #toByteArray
 */
public static ValueFilter parseFrom(final byte [] pbBytes)
throws DeserializationException {
  FilterProtos.ValueFilter proto;
  try {
    proto = FilterProtos.ValueFilter.parseFrom(pbBytes);
  } catch (InvalidProtocolBufferException e) {
    throw new DeserializationException(e);
  }
  final CompareOp valueCompareOp =
    CompareOp.valueOf(proto.getCompareFilter().getCompareOp().name());
  ByteArrayComparable valueComparator = null;
  try {
    if (proto.getCompareFilter().hasComparator()) {
      valueComparator = ProtobufUtil.toComparator(proto.getCompareFilter().getComparator());
    }
  } catch (IOException ioe) {
    throw new DeserializationException(ioe);
  }
  return new ValueFilter(valueCompareOp,valueComparator);
}

/**
 * @param pbBytes A pb serialized {@link FamilyFilter} instance
 * @return An instance of {@link FamilyFilter} made from <code>bytes</code>
 * @throws DeserializationException
 * @see #toByteArray
 */
public static FamilyFilter parseFrom(final byte [] pbBytes)
throws DeserializationException {
  FilterProtos.FamilyFilter proto;
  try {
    proto = FilterProtos.FamilyFilter.parseFrom(pbBytes);
  } catch (InvalidProtocolBufferException e) {
    throw new DeserializationException(e);
  }
  final CompareOp valueCompareOp =
    CompareOp.valueOf(proto.getCompareFilter().getCompareOp().name());
  ByteArrayComparable valueComparator = null;
  try {
    if (proto.getCompareFilter().hasComparator()) {
      valueComparator = ProtobufUtil.toComparator(proto.getCompareFilter().getComparator());
    }
  } catch (IOException ioe) {
    throw new DeserializationException(ioe);
  }
  return new FamilyFilter(valueCompareOp,valueComparator);
}

/**
 * Get list of table names by namespace
 * @param name namespace name
 * @return The list of table names in the namespace
 * @throws IOException
 */
@Override
public TableName[] listTableNamesByNamespace(final String name) throws IOException {
  return
      executeCallable(new MasterCallable<TableName[]>(getConnection()) {
        @Override
        public TableName[] call(int callTimeout) throws Exception {
          PayloadCarryingRpcController controller = rpcControllerFactory.newController();
          controller.setCallTimeout(callTimeout);
          List<HBaseProtos.TableName> tableNames =
            master.listTableNamesByNamespace(controller, ListTableNamesByNamespaceRequest.
              newBuilder().setNamespaceName(name).build())
              .getTableNameList();
          TableName[] result = new TableName[tableNames.size()];
          for (int i = 0; i < tableNames.size(); i++) {
            result[i] = ProtobufUtil.toTableName(tableNames.get(i));
          }
          return result;
        }
      });
}

@Test
public void testEndpoint() throws Exception {
  final ServerName serverName = TEST_UTIL.getHBaseCluster().getRegionServer(0).getServerName();
  final ServerRpcController controller = new ServerRpcController();
  final BlockingRpcCallback<DummyRegionServerEndpointProtos.DummyResponse> rpcCallback =
      new BlockingRpcCallback<DummyRegionServerEndpointProtos.DummyResponse>();
  DummyRegionServerEndpointProtos.DummyService service =
      ProtobufUtil.newServiceStub(DummyRegionServerEndpointProtos.DummyService.class,
          TEST_UTIL.getHBaseAdmin().coprocessorService(serverName));
  service.dummyCall(controller,
      DummyRegionServerEndpointProtos.DummyRequest.getDefaultInstance(), rpcCallback);
  assertEquals(DUMMY_VALUE, rpcCallback.get().getValue());
  if (controller.failedOnException()) {
    throw controller.getFailedOn();
  }
}

/**
 * Revoke permissions globally from the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void revokeGlobal(final HBaseTestingUtility util, final String user,
    final Permission.Action... actions) throws Exception {
  SecureTestUtil.updateACLs(util, new Callable<Void>() {
    @Override
    public Void call() throws Exception {
      try (Connection connection = ConnectionFactory.createConnection(util.getConfiguration())) {
        try (Table acl = connection.getTable(AccessControlLists.ACL_TABLE_NAME)) {
          BlockingRpcChannel service = acl.coprocessorService(HConstants.EMPTY_START_ROW);
          AccessControlService.BlockingInterface protocol =
              AccessControlService.newBlockingStub(service);
          ProtobufUtil.revoke(null, protocol, user, actions);
        }
      }
      return null;
    }
  });
}

@Test
public void testAttributesSerialization() throws IOException {
  Scan scan = new Scan();
  scan.setAttribute("attribute1", Bytes.toBytes("value1"));
  scan.setAttribute("attribute2", Bytes.toBytes("value2"));
  scan.setAttribute("attribute3", Bytes.toBytes("value3"));

  ClientProtos.Scan scanProto = ProtobufUtil.toScan(scan);

  Scan scan2 = ProtobufUtil.toScan(scanProto);

  Assert.assertNull(scan2.getAttribute("absent"));
  Assert.assertTrue(Arrays.equals(Bytes.toBytes("value1"), scan2.getAttribute("attribute1")));
  Assert.assertTrue(Arrays.equals(Bytes.toBytes("value2"), scan2.getAttribute("attribute2")));
  Assert.assertTrue(Arrays.equals(Bytes.toBytes("value3"), scan2.getAttribute("attribute3")));
  Assert.assertEquals(3, scan2.getAttributesMap().size());
}

private void testTokenAuth(Class<? extends RpcClient> rpcImplClass) throws IOException,
    ServiceException {
  TEST_UTIL.getConfiguration().set(RpcClientFactory.CUSTOM_RPC_CLIENT_IMPL_CONF_KEY,
    rpcImplClass.getName());
  try (Connection conn = ConnectionFactory.createConnection(TEST_UTIL.getConfiguration());
      Table table = conn.getTable(TableName.META_TABLE_NAME)) {
    CoprocessorRpcChannel rpcChannel = table.coprocessorService(HConstants.EMPTY_START_ROW);
    AuthenticationProtos.AuthenticationService.BlockingInterface service =
        AuthenticationProtos.AuthenticationService.newBlockingStub(rpcChannel);
    WhoAmIResponse response = service.whoAmI(null, WhoAmIRequest.getDefaultInstance());
    assertEquals(USERNAME, response.getUsername());
    assertEquals(AuthenticationMethod.TOKEN.name(), response.getAuthMethod());
    try {
      service.getAuthenticationToken(null, GetAuthenticationTokenRequest.getDefaultInstance());
    } catch (ServiceException e) {
      AccessDeniedException exc = (AccessDeniedException) ProtobufUtil.getRemoteException(e);
      assertTrue(exc.getMessage().contains(
        "Token generation only allowed for Kerberos authenticated clients"));
    }
  }
}

/**
 * Create a new namespace
 * @param descriptor descriptor which describes the new namespace
 * @throws IOException
 */
@Override
public void createNamespace(final NamespaceDescriptor descriptor) throws IOException {
  executeCallable(new MasterCallable<Void>(getConnection()) {
    @Override
    public Void call(int callTimeout) throws Exception {
      PayloadCarryingRpcController controller = rpcControllerFactory.newController();
      controller.setCallTimeout(callTimeout);
      // TODO: set priority based on NS?
      master.createNamespace(controller,
        CreateNamespaceRequest.newBuilder()
          .setNamespaceDescriptor(ProtobufUtil
            .toProtoNamespaceDescriptor(descriptor)).build()
      );
      return null;
    }
  });
}

private InterProcessLock createTableLock() {
  String tableLockZNode = ZKUtil.joinZNode(zkWatcher.tableLockZNode,
      tableName.getNameAsString());

  ZooKeeperProtos.TableLock data = ZooKeeperProtos.TableLock.newBuilder()
    .setTableName(ProtobufUtil.toProtoTableName(tableName))
    .setLockOwner(ProtobufUtil.toServerName(serverName))
    .setThreadId(Thread.currentThread().getId())
    .setPurpose(purpose)
    .setIsShared(isShared)
    .setCreateTime(EnvironmentEdgeManager.currentTime()).build();
  byte[] lockMetadata = toBytes(data);

  InterProcessReadWriteLock lock = new ZKInterProcessReadWriteLock(zkWatcher, tableLockZNode,
    METADATA_HANDLER);
  return isShared ? lock.readLock(lockMetadata) : lock.writeLock(lockMetadata);
}

/**
 * {@inheritDoc}
 */
@Override
public CompactionState getCompactionStateForRegion(final byte[] regionName)
throws IOException {
  try {
    Pair<HRegionInfo, ServerName> regionServerPair = getRegion(regionName);
    if (regionServerPair == null) {
      throw new IllegalArgumentException("Invalid region: " + Bytes.toStringBinary(regionName));
    }
    if (regionServerPair.getSecond() == null) {
      throw new NoServerForRegionException(Bytes.toStringBinary(regionName));
    }
    ServerName sn = regionServerPair.getSecond();
    AdminService.BlockingInterface admin = this.connection.getAdmin(sn);
    GetRegionInfoRequest request = RequestConverter.buildGetRegionInfoRequest(
      regionServerPair.getFirst().getRegionName(), true);
    PayloadCarryingRpcController controller = rpcControllerFactory.newController();
    // TODO: this does not do retries, it should. Set priority and timeout in controller
    GetRegionInfoResponse response = admin.getRegionInfo(controller, request);
    return response.getCompactionState();
  } catch (ServiceException se) {
    throw ProtobufUtil.getRemoteException(se);
  }
}

/**
 * @param regionLastFlushedSequenceId the flushed sequence id of a region which is the min of its
 *          store max seq ids
 * @param storeSequenceIds column family to sequence Id map
 * @return Serialized protobuf of <code>RegionSequenceIds</code> with pb magic prefix prepended
 *         suitable for use to filter wal edits in distributedLogReplay mode
 */
public static byte[] regionSequenceIdsToByteArray(final Long regionLastFlushedSequenceId,
    final Map<byte[], Long> storeSequenceIds) {
  ClusterStatusProtos.RegionStoreSequenceIds.Builder regionSequenceIdsBuilder =
      ClusterStatusProtos.RegionStoreSequenceIds.newBuilder();
  ClusterStatusProtos.StoreSequenceId.Builder storeSequenceIdBuilder =
      ClusterStatusProtos.StoreSequenceId.newBuilder();
  if (storeSequenceIds != null) {
    for (Map.Entry<byte[], Long> e : storeSequenceIds.entrySet()){
      byte[] columnFamilyName = e.getKey();
      Long curSeqId = e.getValue();
      storeSequenceIdBuilder.setFamilyName(ByteStringer.wrap(columnFamilyName));
      storeSequenceIdBuilder.setSequenceId(curSeqId);
      regionSequenceIdsBuilder.addStoreSequenceId(storeSequenceIdBuilder.build());
      storeSequenceIdBuilder.clear();
    }
  }
  regionSequenceIdsBuilder.setLastFlushedSequenceId(regionLastFlushedSequenceId);
  byte[] result = regionSequenceIdsBuilder.build().toByteArray();
  return ProtobufUtil.prependPBMagic(result);
}

/**
 * Sends an MERGE REGIONS RPC to the specified server to merge the specified
 * regions.
 * <p>
 * A region server could reject the close request because it either does not
 * have the specified region.
 * @param server server to merge regions
 * @param region_a region to merge
 * @param region_b region to merge
 * @param forcible true if do a compulsory merge, otherwise we will only merge
 *          two adjacent regions
 * @throws IOException
 */
public void sendRegionsMerge(ServerName server, HRegionInfo region_a,
    HRegionInfo region_b, boolean forcible) throws IOException {
  if (server == null)
    throw new NullPointerException("Passed server is null");
  if (region_a == null || region_b == null)
    throw new NullPointerException("Passed region is null");
  AdminService.BlockingInterface admin = getRsAdmin(server);
  if (admin == null) {
    throw new IOException("Attempting to send MERGE REGIONS RPC to server "
        + server.toString() + " for region "
        + region_a.getRegionNameAsString() + ","
        + region_b.getRegionNameAsString()
        + " failed because no RPC connection found to this server");
  }
  PayloadCarryingRpcController controller = newRpcController();
  ProtobufUtil.mergeRegions(controller, admin, region_a, region_b, forcible);
}

@Override
public void serializeStateData(final OutputStream stream) throws IOException {
  super.serializeStateData(stream);

  MasterProcedureProtos.TruncateTableStateData.Builder state =
    MasterProcedureProtos.TruncateTableStateData.newBuilder()
      .setUserInfo(MasterProcedureUtil.toProtoUserInfo(this.user))
      .setPreserveSplits(preserveSplits);
  if (hTableDescriptor != null) {
    state.setTableSchema(hTableDescriptor.convert());
  } else {
    state.setTableName(ProtobufUtil.toProtoTableName(tableName));
  }
  if (regions != null) {
    for (HRegionInfo hri: regions) {
      state.addRegionInfo(HRegionInfo.convert(hri));
    }
  }
  state.build().writeDelimitedTo(stream);
}

@Override
public void deserializeStateData(final InputStream stream) throws IOException {
  super.deserializeStateData(stream);

  MasterProcedureProtos.TruncateTableStateData state =
    MasterProcedureProtos.TruncateTableStateData.parseDelimitedFrom(stream);
  user = MasterProcedureUtil.toUserInfo(state.getUserInfo());
  if (state.hasTableSchema()) {
    hTableDescriptor = HTableDescriptor.convert(state.getTableSchema());
    tableName = hTableDescriptor.getTableName();
  } else {
    tableName = ProtobufUtil.toTableName(state.getTableName());
  }
  preserveSplits = state.getPreserveSplits();
  if (state.getRegionInfoCount() == 0) {
    regions = null;
  } else {
    regions = new ArrayList<HRegionInfo>(state.getRegionInfoCount());
    for (HBaseProtos.RegionInfo hri: state.getRegionInfoList()) {
      regions.add(HRegionInfo.convert(hri));
    }
  }
}

/**
 * Revoke permissions on a namespace from the given user. Will wait until all active
 * AccessController instances have updated their permissions caches or will
 * throw an exception upon timeout (10 seconds).
 */
public static void revokeFromNamespace(final HBaseTestingUtility util, final String user,
    final String namespace, final Permission.Action... actions) throws Exception {
  SecureTestUtil.updateACLs(util, new Callable<Void>() {
    @Override
    public Void call() throws Exception {
      try (Connection connection = ConnectionFactory.createConnection(util.getConfiguration())) {
        try (Table acl = connection.getTable(AccessControlLists.ACL_TABLE_NAME)) {
          BlockingRpcChannel service = acl.coprocessorService(HConstants.EMPTY_START_ROW);
          AccessControlService.BlockingInterface protocol =
              AccessControlService.newBlockingStub(service);
          ProtobufUtil.revoke(null, protocol, user, namespace, actions);
        }
      }
      return null;
    }
  });
}

@Test (timeout=180000)
public void testGlobalPermissionList() throws Exception {
  List<UserPermission> perms;
  Table acl = systemUserConnection.getTable(AccessControlLists.ACL_TABLE_NAME);
  try {
    BlockingRpcChannel service = acl.coprocessorService(HConstants.EMPTY_START_ROW);
    AccessControlService.BlockingInterface protocol =
      AccessControlService.newBlockingStub(service);
    perms = ProtobufUtil.getUserPermissions(null, protocol);
  } finally {
    acl.close();
  }
  UserPermission adminPerm = new UserPermission(Bytes.toBytes(USER_ADMIN.getShortName()),
    AccessControlLists.ACL_TABLE_NAME, null, null, Bytes.toBytes("ACRW"));
  assertTrue("Only global users and user admin has permission on table _acl_ per setup",
    perms.size() == 5 && hasFoundUserPermission(adminPerm, perms));
}

public synchronized NavigableSet<NamespaceDescriptor> list() throws IOException {
  NavigableSet<NamespaceDescriptor> ret =
      Sets.newTreeSet(NamespaceDescriptor.NAMESPACE_DESCRIPTOR_COMPARATOR);
  ResultScanner scanner = getNamespaceTable().getScanner(HTableDescriptor.NAMESPACE_FAMILY_INFO_BYTES);
  try {
    for(Result r : scanner) {
      byte[] val = CellUtil.cloneValue(r.getColumnLatestCell(
        HTableDescriptor.NAMESPACE_FAMILY_INFO_BYTES,
        HTableDescriptor.NAMESPACE_COL_DESC_BYTES));
      ret.add(ProtobufUtil.toNamespaceDescriptor(
          HBaseProtos.NamespaceDescriptor.parseFrom(val)));
    }
  } finally {
    scanner.close();
  }
  return ret;
}

@Override
public RegionServerReportResponse regionServerReport(
    RpcController controller, RegionServerReportRequest request) throws ServiceException {
  try {
    master.checkServiceStarted();
    ClusterStatusProtos.ServerLoad sl = request.getLoad();
    ServerName serverName = ProtobufUtil.toServerName(request.getServer());
    ServerLoad oldLoad = master.serverManager.getLoad(serverName);
    master.serverManager.regionServerReport(serverName, new ServerLoad(sl));
    if (sl != null && master.metricsMaster != null) {
      // Up our metrics.
      master.metricsMaster.incrementRequests(sl.getTotalNumberOfRequests()
          - (oldLoad != null ? oldLoad.getTotalNumberOfRequests() : 0));
    }
  } catch (IOException ioe) {
    throw new ServiceException(ioe);
  }
  return RegionServerReportResponse.newBuilder().build();
}

@Override
public Result call(int callTimeout) throws Exception {
  if (controller.isCanceled()) return null;

  if (Thread.interrupted()) {
    throw new InterruptedIOException();
  }

  byte[] reg = location.getRegionInfo().getRegionName();

  ClientProtos.GetRequest request =
      RequestConverter.buildGetRequest(reg, get);
  controller.setCallTimeout(callTimeout);

  try {
    ClientProtos.GetResponse response = getStub().get(controller, request);
    if (response == null) {
      return null;
    }
    return ProtobufUtil.toResult(response.getResult());
  } catch (ServiceException se) {
    throw ProtobufUtil.getRemoteException(se);
  }
}

/**
 * Get the number of regions of the table that have been updated by the alter.
 *
 * @return Pair indicating the number of regions updated Pair.getFirst is the
 *         regions that are yet to be updated Pair.getSecond is the total number
 *         of regions of the table
 * @throws ServiceException
 */
@Override
public GetSchemaAlterStatusResponse getSchemaAlterStatus(
    RpcController controller, GetSchemaAlterStatusRequest req) throws ServiceException {
  // TODO: currently, we query using the table name on the client side. this
  // may overlap with other table operations or the table operation may
  // have completed before querying this API. We need to refactor to a
  // transaction system in the future to avoid these ambiguities.
  TableName tableName = ProtobufUtil.toTableName(req.getTableName());

  try {
    master.checkInitialized();
    Pair<Integer,Integer> pair = master.assignmentManager.getReopenStatus(tableName);
    GetSchemaAlterStatusResponse.Builder ret = GetSchemaAlterStatusResponse.newBuilder();
    ret.setYetToUpdateRegions(pair.getFirst());
    ret.setTotalRegions(pair.getSecond());
    return ret.build();
  } catch (IOException ioe) {
    throw new ServiceException(ioe);
  }
}

/**
 * Get list of userspace table names
 * @param controller Unused (set to null).
 * @param req GetTableNamesRequest
 * @return GetTableNamesResponse
 * @throws ServiceException
 */
@Override
public GetTableNamesResponse getTableNames(RpcController controller,
    GetTableNamesRequest req) throws ServiceException {
  try {
    master.checkInitialized();

    final String regex = req.hasRegex() ? req.getRegex() : null;
    final String namespace = req.hasNamespace() ? req.getNamespace() : null;
    List<TableName> tableNames = master.listTableNames(namespace, regex,
        req.getIncludeSysTables());

    GetTableNamesResponse.Builder builder = GetTableNamesResponse.newBuilder();
    if (tableNames != null && tableNames.size() > 0) {
      // Add the table names to the response
      for (TableName table: tableNames) {
        builder.addTableNames(ProtobufUtil.toProtoTableName(table));
      }
    }
    return builder.build();
  } catch (IOException e) {
    throw new ServiceException(e);
  }
}

/**
 * This acts as the ack for a completed procedure
 */
@Override
public void sendMemberCompleted(Subprocedure sub, byte[] data) throws IOException {
  String procName = sub.getName();
  LOG.debug("Marking procedure  '" + procName + "' completed for member '" + memberName
      + "' in zk");
  String joinPath = ZKUtil.joinZNode(zkController.getReachedBarrierNode(procName), memberName);
  // ProtobufUtil.prependPBMagic does not take care of null
  if (data == null) {
    data = new byte[0];
  }
  try {
    ZKUtil.createAndFailSilent(zkController.getWatcher(), joinPath,
      ProtobufUtil.prependPBMagic(data));
  } catch (KeeperException e) {
    member.controllerConnectionFailure("Failed to post zk node:" + joinPath
        + " to join procedure barrier.", e, procName);
  }
}

/**
 * Sets the location of <code>hbase:meta</code> in ZooKeeper to the
 * specified server address.
 * @param zookeeper
 * @param serverName
 * @param replicaId
 * @param state
 * @throws KeeperException
 */
public static void setMetaLocation(ZooKeeperWatcher zookeeper,
    ServerName serverName, int replicaId, RegionState.State state) throws KeeperException {
  LOG.info("Setting hbase:meta region location in ZooKeeper as " + serverName);
  // Make the MetaRegionServer pb and then get its bytes and save this as
  // the znode content.
  MetaRegionServer pbrsr = MetaRegionServer.newBuilder()
    .setServer(ProtobufUtil.toServerName(serverName))
    .setRpcVersion(HConstants.RPC_CURRENT_VERSION)
    .setState(state.convert()).build();
  byte[] data = ProtobufUtil.prependPBMagic(pbrsr.toByteArray());
  try {
    ZKUtil.setData(zookeeper, zookeeper.getZNodeForReplica(replicaId), data);
  } catch(KeeperException.NoNodeException nne) {
    if (replicaId == HRegionInfo.DEFAULT_REPLICA_ID) {
      LOG.debug("META region location doesn't exist, create it");
    } else {
      LOG.debug("META region location doesn't exist for replicaId " + replicaId +
          ", create it");
    }
    ZKUtil.createAndWatch(zookeeper, zookeeper.getZNodeForReplica(replicaId), data);
  }
}

/**
 * Verifies current version of file system
 *
 * @param fs filesystem object
 * @param rootdir root hbase directory
 * @return null if no version file exists, version string otherwise.
 * @throws IOException e
 * @throws org.apache.hadoop.hbase.exceptions.DeserializationException
 */
public static String getVersion(FileSystem fs, Path rootdir)
throws IOException, DeserializationException {
  Path versionFile = new Path(rootdir, HConstants.VERSION_FILE_NAME);
  FileStatus[] status = null;
  try {
    // hadoop 2.0 throws FNFE if directory does not exist.
    // hadoop 1.0 returns null if directory does not exist.
    status = fs.listStatus(versionFile);
  } catch (FileNotFoundException fnfe) {
    return null;
  }
  if (status == null || status.length == 0) return null;
  String version = null;
  byte [] content = new byte [(int)status[0].getLen()];
  FSDataInputStream s = fs.open(versionFile);
  try {
    IOUtils.readFully(s, content, 0, content.length);
    if (ProtobufUtil.isPBMagicPrefix(content)) {
      version = parseVersionFrom(content);
    } else {
      // Presume it pre-pb format.
      InputStream is = new ByteArrayInputStream(content);
      DataInputStream dis = new DataInputStream(is);
      try {
        version = dis.readUTF();
      } finally {
        dis.close();
      }
    }
  } catch (EOFException eof) {
    LOG.warn("Version file was empty, odd, will try to set it.");
  } finally {
    s.close();
  }
  return version;
}

public static Filter deserializeFilter(byte[] filterBytes) {
  if (filterBytes == null) {
    return null;
  }
  try {
    FilterProtos.Filter pbFilter = FilterProtos.Filter.parseFrom(filterBytes);
    return ProtobufUtil.toFilter(pbFilter);
  } catch (Exception e) {
    throw new DrillRuntimeException("Error deserializing filter: " + filterBytes, e);
  }
}

@Test(timeout = 300000)
public void testClusterRequests() throws Exception {

  // sending fake request to master to see how metric value has changed
  RegionServerStatusProtos.RegionServerReportRequest.Builder request =
      RegionServerStatusProtos.RegionServerReportRequest.newBuilder();
  ServerName serverName = cluster.getMaster(0).getServerName();
  request.setServer(ProtobufUtil.toServerName(serverName));

  MetricsMasterSource masterSource = master.getMasterMetrics().getMetricsSource();
  ClusterStatusProtos.ServerLoad sl = ClusterStatusProtos.ServerLoad.newBuilder()
                                         .setTotalNumberOfRequests(10000)
                                         .build();
  masterSource.init();
  request.setLoad(sl);
  master.getMasterRpcServices().regionServerReport(null, request.build());

  metricsHelper.assertCounter("cluster_requests", 10000, masterSource);

  sl = ClusterStatusProtos.ServerLoad.newBuilder()
      .setTotalNumberOfRequests(15000)
      .build();
  request.setLoad(sl);
  master.getMasterRpcServices().regionServerReport(null, request.build());

  metricsHelper.assertCounter("cluster_requests", 15000, masterSource);

  master.getMasterRpcServices().regionServerReport(null, request.build());

  metricsHelper.assertCounter("cluster_requests", 15000, masterSource);
  master.stopMaster();
}

public static Filter deserializeFilter(byte[] filterBytes) {
  if (filterBytes == null) {
    return null;
  }
  try {
    FilterProtos.Filter pbFilter = FilterProtos.Filter.parseFrom(filterBytes);
    return ProtobufUtil.toFilter(pbFilter);
  } catch (Exception e) {
    throw new RuntimeException("Error deserializing filter: " + filterBytes, e);
  }
}

@Test
public void testFamilyFilter() throws Exception {
  FamilyFilter familyFilter = new FamilyFilter(CompareFilter.CompareOp.EQUAL,
    new BinaryPrefixComparator(Bytes.toBytes("testValueOne")));
  assertTrue(familyFilter.areSerializedFieldsEqual(
    ProtobufUtil.toFilter(ProtobufUtil.toFilter(familyFilter))));
}

/**
 * Reads back from the zookeeper. The data read here is of the form written by
 * writeToZooKeeper(Map<byte[], Integer> entries).
 * 
 * @param data
 * @return Labels and their ordinal details
 * @throws DeserializationException
 */
public static List<VisibilityLabel> readLabelsFromZKData(byte[] data)
    throws DeserializationException {
  if (ProtobufUtil.isPBMagicPrefix(data)) {
    int pblen = ProtobufUtil.lengthOfPBMagic();
    try {
      VisibilityLabelsRequest.Builder builder = VisibilityLabelsRequest.newBuilder();
      ProtobufUtil.mergeFrom(builder, data, pblen, data.length - pblen);
      return builder.getVisLabelList();
    } catch (IOException e) {
      throw new DeserializationException(e);
    }
  }
  return null;
}

@Test
public void testKeyOnlyFilter() throws Exception {
  // KeyOnlyFilter with lenAsVal
  KeyOnlyFilter keyOnlyFilter = new KeyOnlyFilter(true);
  assertTrue(keyOnlyFilter.areSerializedFieldsEqual(
    ProtobufUtil.toFilter(ProtobufUtil.toFilter(keyOnlyFilter))));

  // KeyOnlyFilter without lenAsVal
  keyOnlyFilter = new KeyOnlyFilter();
  assertTrue(keyOnlyFilter.areSerializedFieldsEqual(
    ProtobufUtil.toFilter(ProtobufUtil.toFilter(keyOnlyFilter))));
}

private void writeNamespace(NamespaceDescriptor ns) throws IOException {
  String zNode = ZKUtil.joinZNode(nsZNode, ns.getName());
  try {
    ZKUtil.createWithParents(watcher, zNode);
    ZKUtil.updateExistingNodeData(watcher, zNode,
        ProtobufUtil.toProtoNamespaceDescriptor(ns).toByteArray(), -1);
  } catch (KeeperException e) {
    LOG.error("Failed updating permissions for namespace "+ns.getName(), e);
    throw new IOException("Failed updating permissions for namespace "+ns.getName(), e);
  }
}

private static QuotaSettings throttle(final String userName, final TableName tableName,
    final String namespace, final ThrottleType type, final long limit, final TimeUnit timeUnit) {
  QuotaProtos.ThrottleRequest.Builder builder = QuotaProtos.ThrottleRequest.newBuilder();
  if (type != null) {
    builder.setType(ProtobufUtil.toProtoThrottleType(type));
  }
  if (timeUnit != null) {
    builder.setTimedQuota(ProtobufUtil.toTimedQuota(limit, timeUnit, QuotaScope.MACHINE));
  }
  return new ThrottleSettings(userName, tableName, namespace, builder.build());
}

/**
 * List all the userPermissions matching the given pattern.
 * @param connection The Connection instance to use
 * @param tableRegex The regular expression string to match against
 * @return - returns an array of UserPermissions
 * @throws Throwable
 */
public static List<UserPermission> getUserPermissions(Connection connection, String tableRegex)
    throws Throwable {
  PayloadCarryingRpcController controller
    = ((ClusterConnection) connection).getRpcControllerFactory().newController();
  List<UserPermission> permList = new ArrayList<UserPermission>();
  try (Table table = connection.getTable(ACL_TABLE_NAME)) {
    try (Admin admin = connection.getAdmin()) {
      CoprocessorRpcChannel service = table.coprocessorService(HConstants.EMPTY_START_ROW);
      BlockingInterface protocol =
          AccessControlProtos.AccessControlService.newBlockingStub(service);
      HTableDescriptor[] htds = null;
      if (tableRegex == null || tableRegex.isEmpty()) {
        permList = ProtobufUtil.getUserPermissions(controller, protocol);
      } else if (tableRegex.charAt(0) == '@') {
        String namespace = tableRegex.substring(1);
        permList = ProtobufUtil.getUserPermissions(controller, protocol,
          Bytes.toBytes(namespace));
      } else {
        htds = admin.listTables(Pattern.compile(tableRegex), true);
        for (HTableDescriptor hd : htds) {
          permList.addAll(ProtobufUtil.getUserPermissions(controller, protocol,
            hd.getTableName()));
        }
      }
    }
  }
  return permList;
}

@Override
public ProtobufMessageHandler getObjectFromMessage(byte[] message)
    throws IOException {
  Cell.Builder builder = Cell.newBuilder();
  ProtobufUtil.mergeFrom(builder, message);
  setColumn(builder.getColumn().toByteArray());
  setValue(builder.getData().toByteArray());
  if (builder.hasTimestamp()) {
    setTimestamp(builder.getTimestamp());
  }
  return this;
}

private RegionNormalizerProtos.RegionNormalizerState parseFrom(byte [] pbBytes)
  throws DeserializationException {
  ProtobufUtil.expectPBMagicPrefix(pbBytes);
  RegionNormalizerProtos.RegionNormalizerState.Builder builder =
    RegionNormalizerProtos.RegionNormalizerState.newBuilder();
  try {
    int magicLen = ProtobufUtil.lengthOfPBMagic();
    ProtobufUtil.mergeFrom(builder, pbBytes, magicLen, pbBytes.length - magicLen);
  } catch (IOException e) {
    throw new DeserializationException(e);
  }
  return builder.build();
}

/**
 * @param data zookeeper data. may be null
 * @return pb object of master, null if no active master
 * @throws DeserializationException
 */
public static ZooKeeperProtos.Master parse(byte[] data) throws DeserializationException {
  if (data == null) {
    return null;
  }
  int prefixLen = ProtobufUtil.lengthOfPBMagic();
  try {
    return ZooKeeperProtos.Master.PARSER.parseFrom(data, prefixLen, data.length - prefixLen);
  } catch (InvalidProtocolBufferException e) {
    throw new DeserializationException(e);
  }
}

/**
 * @param data Serialized date to parse.
 * @return An SplitLogTaskState instance made of the passed <code>data</code>
 * @throws DeserializationException 
 * @see #toByteArray()
 */
public static SplitLogTask parseFrom(final byte [] data) throws DeserializationException {
  ProtobufUtil.expectPBMagicPrefix(data);
  try {
    int prefixLen = ProtobufUtil.lengthOfPBMagic();
    ZooKeeperProtos.SplitLogTask.Builder builder = ZooKeeperProtos.SplitLogTask.newBuilder();
    ProtobufUtil.mergeFrom(builder, data, prefixLen, data.length - prefixLen);
    return new SplitLogTask(builder.build());
  } catch (IOException e) {
    throw new DeserializationException(Bytes.toStringBinary(data, 0, 64), e);
  }
}

private void writeRegionOpenMarker(WAL wal, long openSeqId) throws IOException {
  Map<byte[], List<Path>> storeFiles = getStoreFiles();
  RegionEventDescriptor regionOpenDesc = ProtobufUtil
      .toRegionEventDescriptor(RegionEventDescriptor.EventType.REGION_OPEN, getRegionInfo(),
          openSeqId, getRegionServerServices().getServerName(), storeFiles);
  WALUtil.writeRegionEventMarker(wal, getTableDesc(), getRegionInfo(), regionOpenDesc, mvcc);
}

/**
 * Get tableDescriptors
 * @param tableNames List of table names
 * @return HTD[] the tableDescriptor
 * @throws IOException if a remote or network exception occurs
 */
@Override
public HTableDescriptor[] getTableDescriptorsByTableName(final List<TableName> tableNames)
throws IOException {
  return executeCallable(new MasterCallable<HTableDescriptor[]>(getConnection()) {
    @Override
    public HTableDescriptor[] call(int callTimeout) throws Exception {
      PayloadCarryingRpcController controller = rpcControllerFactory.newController();
      controller.setCallTimeout(callTimeout);
      GetTableDescriptorsRequest req =
          RequestConverter.buildGetTableDescriptorsRequest(tableNames);
        return ProtobufUtil.getHTableDescriptorArray(master.getTableDescriptors(controller, req));
    }
  });
}

@Override
public GetResponse get(RpcController controller, GetRequest request)
throws ServiceException {
  byte[] regionName = request.getRegion().getValue().toByteArray();
  Map<byte [], Result> m = this.gets.get(regionName);
  GetResponse.Builder builder = GetResponse.newBuilder();
  if (m != null) {
    byte[] row = request.getGet().getRow().toByteArray();
    builder.setResult(ProtobufUtil.toResult(m.get(row)));
  }
  return builder.build();
}