Java 类org.apache.hadoop.hbase.CellScanner 实例源码

private void replayToServer(HRegionInfo regionInfo, List<Entry> entries)
    throws IOException, ServiceException {
  if (entries.isEmpty()) return;

  Entry[] entriesArray = new Entry[entries.size()];
  entriesArray = entries.toArray(entriesArray);
  AdminService.BlockingInterface remoteSvr = conn.getAdmin(getLocation().getServerName());

  Pair<AdminProtos.ReplicateWALEntryRequest, CellScanner> p =
      ReplicationProtbufUtil.buildReplicateWALEntryRequest(entriesArray);
  PayloadCarryingRpcController controller = rpcControllerFactory.newController(p.getSecond());
  try {
    remoteSvr.replay(controller, p.getFirst());
  } catch (ServiceException se) {
    throw ProtobufUtil.getRemoteException(se);
  }
}

/**
 * Replicate WAL entries on the region server.
 *
 * @param controller the RPC controller
 * @param request the request
 * @throws ServiceException
 */
@Override
@QosPriority(priority=HConstants.REPLICATION_QOS)
public ReplicateWALEntryResponse replicateWALEntry(final RpcController controller,
    final ReplicateWALEntryRequest request) throws ServiceException {
  try {
    checkOpen();
    if (regionServer.replicationSinkHandler != null) {
      requestCount.increment();
      List<WALEntry> entries = request.getEntryList();
      CellScanner cellScanner = ((PayloadCarryingRpcController)controller).cellScanner();
      regionServer.getRegionServerCoprocessorHost().preReplicateLogEntries(entries, cellScanner);
      regionServer.replicationSinkHandler.replicateLogEntries(entries, cellScanner);
      regionServer.getRegionServerCoprocessorHost().postReplicateLogEntries(entries, cellScanner);
      return ReplicateWALEntryResponse.newBuilder().build();
    } else {
      throw new ServiceException("Replication services are not initialized yet");
    }
  } catch (IOException ie) {
    throw new ServiceException(ie);
  }
}

private void skipCellsForMutation(Action action, CellScanner cellScanner) {
  try {
    if (action.hasMutation()) {
      MutationProto m = action.getMutation();
      if (m.hasAssociatedCellCount()) {
        for (int i = 0; i < m.getAssociatedCellCount(); i++) {
          cellScanner.advance();
        }
      }
    }
  } catch (IOException e) {
    // No need to handle these Individual Muatation level issue. Any way this entire RegionAction
    // marked as failed as we could not see the Region here. At client side the top level
    // RegionAction exception will be considered first.
    LOG.error("Error while skipping Cells in CellScanner for invalid Region Mutations", e);
  }
}

@edu.umd.cs.findbugs.annotations.SuppressWarnings(value="NP_NULL_ON_SOME_PATH",
    justification="Can't figure why this complaint is happening... see below")
Call(int id, final BlockingService service, final MethodDescriptor md, RequestHeader header,
     Message param, CellScanner cellScanner, Connection connection, Responder responder,
     long size, TraceInfo tinfo, final InetAddress remoteAddress) {
  this.id = id;
  this.service = service;
  this.md = md;
  this.header = header;
  this.param = param;
  this.cellScanner = cellScanner;
  this.connection = connection;
  this.timestamp = System.currentTimeMillis();
  this.response = null;
  this.responder = responder;
  this.isError = false;
  this.size = size;
  this.tinfo = tinfo;
  this.user = connection == null? null: connection.user; // FindBugs: NP_NULL_ON_SOME_PATH
  this.remoteAddress = remoteAddress;
  this.retryImmediatelySupported =
      connection == null? null: connection.retryImmediatelySupported;
}

@Test
public void testSimpleVisibilityLabels() throws Exception {
  TableName tableName = TableName.valueOf(TEST_NAME.getMethodName());
  try (Table table = createTableAndWriteDataWithLabels(tableName, SECRET + "|" + CONFIDENTIAL,
      PRIVATE + "|" + CONFIDENTIAL)) {
    Scan s = new Scan();
    s.setAuthorizations(new Authorizations(SECRET, CONFIDENTIAL, PRIVATE));
    ResultScanner scanner = table.getScanner(s);
    Result[] next = scanner.next(3);

    assertTrue(next.length == 2);
    CellScanner cellScanner = next[0].cellScanner();
    cellScanner.advance();
    Cell current = cellScanner.current();
    assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(),
        current.getRowLength(), row1, 0, row1.length));
    cellScanner = next[1].cellScanner();
    cellScanner.advance();
    current = cellScanner.current();
    assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(),
        current.getRowLength(), row2, 0, row2.length));
  }
}

private static void verifyRow(Result result) throws IOException {
  byte[] row = result.getRow();
  CellScanner scanner = result.cellScanner();
  while (scanner.advance()) {
    Cell cell = scanner.current();

    //assert that all Cells in the Result have the same key
   Assert.assertEquals(0, Bytes.compareTo(row, 0, row.length,
       cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()));
  }

  for (int j = 0; j < FAMILIES.length; j++) {
    byte[] actual = result.getValue(FAMILIES[j], null);
    Assert.assertArrayEquals("Row in snapshot does not match, expected:" + Bytes.toString(row)
        + " ,actual:" + Bytes.toString(actual), row, actual);
  }
}

protected static void verifyRowFromMap(ImmutableBytesWritable key, Result result)
  throws IOException {
  byte[] row = key.get();
  CellScanner scanner = result.cellScanner();
  while (scanner.advance()) {
    Cell cell = scanner.current();

    //assert that all Cells in the Result have the same key
    Assert.assertEquals(0, Bytes.compareTo(row, 0, row.length,
      cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()));
  }

  for (int j = 0; j < FAMILIES.length; j++) {
    byte[] actual = result.getValue(FAMILIES[j], null);
    Assert.assertArrayEquals("Row in snapshot does not match, expected:" + Bytes.toString(row)
      + " ,actual:" + Bytes.toString(actual), row, actual);
  }
}

@Override
public boolean postScannerNext(ObserverContext<RegionCoprocessorEnvironment> e,
    InternalScanner s, List<Result> results, int limit, boolean hasMore) throws IOException {
  if (checkTagPresence) {
    if (results.size() > 0) {
      // Check tag presence in the 1st cell in 1st Result
      Result result = results.get(0);
      CellScanner cellScanner = result.cellScanner();
      if (cellScanner.advance()) {
        Cell cell = cellScanner.current();
        tags = Tag.asList(cell.getTagsArray(), cell.getTagsOffset(),
            cell.getTagsLength());
      }
    }
  }
  return hasMore;
}

/**
 * Ensure we do not HAVE TO HAVE a codec.
 * @throws InterruptedException
 * @throws IOException
 */
@Test
public void testNoCodec() throws InterruptedException, IOException {
  Configuration conf = HBaseConfiguration.create();
  AbstractRpcClient client = createRpcClientNoCodec(conf);
  TestRpcServer rpcServer = new TestRpcServer();
  try {
    rpcServer.start();
    MethodDescriptor md = SERVICE.getDescriptorForType().findMethodByName("echo");
    final String message = "hello";
    EchoRequestProto param = EchoRequestProto.newBuilder().setMessage(message).build();
    InetSocketAddress address = rpcServer.getListenerAddress();
    if (address == null) {
      throw new IOException("Listener channel is closed");
    }
    Pair<Message, CellScanner> r =
        client.call(null, md, param, md.getOutputType().toProto(), User.getCurrent(), address,
            new MetricsConnection.CallStats());
    assertTrue(r.getSecond() == null);
    // Silly assertion that the message is in the returned pb.
    assertTrue(r.getFirst().toString().contains(message));
  } finally {
    client.close();
    rpcServer.stop();
  }
}

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

static CellScanner getSizedCellScanner(final Cell [] cells) {
  int size = -1;
  for (Cell cell: cells) {
    size += CellUtil.estimatedSerializedSizeOf(cell);
  }
  final int totalSize = ClassSize.align(size);
  final CellScanner cellScanner = CellUtil.createCellScanner(cells);
  return new SizedCellScanner() {
    @Override
    public long heapSize() {
      return totalSize;
    }

    @Override
    public Cell current() {
      return cellScanner.current();
    }

    @Override
    public boolean advance() throws IOException {
      return cellScanner.advance();
    }
  };
}

private void scanRow(final Result result, final RowKeyBuilder simpleRowKeyBuilder, final RowKey rowKey,
                     final StatisticType statsType, EventStoreTimeIntervalEnum interval) throws IOException {
    final CellScanner cellScanner = result.cellScanner();
    while (cellScanner.advance()) {
        final Cell cell = cellScanner.current();

        // get the column qualifier
        final byte[] bTimeQualifier = new byte[cell.getQualifierLength()];
        System.arraycopy(cell.getQualifierArray(), cell.getQualifierOffset(), bTimeQualifier, 0,
                cell.getQualifierLength());

        // convert this into a true time, albeit rounded to the column
        // interval granularity
        final long columnIntervalNo = Bytes.toInt(bTimeQualifier);
        final long columnIntervalSize = interval.columnInterval();
        final long columnTimeComponentMillis = columnIntervalNo * columnIntervalSize;
        final long rowKeyPartialTimeMillis = simpleRowKeyBuilder.getPartialTimestamp(rowKey);
        final long fullTimestamp = rowKeyPartialTimeMillis + columnTimeComponentMillis;

        LOGGER.debug("Col: [" + ByteArrayUtils.byteArrayToHex(bTimeQualifier) + "] - ["
                + Bytes.toInt(bTimeQualifier) + "] - [" + fullTimestamp + "] - ["
                + DateUtil.createNormalDateTimeString(fullTimestamp) + "]");

        final byte[] bValue = new byte[cell.getValueLength()];
        System.arraycopy(cell.getValueArray(), cell.getValueOffset(), bValue, 0, cell.getValueLength());

        switch (statsType) {
            case VALUE:
                final ValueCellValue cellValue = new ValueCellValue(bValue);

                LOGGER.debug("Val: " + cellValue);
                break;
            case COUNT:
                LOGGER.debug("Val: " + Bytes.toLong(bValue));
                break;
        }

    }
}

@Override
public Result preAppend(ObserverContext<RegionCoprocessorEnvironment> e, Append append)
    throws IOException {
  // If authorization is not enabled, we don't care about reserved tags
  if (!authorizationEnabled) {
    return null;
  }
  for (CellScanner cellScanner = append.cellScanner(); cellScanner.advance();) {
    if (!checkForReservedVisibilityTagPresence(cellScanner.current())) {
      throw new FailedSanityCheckException("Append contains cell with reserved type tag");
    }
  }
  return null;
}

@Override
public Result preIncrement(ObserverContext<RegionCoprocessorEnvironment> e, Increment increment)
    throws IOException {
  // If authorization is not enabled, we don't care about reserved tags
  if (!authorizationEnabled) {
    return null;
  }
  for (CellScanner cellScanner = increment.cellScanner(); cellScanner.advance();) {
    if (!checkForReservedVisibilityTagPresence(cellScanner.current())) {
      throw new FailedSanityCheckException("Increment contains cell with reserved type tag");
    }
  }
  return null;
}

private void checkForReservedTagPresence(User user, Mutation m) throws IOException {
  // No need to check if we're not going to throw
  if (!authorizationEnabled) {
    m.setAttribute(TAG_CHECK_PASSED, TRUE);
    return;
  }
  // Superusers are allowed to store cells unconditionally.
  if (Superusers.isSuperUser(user)) {
    m.setAttribute(TAG_CHECK_PASSED, TRUE);
    return;
  }
  // We already checked (prePut vs preBatchMutation)
  if (m.getAttribute(TAG_CHECK_PASSED) != null) {
    return;
  }
  for (CellScanner cellScanner = m.cellScanner(); cellScanner.advance();) {
    Cell cell = cellScanner.current();
    if (cell.getTagsLength() > 0) {
      Iterator<Tag> tagsItr = CellUtil.tagsIterator(cell.getTagsArray(), cell.getTagsOffset(),
        cell.getTagsLength());
      while (tagsItr.hasNext()) {
        if (tagsItr.next().getType() == AccessControlLists.ACL_TAG_TYPE) {
          throw new AccessDeniedException("Mutation contains cell with reserved type tag");
        }
      }
    }
  }
  m.setAttribute(TAG_CHECK_PASSED, TRUE);
}

/**
 * A helper to replicate a list of WAL entries using admin protocol.
 *
 * @param admin
 * @param entries
 * @throws java.io.IOException
 */
public static void replicateWALEntry(final AdminService.BlockingInterface admin,
    final Entry[] entries) throws IOException {
  Pair<AdminProtos.ReplicateWALEntryRequest, CellScanner> p =
    buildReplicateWALEntryRequest(entries, null);
  PayloadCarryingRpcController controller = new PayloadCarryingRpcController(p.getSecond());
  try {
    admin.replicateWALEntry(controller, p.getFirst());
  } catch (ServiceException se) {
    throw ProtobufUtil.getRemoteException(se);
  }
}

/**
 * @param cells
 * @return <code>cells</code> packaged as a CellScanner
 */
static CellScanner getCellScanner(final List<List<? extends Cell>> cells, final int size) {
  return new SizedCellScanner() {
    private final Iterator<List<? extends Cell>> entries = cells.iterator();
    private Iterator<? extends Cell> currentIterator = null;
    private Cell currentCell;

    @Override
    public Cell current() {
      return this.currentCell;
    }

    @Override
    public boolean advance() {
      if (this.currentIterator == null) {
        if (!this.entries.hasNext()) return false;
        this.currentIterator = this.entries.next().iterator();
      }
      if (this.currentIterator.hasNext()) {
        this.currentCell = this.currentIterator.next();
        return true;
      }
      this.currentCell = null;
      this.currentIterator = null;
      return advance();
    }

    @Override
    public long heapSize() {
      return size;
    }
  };
}

public void preReplicateLogEntries(final List<WALEntry> entries, final CellScanner cells)
    throws IOException {
  execOperation(coprocessors.isEmpty() ? null : new CoprocessorOperation() {
    @Override
    public void call(RegionServerObserver oserver,
        ObserverContext<RegionServerCoprocessorEnvironment> ctx) throws IOException {
      oserver.preReplicateLogEntries(ctx, entries, cells);
    }
  });
}

public void postReplicateLogEntries(final List<WALEntry> entries, final CellScanner cells)
    throws IOException {
  execOperation(coprocessors.isEmpty() ? null : new CoprocessorOperation() {
    @Override
    public void call(RegionServerObserver oserver,
        ObserverContext<RegionServerCoprocessorEnvironment> ctx) throws IOException {
      oserver.postReplicateLogEntries(ctx, entries, cells);
    }
  });
}

@Test
public void testVisibilityLabelsWithDeleteColumns() throws Throwable {
  setAuths();
  final TableName tableName = TableName.valueOf(TEST_NAME.getMethodName());

  try (Table table = createTableAndWriteDataWithLabels(tableName,
      SECRET + "&" + TOPSECRET, SECRET)) {
    PrivilegedExceptionAction<Void> actiona = new PrivilegedExceptionAction<Void>() {
      @Override
      public Void run() throws Exception {
        try (Connection connection = ConnectionFactory.createConnection(conf);
             Table table = connection.getTable(tableName)) {
          Delete d = new Delete(row1);
          d.setCellVisibility(new CellVisibility(TOPSECRET + "&" + SECRET));
          d.addColumns(fam, qual);
          table.delete(d);
        } catch (Throwable t) {
          throw new IOException(t);
        }
        return null;
      }
    };
    SUPERUSER.runAs(actiona);

    TEST_UTIL.getHBaseAdmin().flush(tableName);
    Scan s = new Scan();
    s.setAuthorizations(new Authorizations(SECRET, PRIVATE, CONFIDENTIAL));
    ResultScanner scanner = table.getScanner(s);
    Result[] next = scanner.next(3);
    assertTrue(next.length == 1);
    CellScanner cellScanner = next[0].cellScanner();
    cellScanner.advance();
    Cell current = cellScanner.current();
    assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(),
        current.getRowLength(), row2, 0, row2.length));

  }
}

@Test
public void testVisibilityLabelsWithDeleteFamily() throws Exception {
  setAuths();
  final TableName tableName = TableName.valueOf(TEST_NAME.getMethodName());
  try (Table table = createTableAndWriteDataWithLabels(tableName, SECRET,
      CONFIDENTIAL + "|" + TOPSECRET);) {
    PrivilegedExceptionAction<Void> actiona = new PrivilegedExceptionAction<Void>() {
      @Override
      public Void run() throws Exception {
        try (Connection connection = ConnectionFactory.createConnection(conf);
             Table table = connection.getTable(tableName)) {
          Delete d = new Delete(row2);
          d.setCellVisibility(new CellVisibility(TOPSECRET + "|" + CONFIDENTIAL));
          d.addFamily(fam);
          table.delete(d);
        } catch (Throwable t) {
          throw new IOException(t);
        }
        return null;
      }
    };
    SUPERUSER.runAs(actiona);

    TEST_UTIL.getHBaseAdmin().flush(tableName);
    Scan s = new Scan();
    s.setAuthorizations(new Authorizations(SECRET, PRIVATE, CONFIDENTIAL));
    ResultScanner scanner = table.getScanner(s);
    Result[] next = scanner.next(3);
    assertTrue(next.length == 1);
    CellScanner cellScanner = next[0].cellScanner();
    cellScanner.advance();
    Cell current = cellScanner.current();
    assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(),
        current.getRowLength(), row1, 0, row1.length));
  }
}

@Test
public void testVisibilityLabelsWithDeleteFamilyVersion() throws Exception {
  setAuths();
  final TableName tableName = TableName.valueOf(TEST_NAME.getMethodName());
  long[] ts = new long[] { 123l, 125l };
  try (Table table = createTableAndWriteDataWithLabels(tableName, ts,
      CONFIDENTIAL + "|" + TOPSECRET, SECRET)) {
    PrivilegedExceptionAction<Void> actiona = new PrivilegedExceptionAction<Void>() {
      @Override
      public Void run() throws Exception {
        try (Connection connection = ConnectionFactory.createConnection(conf);
             Table table = connection.getTable(tableName)) {
          Delete d = new Delete(row1);
          d.setCellVisibility(new CellVisibility(TOPSECRET + "|" + CONFIDENTIAL));
          d.deleteFamilyVersion(fam, 123l);
          table.delete(d);
        } catch (Throwable t) {
          throw new IOException(t);
        }
        return null;
      }
    };
    SUPERUSER.runAs(actiona);

    TEST_UTIL.getHBaseAdmin().flush(tableName);
    Scan s = new Scan();
    s.setAuthorizations(new Authorizations(SECRET, PRIVATE, CONFIDENTIAL));
    ResultScanner scanner = table.getScanner(s);
    Result[] next = scanner.next(3);
    assertTrue(next.length == 1);
    CellScanner cellScanner = next[0].cellScanner();
    cellScanner.advance();
    Cell current = cellScanner.current();
    assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(),
        current.getRowLength(), row2, 0, row2.length));
  }
}

@Test
public void testVisibilityLabelsWithDeleteColumnExactVersion() throws Exception {
  setAuths();
  final TableName tableName = TableName.valueOf(TEST_NAME.getMethodName());
  long[] ts = new long[] { 123l, 125l };
  try (Table table = createTableAndWriteDataWithLabels(tableName, ts,
      CONFIDENTIAL + "|" + TOPSECRET, SECRET);) {
    PrivilegedExceptionAction<Void> actiona = new PrivilegedExceptionAction<Void>() {
      @Override
      public Void run() throws Exception {
        try (Connection connection = ConnectionFactory.createConnection(conf);
             Table table = connection.getTable(tableName)) {
          Delete d = new Delete(row1);
          d.setCellVisibility(new CellVisibility(TOPSECRET + "|" + CONFIDENTIAL));
          d.addColumn(fam, qual, 123l);
          table.delete(d);
        } catch (Throwable t) {
          throw new IOException(t);
        }
        return null;
      }
    };
    SUPERUSER.runAs(actiona);

    TEST_UTIL.getHBaseAdmin().flush(tableName);
    Scan s = new Scan();
    s.setAuthorizations(new Authorizations(SECRET, PRIVATE, CONFIDENTIAL));
    ResultScanner scanner = table.getScanner(s);
    Result[] next = scanner.next(3);
    assertTrue(next.length == 1);
    CellScanner cellScanner = next[0].cellScanner();
    cellScanner.advance();
    Cell current = cellScanner.current();
    assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(),
        current.getRowLength(), row2, 0, row2.length));
  }
}

@Test
public void testDeleteFamilyWithoutCellVisibilityWithMulipleVersions() throws Exception {
  setAuths();
  final TableName tableName = TableName.valueOf(TEST_NAME.getMethodName());
  try (Table table = doPutsWithoutVisibility(tableName)) {
    TEST_UTIL.getHBaseAdmin().flush(tableName);
    PrivilegedExceptionAction<Void> actiona = new PrivilegedExceptionAction<Void>() {
      @Override
      public Void run() throws Exception {
        try (Connection connection = ConnectionFactory.createConnection(conf);
             Table table = connection.getTable(tableName)) {
          Delete d = new Delete(row1);
          d.addFamily(fam);
          table.delete(d);
        } catch (Throwable t) {
          throw new IOException(t);
        }
        return null;
      }
    };
    SUPERUSER.runAs(actiona);

    TEST_UTIL.getHBaseAdmin().flush(tableName);
    Scan s = new Scan();
    s.setMaxVersions(5);
    s.setAuthorizations(new Authorizations(SECRET, PRIVATE, CONFIDENTIAL, TOPSECRET));
    ResultScanner scanner = table.getScanner(s);
    Result[] next = scanner.next(3);
    assertTrue(next.length == 1);
    // All cells wrt row1 should be deleted as we are not passing the Cell Visibility
    CellScanner cellScanner = next[0].cellScanner();
    cellScanner.advance();
    Cell current = cellScanner.current();
    assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(),
        current.getRowLength(), row2, 0, row2.length));
  }
}

private void scanAll(Result[] next) throws IOException {
  CellScanner cellScanner = next[0].cellScanner();
  cellScanner.advance();
  Cell current = cellScanner.current();
  assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(), current.getRowLength(),
      row1, 0, row1.length));
  assertEquals(current.getTimestamp(), 127l);
  cellScanner.advance();
  current = cellScanner.current();
  assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(), current.getRowLength(),
      row1, 0, row1.length));
  assertEquals(current.getTimestamp(), 126l);
  cellScanner.advance();
  current = cellScanner.current();
  assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(), current.getRowLength(),
      row1, 0, row1.length));
  assertEquals(current.getTimestamp(), 125l);
  cellScanner.advance();
  current = cellScanner.current();
  assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(), current.getRowLength(),
      row1, 0, row1.length));
  assertEquals(current.getTimestamp(), 124l);
  cellScanner.advance();
  current = cellScanner.current();
  assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(), current.getRowLength(),
      row1, 0, row1.length));
  assertEquals(current.getTimestamp(), 123l);
  cellScanner = next[1].cellScanner();
  cellScanner.advance();
  current = cellScanner.current();
  assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(), current.getRowLength(),
      row2, 0, row2.length));
}

@Test
public void testSimpleVisibilityLabelsWithUniCodeCharacters() throws Exception {
  TableName tableName = TableName.valueOf(TEST_NAME.getMethodName());
  try (Table table = createTableAndWriteDataWithLabels(tableName,
      SECRET + "|" + CellVisibility.quote(COPYRIGHT), "(" + CellVisibility.quote(COPYRIGHT)
          + "&"  + CellVisibility.quote(ACCENT) + ")|" + CONFIDENTIAL,
      CellVisibility.quote(UNICODE_VIS_TAG) + "&" + SECRET)) {
    Scan s = new Scan();
    s.setAuthorizations(new Authorizations(SECRET, CONFIDENTIAL, PRIVATE, COPYRIGHT, ACCENT,
        UNICODE_VIS_TAG));
    ResultScanner scanner = table.getScanner(s);
    Result[] next = scanner.next(3);
    assertTrue(next.length == 3);
    CellScanner cellScanner = next[0].cellScanner();
    cellScanner.advance();
    Cell current = cellScanner.current();
    assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(),
        current.getRowLength(), row1, 0, row1.length));
    cellScanner = next[1].cellScanner();
    cellScanner.advance();
    current = cellScanner.current();
    assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(),
        current.getRowLength(), row2, 0, row2.length));
    cellScanner = next[2].cellScanner();
    cellScanner.advance();
    current = cellScanner.current();
    assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(),
        current.getRowLength(), row3, 0, row3.length));
  }
}

@Test
public void testAuthorizationsWithSpecialUnicodeCharacters() throws Exception {
  TableName tableName = TableName.valueOf(TEST_NAME.getMethodName());
  try (Table table = createTableAndWriteDataWithLabels(tableName,
      CellVisibility.quote(UC1) + "|" + CellVisibility.quote(UC2), CellVisibility.quote(UC1),
      CellVisibility.quote(UNICODE_VIS_TAG))) {
    Scan s = new Scan();
    s.setAuthorizations(new Authorizations(UC1, UC2, ACCENT,
        UNICODE_VIS_TAG));
    ResultScanner scanner = table.getScanner(s);
    Result[] next = scanner.next(3);
    assertTrue(next.length == 3);
    CellScanner cellScanner = next[0].cellScanner();
    cellScanner.advance();
    Cell current = cellScanner.current();
    assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(),
        current.getRowLength(), row1, 0, row1.length));
    cellScanner = next[1].cellScanner();
    cellScanner.advance();
    current = cellScanner.current();
    assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(),
        current.getRowLength(), row2, 0, row2.length));
    cellScanner = next[2].cellScanner();
    cellScanner.advance();
    current = cellScanner.current();
    assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(),
        current.getRowLength(), row3, 0, row3.length));
  }
}

@Test
public void testVisibilityLabelsWithComplexLabels() throws Exception {
  TableName tableName = TableName.valueOf(TEST_NAME.getMethodName());
  try (Table table = createTableAndWriteDataWithLabels(tableName, "(" + SECRET + "|"
      + CONFIDENTIAL + ")" + "&" + "!" + TOPSECRET, "(" + PRIVATE + "&" + CONFIDENTIAL + "&"
      + SECRET + ")", "(" + PRIVATE + "&" + CONFIDENTIAL + "&" + SECRET + ")", "(" + PRIVATE
      + "&" + CONFIDENTIAL + "&" + SECRET + ")")) {
    Scan s = new Scan();
    s.setAuthorizations(new Authorizations(TOPSECRET, CONFIDENTIAL, PRIVATE, PUBLIC, SECRET));
    ResultScanner scanner = table.getScanner(s);
    Result[] next = scanner.next(4);
    assertEquals(3, next.length);
    CellScanner cellScanner = next[0].cellScanner();
    cellScanner.advance();
    Cell current = cellScanner.current();
    assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(),
        current.getRowLength(), row2, 0, row2.length));
    cellScanner = next[1].cellScanner();
    cellScanner.advance();
    current = cellScanner.current();
    assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(),
        current.getRowLength(), row3, 0, row3.length));
    cellScanner = next[2].cellScanner();
    cellScanner.advance();
    current = cellScanner.current();
    assertTrue(Bytes.equals(current.getRowArray(), current.getRowOffset(),
        current.getRowLength(), row4, 0, row4.length));
  }
}

@Override
protected void verifyGet(final byte[] row, final String visString, final int expected,
    final boolean nullExpected, final String... auths) throws IOException,
    InterruptedException {
  PrivilegedExceptionAction<Void> scanAction = new PrivilegedExceptionAction<Void>() {
    public Void run() throws Exception {
      try (Connection connection = ConnectionFactory.createConnection(conf1);
           Table table2 = connection.getTable(TableName.valueOf(TABLE_NAME))) {
        CellScanner cellScanner;
        Cell current;
        Get get = new Get(row);
        get.setAuthorizations(new Authorizations(auths));
        Result result = table2.get(get);
        cellScanner = result.cellScanner();
        boolean advance = cellScanner.advance();
        if (nullExpected) {
          assertTrue(!advance);
          return null;
        }
        current = cellScanner.current();
        assertArrayEquals(CellUtil.cloneRow(current), row);
        assertEquals(expected, TestCoprocessorForTagsAtSink.tags.size());
        boolean foundNonVisTag = false;
        for(Tag t : TestCoprocessorForTagsAtSink.tags) {
          if(t.getType() == NON_VIS_TAG_TYPE) {
            assertEquals(TEMP, Bytes.toString(t.getValue()));
            foundNonVisTag = true;
            break;
          }
        }
        doAssert(row, visString);
        assertTrue(foundNonVisTag);
        return null;
      }
    }
  };
  USER1.runAs(scanAction);
}

@Override
public TestProtos.EchoResponseProto echo(RpcController controller,
                                         TestProtos.EchoRequestProto request)
    throws ServiceException {
  if (controller instanceof PayloadCarryingRpcController) {
    PayloadCarryingRpcController pcrc = (PayloadCarryingRpcController) controller;
    // If cells, scan them to check we are able to iterate what we were given and since
    // this is
    // an echo, just put them back on the controller creating a new block. Tests our
    // block
    // building.
    CellScanner cellScanner = pcrc.cellScanner();
    List<Cell> list = null;
    if (cellScanner != null) {
      list = new ArrayList<Cell>();
      try {
        while (cellScanner.advance()) {
          list.add(cellScanner.current());
        }
      } catch (IOException e) {
        throw new ServiceException(e);
      }
    }
    cellScanner = CellUtil.createCellScanner(list);
    ((PayloadCarryingRpcController) controller).setCellScanner(cellScanner);
  }
  return TestProtos.EchoResponseProto.newBuilder()
      .setMessage(request.getMessage()).build();
}

/**
 * Little test to check we can basically convert list of a list of KVs into a CellScanner
 * @throws IOException
 */
@Test
public void testGetCellScanner() throws IOException {
  List<Cell> a = new ArrayList<Cell>();
  KeyValue akv = new KeyValue(Bytes.toBytes("a"), -1L);
  a.add(akv);
  // Add a few just to make it less regular.
  a.add(new KeyValue(Bytes.toBytes("aa"), -1L));
  a.add(new KeyValue(Bytes.toBytes("aaa"), -1L));
  List<Cell> b = new ArrayList<Cell>();
  KeyValue bkv = new KeyValue(Bytes.toBytes("b"), -1L);
  a.add(bkv);
  List<Cell> c = new ArrayList<Cell>();
  KeyValue ckv = new KeyValue(Bytes.toBytes("c"), -1L);
  c.add(ckv);
  List<List<? extends Cell>> all = new ArrayList<List<? extends Cell>>();
  all.add(a);
  all.add(b);
  all.add(c);
  CellScanner scanner = ReplicationProtbufUtil.getCellScanner(all, 0);
  testAdvancetHasSameRow(scanner, akv);
  // Skip over aa
  scanner.advance();
  // Skip over aaa
  scanner.advance();
  testAdvancetHasSameRow(scanner, bkv);
  testAdvancetHasSameRow(scanner, ckv);
  assertFalse(scanner.advance());
}

private void assertSame(final CellScanner cellScanner, final Cell [] cells) throws IOException {
  int count = 0;
  while (cellScanner.advance()) {
    assertTrue(cells[count].equals(cellScanner.current()));
    count++;
  }
  assertEquals(cells.length, count);
}

static Cell [] runDecoderTest(final int index, final int count, final CellScanner decoder)
throws IOException {
  Cell [] cells = new Cell[count];
  long startTime = System.currentTimeMillis();
  for (int i = 0; decoder.advance(); i++) {
    cells[i] = decoder.current();
  }
  LOG.info("" + index + " decoded count=" + cells.length + " in " +
    (System.currentTimeMillis() - startTime) + "ms for decoder " + decoder);
  // Ensure we did not have to grow the backing buffer.
  assertTrue(cells.length == count);
  return cells;
}

@Override
public EchoResponseProto echo(RpcController controller, EchoRequestProto request)
    throws ServiceException {
  if (controller instanceof PayloadCarryingRpcController) {
    PayloadCarryingRpcController pcrc = (PayloadCarryingRpcController) controller;
    // If cells, scan them to check we are able to iterate what we were given and since
    // this is
    // an echo, just put them back on the controller creating a new block. Tests our
    // block
    // building.
    CellScanner cellScanner = pcrc.cellScanner();
    List<Cell> list = null;
    if (cellScanner != null) {
      list = new ArrayList<Cell>();
      try {
        while (cellScanner.advance()) {
          list.add(cellScanner.current());
        }
      } catch (IOException e) {
        throw new ServiceException(e);
      }
    }
    cellScanner = CellUtil.createCellScanner(list);
    ((PayloadCarryingRpcController) controller).setCellScanner(cellScanner);
  }
  return EchoResponseProto.newBuilder().setMessage(request.getMessage()).build();
}

/**
 * It is hard to verify the compression is actually happening under the wraps. Hope that if
 * unsupported, we'll get an exception out of some time (meantime, have to trace it manually to
 * confirm that compression is happening down in the client and server).
 * @throws IOException
 * @throws InterruptedException
 * @throws SecurityException
 * @throws NoSuchMethodException
 */
@Test
public void testCompressCellBlock() throws IOException, InterruptedException, SecurityException,
    NoSuchMethodException, ServiceException {
  Configuration conf = new Configuration(HBaseConfiguration.create());
  conf.set("hbase.client.rpc.compressor", GzipCodec.class.getCanonicalName());
  List<Cell> cells = new ArrayList<Cell>();
  int count = 3;
  for (int i = 0; i < count; i++) {
    cells.add(CELL);
  }
  AbstractRpcClient client = createRpcClient(conf);
  TestRpcServer rpcServer = new TestRpcServer();
  try {
    rpcServer.start();
    MethodDescriptor md = SERVICE.getDescriptorForType().findMethodByName("echo");
    EchoRequestProto param = EchoRequestProto.newBuilder().setMessage("hello").build();
    PayloadCarryingRpcController pcrc =
        new PayloadCarryingRpcController(CellUtil.createCellScanner(cells));
    InetSocketAddress address = rpcServer.getListenerAddress();
    if (address == null) {
      throw new IOException("Listener channel is closed");
    }
    Pair<Message, CellScanner> r =
        client.call(pcrc, md, param, md.getOutputType().toProto(), User.getCurrent(), address,
            new MetricsConnection.CallStats());
    int index = 0;
    while (r.getSecond().advance()) {
      assertTrue(CELL.equals(r.getSecond().current()));
      index++;
    }
    assertEquals(count, index);
  } finally {
    client.close();
    rpcServer.stop();
  }
}

/**
 * Convert a protocol buffer Result to a client Result
 *
 * @param proto the protocol buffer Result to convert
 * @param scanner Optional cell scanner.
 * @return the converted client Result
 * @throws IOException
 */
public static Result toResult(final ClientProtos.Result proto, final CellScanner scanner)
throws IOException {
  List<CellProtos.Cell> values = proto.getCellList();

  if (proto.hasExists()) {
    if ((values != null && !values.isEmpty()) ||
        (proto.hasAssociatedCellCount() && proto.getAssociatedCellCount() > 0)) {
      throw new IllegalArgumentException("bad proto: exists with cells is no allowed " + proto);
    }
    if (proto.getStale()) {
      return proto.getExists() ? EMPTY_RESULT_EXISTS_TRUE_STALE :EMPTY_RESULT_EXISTS_FALSE_STALE;
    }
    return proto.getExists() ? EMPTY_RESULT_EXISTS_TRUE : EMPTY_RESULT_EXISTS_FALSE;
  }

  // TODO: Unit test that has some Cells in scanner and some in the proto.
  List<Cell> cells = null;
  if (proto.hasAssociatedCellCount()) {
    int count = proto.getAssociatedCellCount();
    cells = new ArrayList<Cell>(count + values.size());
    for (int i = 0; i < count; i++) {
      if (!scanner.advance()) throw new IOException("Failed get " + i + " of " + count);
      cells.add(scanner.current());
    }
  }

  if (!values.isEmpty()){
    if (cells == null) cells = new ArrayList<Cell>(values.size());
    for (CellProtos.Cell c: values) {
      cells.add(toCell(c));
    }
  }

  return (cells == null || cells.isEmpty())
      ? (proto.getStale() ? EMPTY_RESULT_STALE : EMPTY_RESULT)
      : Result.create(cells, null, proto.getStale());
}

/**
 * Set success with a cellBlockScanner
 *
 * @param value            to set
 * @param cellBlockScanner to set
 */
public void setSuccess(Message value, CellScanner cellBlockScanner) {
  if (cellBlockScanner != null) {
    controller.setCellScanner(cellBlockScanner);
  }

  if (LOG.isTraceEnabled()) {
    long callTime = EnvironmentEdgeManager.currentTime() - startTime;
    LOG.trace("Call: " + method.getName() + ", callTime: " + callTime + "ms");
  }

  this.setSuccess(value);
}

/**
 * Make a blocking call. Throws exceptions if there are network problems or if the remote code
 * threw an exception.
 *
 * @param ticket Be careful which ticket you pass. A new user will mean a new Connection.
 *               {@link UserProvider#getCurrent()} makes a new instance of User each time so
 *               will be a
 *               new Connection each time.
 * @return A pair with the Message response and the Cell data (if any).
 */
Message callBlockingMethod(Descriptors.MethodDescriptor md, PayloadCarryingRpcController pcrc,
    Message param, Message returnType, final User ticket, final InetSocketAddress isa)
    throws ServiceException {
  if (pcrc == null) {
    pcrc = new PayloadCarryingRpcController();
  }

  Pair<Message, CellScanner> val;
  try {
    final MetricsConnection.CallStats cs = MetricsConnection.newCallStats();
    cs.setStartTime(EnvironmentEdgeManager.currentTime());
    val = call(pcrc, md, param, returnType, ticket, isa, cs);
    // Shove the results into controller so can be carried across the proxy/pb service void.
    pcrc.setCellScanner(val.getSecond());

    cs.setCallTimeMs(EnvironmentEdgeManager.currentTime() - cs.getStartTime());
    if (metrics != null) {
      metrics.updateRpc(md, param, cs);
    }
    if (LOG.isTraceEnabled()) {
      LOG.trace("Call: " + md.getName() + ", callTime: " + cs.getCallTimeMs() + "ms");
    }
    return val.getFirst();
  } catch (Throwable e) {
    throw new ServiceException(e);
  }
}

protected Call(int id, final Descriptors.MethodDescriptor md, Message param,
    final CellScanner cells, final Message responseDefaultType, int timeout,
    MetricsConnection.CallStats callStats) {
  this.param = param;
  this.md = md;
  this.cells = cells;
  this.callStats = callStats;
  this.callStats.setStartTime(EnvironmentEdgeManager.currentTime());
  this.responseDefaultType = responseDefaultType;
  this.id = id;
  this.timeout = timeout;
}

static void doBuildCellBlockUndoCellBlock(final IPCUtil util, final Codec codec,
  final CompressionCodec compressor, final int count, final int size, final boolean sized)
throws IOException {
  Cell [] cells = getCells(count, size);
  CellScanner cellScanner = sized? getSizedCellScanner(cells):
    CellUtil.createCellScanner(Arrays.asList(cells).iterator());
  ByteBuffer bb = util.buildCellBlock(codec, compressor, cellScanner);
  cellScanner = util.createCellScanner(codec, compressor, bb.array(), 0, bb.limit());
  int i = 0;
  while (cellScanner.advance()) {
    i++;
  }
  assertEquals(count, i);
}