Java 类org.apache.hadoop.hbase.client.Table 实例源码

private void runScanner(Table hTable, int expectedSize, Filter... filters) throws IOException {
  String cf = "f";
  Scan scan = new Scan();
  scan.addFamily(cf.getBytes());
  FilterList filterList = new FilterList(filters);
  scan.setFilter(filterList);

  ResultScanner scanner = hTable.getScanner(scan);
  List<Cell> results = new ArrayList<Cell>();
  Result result;
  long timeBeforeScan = System.currentTimeMillis();
  while ((result = scanner.next()) != null) {
    for (Cell kv : result.listCells()) {
      LOG.info("Got rk: " + Bytes.toStringBinary(CellUtil.cloneRow(kv)) + " cq: "
              + Bytes.toStringBinary(CellUtil.cloneQualifier(kv)));
      results.add(kv);
    }
  }
  long scanTime = System.currentTimeMillis() - timeBeforeScan;
  scanner.close();

  LOG.info("scan time = " + scanTime + "ms");
  LOG.info("found " + results.size() + " results");

  assertEquals(expectedSize, results.size());
}

private void deleteAndWait(byte[] row, Table source, Table... targets)
throws Exception {
  Delete del = new Delete(row);
  source.delete(del);

  Get get = new Get(row);
  for (int i = 0; i < NB_RETRIES; i++) {
    if (i==NB_RETRIES-1) {
      fail("Waited too much time for del replication");
    }
    boolean removedFromAll = true;
    for (Table target : targets) {
      Result res = target.get(get);
      if (res.size() >= 1) {
        LOG.info("Row not deleted");
        removedFromAll = false;
        break;
      }
    }
    if (removedFromAll) {
      break;
    } else {
      Thread.sleep(SLEEP_TIME);
    }
  }
}

/**
 * Removes a previously granted permission from the stored access control
 * lists.  The {@link TablePermission} being removed must exactly match what
 * is stored -- no wildcard matching is attempted.  Ie, if user "bob" has
 * been granted "READ" access to the "data" table, but only to column family
 * plus qualifier "info:colA", then trying to call this method with only
 * user "bob" and the table name "data" (but without specifying the
 * column qualifier "info:colA") will have no effect.
 *
 * @param conf the configuration
 * @param userPerm the details of the permission to be revoked
 * @throws IOException if there is an error accessing the metadata table
 */
static void removeUserPermission(Configuration conf, UserPermission userPerm)
    throws IOException {
  Delete d = new Delete(userPermissionRowKey(userPerm));
  byte[] key = userPermissionKey(userPerm);

  if (LOG.isDebugEnabled()) {
    LOG.debug("Removing permission "+ userPerm.toString());
  }
  d.addColumns(ACL_LIST_FAMILY, key);
  // TODO: Pass in a Connection rather than create one each time.
  try (Connection connection = ConnectionFactory.createConnection(conf)) {
    try (Table table = connection.getTable(ACL_TABLE_NAME)) {
      table.delete(d);
    }
  }
}

@Test (timeout=300000)
public void testIncrementHook() throws IOException {
  TableName tableName = TableName.valueOf(TEST_TABLE.getNameAsString() + ".testIncrementHook");
  Table table = util.createTable(tableName, new byte[][] {A, B, C});
  try {
    Increment inc = new Increment(Bytes.toBytes(0));
    inc.addColumn(A, A, 1);

    verifyMethodResult(SimpleRegionObserver.class,
        new String[] {"hadPreIncrement", "hadPostIncrement", "hadPreIncrementAfterRowLock"},
        tableName,
        new Boolean[] {false, false, false}
        );

    table.increment(inc);

    verifyMethodResult(SimpleRegionObserver.class,
        new String[] {"hadPreIncrement", "hadPostIncrement", "hadPreIncrementAfterRowLock"},
        tableName,
        new Boolean[] {true, true, true}
        );
  } finally {
    util.deleteTable(tableName);
    table.close();
  }
}

private static int insertData(TableName tableName, String column, double prob) throws IOException {
  byte[] k = new byte[3];
  byte[][] famAndQf = KeyValue.parseColumn(Bytes.toBytes(column));

  List<Put> puts = new ArrayList<>();
  for (int i = 0; i < 9; i++) {
    Put put = new Put(Bytes.toBytes("row" + i));
    put.setDurability(Durability.SKIP_WAL);
    put.add(famAndQf[0], famAndQf[1], k);
    put.setCellVisibility(new CellVisibility("(" + SECRET + "|" + CONFIDENTIAL + ")" + "&" + "!"
        + TOPSECRET));
    puts.add(put);
  }
  try (Table table = new HTable(TEST_UTIL.getConfiguration(), tableName)) {
    table.put(puts);
  }
  return puts.size();
}

/**
 * Returns all rows from the hbase:meta table for a given user table
 *
 * @throws IOException When reading the rows fails.
 */
public List<byte[]> getMetaTableRows(TableName tableName) throws IOException {
  // TODO: Redo using MetaTableAccessor.
  Table t = new HTable(new Configuration(this.conf), TableName.META_TABLE_NAME);
  List<byte[]> rows = new ArrayList<byte[]>();
  ResultScanner s = t.getScanner(new Scan());
  for (Result result : s) {
    HRegionInfo info = HRegionInfo.getHRegionInfo(result);
    if (info == null) {
      LOG.error("No region info for row " + Bytes.toString(result.getRow()));
      // TODO figure out what to do for this new hosed case.
      continue;
    }

    if (info.getTable().equals(tableName)) {
      LOG.info("getMetaTableRows: row -> " +
          Bytes.toStringBinary(result.getRow()) + info);
      rows.add(result.getRow());
    }
  }
  s.close();
  t.close();
  return rows;
}

boolean doCheckAndDelete(final byte[] row, final byte[] family, final byte[] qualifier, final byte[] value,
        final Delete delete) {
    boolean result;
    final Table tableInterface = getTable();
    try {
        result = doCheckAndDelete(tableInterface, row, family, qualifier, value, delete);
    } finally {
        closeTable(tableInterface);
    }
    return result;
}

/**
 * Tests that the LogRoller perform the roll even if there are no edits
 */
@Test
public void testNoEdits() throws Exception {
  TableName tableName = TableName.valueOf("TestLogRollPeriodNoEdits");
  TEST_UTIL.createTable(tableName, "cf");
  try {
    Table table = new HTable(TEST_UTIL.getConfiguration(), tableName);
    try {
      HRegionServer server = TEST_UTIL.getRSForFirstRegionInTable(tableName);
      WAL log = server.getWAL(null);
      checkMinLogRolls(log, 5);
    } finally {
      table.close();
    }
  } finally {
    TEST_UTIL.deleteTable(tableName);
  }
}

public static void blockUntilRegionIsOpened(Configuration conf, long timeout, HRegionInfo hri)
    throws IOException, InterruptedException {
  log("blocking until region is opened for reading:" + hri.getRegionNameAsString());
  long start = System.currentTimeMillis();
  try (Connection conn = ConnectionFactory.createConnection(conf);
      Table table = conn.getTable(hri.getTable())) {
    byte[] row = hri.getStartKey();
    // Check for null/empty row. If we find one, use a key that is likely to be in first region.
    if (row == null || row.length <= 0) row = new byte[] { '0' };
    Get get = new Get(row);
    while (System.currentTimeMillis() - start < timeout) {
      try {
        table.get(get);
        break;
      } catch (IOException ex) {
        // wait some more
      }
      Threads.sleep(10);
    }
  }
}

public static void assertMetaLocation(Table meta, byte[] row, ServerName serverName,
    long seqNum, int replicaId, boolean checkSeqNum) throws IOException {
  Get get = new Get(row);
  Result result = meta.get(get);
  assertTrue(Bytes.equals(
    result.getValue(HConstants.CATALOG_FAMILY, MetaTableAccessor.getServerColumn(replicaId)),
    Bytes.toBytes(serverName.getHostAndPort())));
  assertTrue(Bytes.equals(
    result.getValue(HConstants.CATALOG_FAMILY, MetaTableAccessor.getStartCodeColumn(replicaId)),
    Bytes.toBytes(serverName.getStartcode())));
  if (checkSeqNum) {
    assertTrue(Bytes.equals(
      result.getValue(HConstants.CATALOG_FAMILY, MetaTableAccessor.getSeqNumColumn(replicaId)),
      Bytes.toBytes(seqNum)));
  }
}

private void startAndWriteData() throws IOException, InterruptedException {
  // When the hbase:meta table can be opened, the region servers are running
  new HTable(TEST_UTIL.getConfiguration(), TableName.META_TABLE_NAME);
  this.server = cluster.getRegionServerThreads().get(0).getRegionServer();

  Table table = createTestTable(this.tableName);

  server = TEST_UTIL.getRSForFirstRegionInTable(table.getName());
  for (int i = 1; i <= 256; i++) {    // 256 writes should cause 8 log rolls
    doPut(table, i);
    if (i % 32 == 0) {
      // After every 32 writes sleep to let the log roller run
      try {
        Thread.sleep(2000);
      } catch (InterruptedException e) {
        // continue
      }
    }
  }
}

@Test(timeout = 60 * 1000)
public void testVisibilityLabelsOnWALReplay() throws Exception {
  final TableName tableName = TableName.valueOf(TEST_NAME.getMethodName());
  try (Table table = createTableAndWriteDataWithLabels(tableName,
      "(" + SECRET + "|" + CONFIDENTIAL + ")", PRIVATE);) {
    List<RegionServerThread> regionServerThreads = TEST_UTIL.getHBaseCluster()
        .getRegionServerThreads();
    for (RegionServerThread rsThread : regionServerThreads) {
      rsThread.getRegionServer().abort("Aborting ");
    }
    // Start one new RS
    RegionServerThread rs = TEST_UTIL.getHBaseCluster().startRegionServer();
    waitForLabelsRegionAvailability(rs.getRegionServer());
    Scan s = new Scan();
    s.setAuthorizations(new Authorizations(SECRET));
    ResultScanner scanner = table.getScanner(s);
    Result[] next = scanner.next(3);
    assertTrue(next.length == 1);
  }
}

/**
 * 列族过滤器
 *
 * @param tableName 表名
 * @param rowFamily 列族
 * @param count     数量
 */
public void familyFilter(String tableName, String rowFamily, int count) {
    HBaseConfiguration hBaseConfiguration = new HBaseConfiguration();
    Table table = hBaseConfiguration.table(tableName);
    Scan scan = new Scan();
    //使用列族过滤器
    //scan.setFilter(new FamilyFilter(CompareFilter.CompareOp.GREATER, new BinaryComparator(Bytes.toBytes(rowFamily))));//直接行健
    //scan.setFilter(new FamilyFilter(CompareFilter.CompareOp.GREATER_OR_EQUAL, new RegexStringComparator("row.*")));//正则表达式
    //scan.setFilter(new FamilyFilter(CompareFilter.CompareOp.GREATER_OR_EQUAL, new SubstringComparator("row")));//字符串包含
    scan.setFilter(new FamilyFilter(CompareFilter.CompareOp.GREATER_OR_EQUAL, new BinaryPrefixComparator("mm".getBytes())));//字符串前缀
    scan.setCaching(10);
    scan.setBatch(10);
    try {
        ResultScanner scanner = table.getScanner(scan);
        Result[] results = scanner.next(count);
        HBaseResultUtil.print(results);
    } catch (IOException e) {
        e.printStackTrace();
    }
}

/**
 * 列限定符过滤器
 *
 * @param tableName  表名
 * @param columnName 列限定符
 * @param count      数量
 */
public void qualifierFilter(String tableName, String columnName, int count) {
    HBaseConfiguration hBaseConfiguration = new HBaseConfiguration();
    Table table = hBaseConfiguration.table(tableName);
    Scan scan = new Scan();
    //使用列族过滤器
    scan.setFilter(new QualifierFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes(columnName))));//直接行健
    //scan.setFilter(new QualifierFilter(CompareFilter.CompareOp.EQUAL, new RegexStringComparator("row.*")));//正则表达式
    //scan.setFilter(new QualifierFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator("row")));//字符串包含
    //scan.setFilter(new QualifierFilter(CompareFilter.CompareOp.EQUAL, new BinaryPrefixComparator("m".getBytes())));//字符串前缀
    scan.setCaching(10);
    scan.setBatch(10);
    try {
        ResultScanner scanner = table.getScanner(scan);
        Result[] results = scanner.next(count);
        HBaseResultUtil.print(results);
    } catch (IOException e) {
        e.printStackTrace();
    }
}

@Override
public void increment(TIncrement tincrement) throws IOError, TException {

  if (tincrement.getRow().length == 0 || tincrement.getTable().length == 0) {
    throw new TException("Must supply a table and a row key; can't increment");
  }

  if (conf.getBoolean(COALESCE_INC_KEY, false)) {
    this.coalescer.queueIncrement(tincrement);
    return;
  }

  Table table = null;
  try {
    table = getTable(tincrement.getTable());
    Increment inc = ThriftUtilities.incrementFromThrift(tincrement);
    table.increment(inc);
  } catch (IOException e) {
    LOG.warn(e.getMessage(), e);
    throw new IOError(Throwables.getStackTraceAsString(e));
  } finally{
    closeTable(table);
  }
}

@Test
public void testMetaLocationForRegionReplicasIsAddedAtRegionMerge() throws IOException {
  long regionId = System.currentTimeMillis();
  ServerName serverName0 = ServerName.valueOf("foo", 60010, random.nextLong());

  HRegionInfo parentA = new HRegionInfo(TableName.valueOf("table_foo"),
    Bytes.toBytes("a"), HConstants.EMPTY_END_ROW, false, regionId, 0);
  HRegionInfo parentB = new HRegionInfo(TableName.valueOf("table_foo"),
    HConstants.EMPTY_START_ROW, Bytes.toBytes("a"), false, regionId, 0);
  HRegionInfo merged = new HRegionInfo(TableName.valueOf("table_foo"),
    HConstants.EMPTY_START_ROW, HConstants.EMPTY_END_ROW, false, regionId+1, 0);

  Table meta = MetaTableAccessor.getMetaHTable(connection);
  try {
    List<HRegionInfo> regionInfos = Lists.newArrayList(parentA, parentB);
    MetaTableAccessor.addRegionsToMeta(connection, regionInfos, 3);

    MetaTableAccessor.mergeRegions(connection, merged, parentA, parentB, serverName0, 3,
        HConstants.LATEST_TIMESTAMP);

    assertEmptyMetaLocation(meta, merged.getRegionName(), 1);
    assertEmptyMetaLocation(meta, merged.getRegionName(), 2);
  } finally {
    meta.close();
  }
}

/**
 * Obtain and return an authentication token for the current user.
 * @param conn The HBase cluster connection
 * @return the authentication token instance
 */
public static Token<AuthenticationTokenIdentifier> obtainToken(
    Connection conn) throws IOException {
  Table meta = null;
  try {
    meta = conn.getTable(TableName.META_TABLE_NAME);
    CoprocessorRpcChannel rpcChannel = meta.coprocessorService(HConstants.EMPTY_START_ROW);
    AuthenticationProtos.AuthenticationService.BlockingInterface service =
        AuthenticationProtos.AuthenticationService.newBlockingStub(rpcChannel);
    AuthenticationProtos.GetAuthenticationTokenResponse response = service.getAuthenticationToken(null,
        AuthenticationProtos.GetAuthenticationTokenRequest.getDefaultInstance());

    return ProtobufUtil.toToken(response.getToken());
  } catch (ServiceException se) {
    ProtobufUtil.toIOException(se);
  } finally {
    if (meta != null) {
      meta.close();
    }
  }
  // dummy return for ServiceException block
  return null;
}

@BeforeClass
public static void prepareData() {
  try {
    createTable(tableName, columnFamily);
    Table table = openTable(tableName);
    List<Put> puts = new ArrayList<Put>();

    // row1 => <f1:c1, 1_c1>, <f1:c2, 1_c2>, <f1:c3, 1_c3>, <f1:c4,1_c4>,
    // <f1:c5, 1_c5>
    // row2 => <f1:c1, 2_c1>, <f1:c2, 2_c2>, <f1:c3, 2_c3>, <f1:c4,2_c4>,
    // <f1:c5, 2_c5>
    for (int i = 1; i < 4; i++) {
      Put put = new Put(Bytes.toBytes("row" + i));
      for (int j = 1; j < 6; j++) {
        put.add(Bytes.toBytes("f1"), Bytes.toBytes("c" + j),
            Bytes.toBytes(i + "_c" + j));
      }
      puts.add(put);
    }

    table.put(puts);
    table.close();
  } catch (IOException e) {
    assertNull("Exception found while putting data into table", e);
  }
}

private void runTest(Table hTable, int cqStart, int expectedSize) throws IOException {
  // [0, 2, ?, ?, ?, ?, 0, 0, 0, 1]
  byte[] fuzzyKey = new byte[10];
  ByteBuffer buf = ByteBuffer.wrap(fuzzyKey);
  buf.clear();
  buf.putShort((short) 2);
  for (int i = 0; i < 4; i++)
    buf.put((byte)63);
  buf.putInt((short)1);

  byte[] mask = new byte[] {0 , 0, 1, 1, 1, 1, 0, 0, 0, 0};

  Pair<byte[], byte[]> pair = new Pair<byte[], byte[]>(fuzzyKey, mask);
  FuzzyRowFilter fuzzyRowFilter = new FuzzyRowFilter(Lists.newArrayList(pair));
  ColumnRangeFilter columnRangeFilter = new ColumnRangeFilter(Bytes.toBytes(cqStart), true
          , Bytes.toBytes(4), true);
  //regular test
  runScanner(hTable, expectedSize, fuzzyRowFilter, columnRangeFilter);
  //reverse filter order test
  runScanner(hTable, expectedSize, columnRangeFilter, fuzzyRowFilter);
}

private Map<byte [], String> noop(final Table table, final byte [] start,
    final byte [] end)
throws ServiceException, Throwable {
  return table.coprocessorService(PingProtos.PingService.class, start, end,
      new Batch.Call<PingProtos.PingService, String>() {
        @Override
        public String call(PingProtos.PingService instance) throws IOException {
          BlockingRpcCallback<PingProtos.NoopResponse> rpcCallback =
            new BlockingRpcCallback<PingProtos.NoopResponse>();
          PingProtos.NoopRequest.Builder builder = PingProtos.NoopRequest.newBuilder();
          instance.noop(null, builder.build(), rpcCallback);
          rpcCallback.get();
          // Looks like null is expected when void.  That is what the test below is looking for
          return null;
        }
      });
}

@Test
public void testWithEmptyTable() throws Exception {
  Admin admin = HBaseTestsSuite.getAdmin();
  TableName tableName = TableName.valueOf("dremio_ut_empty_table");

  try (Table table = HBaseTestsSuite.getConnection().getTable(tableName);) {
    HTableDescriptor desc = new HTableDescriptor(tableName);
    desc.addFamily(new HColumnDescriptor("f"));
    admin.createTable(desc, Arrays.copyOfRange(TestTableGenerator.SPLIT_KEYS, 0, 2));

    setColumnWidths(new int[] {8, 15});
    runHBaseSQLVerifyCount("SELECT row_key, count(*)\n"
        + "FROM\n"
        + "  hbase.`" + tableName + "` tableName GROUP BY row_key\n"
        , 0);
  } finally {
    try {
      admin.disableTable(tableName);
      admin.deleteTable(tableName);
    } catch (Exception e) { } // ignore
  }
}

/**
 * Tests the case where if there is some data in the primary region, reopening the region replicas
 * (enable/disable table, etc) makes the region replicas readable.
 * @throws IOException
 */
@Test(timeout = 60000)
public void testSecondaryRegionWithNonEmptyRegion() throws IOException {
  // Create a new table with region replication and load some data
  // than disable and enable the table again and verify the data from secondary
  try (Connection connection = ConnectionFactory.createConnection(HTU.getConfiguration());
      Table table = connection.getTable(htd.getTableName())) {

    HTU.loadNumericRows(table, fam, 0, 1000);

    HTU.getHBaseAdmin().disableTable(htd.getTableName());
    HTU.getHBaseAdmin().enableTable(htd.getTableName());

    HTU.verifyNumericRows(table, fam, 0, 1000, 1);
  }
}

@Test (timeout=300000)
public void testAppendHook() throws IOException {
  TableName tableName = TableName.valueOf(TEST_TABLE.getNameAsString() + ".testAppendHook");
  Table table = util.createTable(tableName, new byte[][] {A, B, C});
  try {
    Append app = new Append(Bytes.toBytes(0));
    app.add(A, A, A);

    verifyMethodResult(SimpleRegionObserver.class,
        new String[] {"hadPreAppend", "hadPostAppend", "hadPreAppendAfterRowLock"},
        tableName,
        new Boolean[] {false, false, false}
        );

    table.append(app);

    verifyMethodResult(SimpleRegionObserver.class,
        new String[] {"hadPreAppend", "hadPostAppend", "hadPreAppendAfterRowLock"},
        tableName,
        new Boolean[] {true, true, true}
        );
  } finally {
    util.deleteTable(tableName);
    table.close();
  }
}

@Test
public void testGetForSuperUserWithFewerLabelAuths() throws Throwable {
  String[] auths = { SECRET };
  String user = "admin";
  VisibilityClient.setAuths(TEST_UTIL.getConnection(), auths, user);
  TableName tableName = TableName.valueOf(TEST_NAME.getMethodName());
  final Table table = createTableAndWriteDataWithLabels(tableName, SECRET + "&" + CONFIDENTIAL
      + "&!" + PRIVATE, SECRET + "&!" + PRIVATE);
  PrivilegedExceptionAction<Void> scanAction = new PrivilegedExceptionAction<Void>() {
    public Void run() throws Exception {
      Get g = new Get(row1);
      g.setAuthorizations(new Authorizations(SECRET, CONFIDENTIAL));
      try (Connection connection = ConnectionFactory.createConnection(conf);
           Table t = connection.getTable(table.getName())) {
        Result result = t.get(g);
        assertTrue(!result.isEmpty());
      }
      return null;
    }
  };
  SUPERUSER.runAs(scanAction);
}

public static Table createTableAndWriteDataWithLabels(TableName tableName, String... labelExps)
    throws Exception {
  Table table = null;
  table = TEST_UTIL.createTable(tableName, fam);
  int i = 1;
  List<Put> puts = new ArrayList<Put>();
  for (String labelExp : labelExps) {
    Put put = new Put(Bytes.toBytes("row" + i));
    put.add(fam, qual, HConstants.LATEST_TIMESTAMP, value);
    put.setCellVisibility(new CellVisibility(labelExp));
    puts.add(put);
    table.put(put);
    i++;
  }
  // table.put(puts);
  return table;
}

@Override
public boolean checkAndDelete(ByteBuffer table, ByteBuffer row, ByteBuffer family,
    ByteBuffer qualifier, ByteBuffer value, TDelete deleteSingle) throws TIOError, TException {
  Table htable = getTable(table);

  try {
    if (value == null) {
      return htable.checkAndDelete(byteBufferToByteArray(row), byteBufferToByteArray(family),
        byteBufferToByteArray(qualifier), null, deleteFromThrift(deleteSingle));
    } else {
      return htable.checkAndDelete(byteBufferToByteArray(row), byteBufferToByteArray(family),
        byteBufferToByteArray(qualifier), byteBufferToByteArray(value),
        deleteFromThrift(deleteSingle));
    }
  } catch (IOException e) {
    throw getTIOError(e);
  } finally {
    closeTable(htable);
  }
}

@Test
public void testMetaLocationForRegionReplicasIsAddedAtTableCreation() throws IOException {
  long regionId = System.currentTimeMillis();
  HRegionInfo primary = new HRegionInfo(TableName.valueOf("table_foo"),
    HConstants.EMPTY_START_ROW, HConstants.EMPTY_END_ROW, false, regionId, 0);

  Table meta = MetaTableAccessor.getMetaHTable(connection);
  try {
    List<HRegionInfo> regionInfos = Lists.newArrayList(primary);
    MetaTableAccessor.addRegionsToMeta(connection, regionInfos, 3);

    assertEmptyMetaLocation(meta, primary.getRegionName(), 1);
    assertEmptyMetaLocation(meta, primary.getRegionName(), 2);
  } finally {
    meta.close();
  }
}

/**
 * Checks that all columns have the expected value and that there is the
 * expected number of rows.
 * @throws IOException
 */
void assertExpectedTable(final Connection connection, TableName table, int count, int value)
throws IOException {
  HTableDescriptor [] htds = util.getHBaseAdmin().listTables(table.getNameAsString());
  assertEquals(htds.length, 1);
  Table t = null;
  try {
    t = connection.getTable(table);
    Scan s = new Scan();
    ResultScanner sr = t.getScanner(s);
    int i = 0;
    for (Result r : sr) {
      i++;
      for (NavigableMap<byte[], byte[]> nm : r.getNoVersionMap().values()) {
        for (byte[] val : nm.values()) {
          assertTrue(Bytes.equals(val, value(value)));
        }
      }
    }
    assertEquals(count, i);
  } catch (IOException e) {
    fail("Failed due to exception");
  } finally {
    if (t != null) t.close();
  }
}

public void testBulkDeleteEndpointWhenRowBatchSizeLessThanRowsToDeleteFromARegion()
    throws Throwable {
  TableName tableName = TableName
      .valueOf("testBulkDeleteEndpointWhenRowBatchSizeLessThanRowsToDeleteFromARegion");
  Table ht = createTable(tableName);
  List<Put> puts = new ArrayList<Put>(100);
  for (int j = 0; j < 100; j++) {
    byte[] rowkey = Bytes.toBytes(j);
    puts.add(createPut(rowkey, "v1"));
  }
  ht.put(puts);
  // Deleting all the rows.
  long noOfRowsDeleted = invokeBulkDeleteProtocol(tableName, new Scan(), 10, DeleteType.ROW, null);
  assertEquals(100, noOfRowsDeleted);

  int rows = 0;
  for (Result result : ht.getScanner(new Scan())) {
    rows++;
  }
  assertEquals(0, rows);
  ht.close();
}

public IndexChooser(final IndexTable indexTable) throws IOException {
  this.indexTable = indexTable;
  indexRegionMaps = new TreeMap<byte[], List<HRegionInfo>>(Bytes.BYTES_COMPARATOR);

  for (Map.Entry<byte[], Table> entry : indexTable.getIndexTableMaps().entrySet()) {
    if (!(entry.getValue() instanceof HTable)) {
      throw new IOException(
          "table is not an instance of HTable, it is " + entry.getValue().getClass().getName());
    }
    HTable htable = (HTable) entry.getValue();
    ArrayList<HRegionInfo> list =
        new ArrayList<HRegionInfo>(htable.getRegionLocations().keySet());
    indexRegionMaps.put(entry.getKey(), list);
  }

  speedTimes = DEFAULT_SPEED_TIMES;
}

private Map<byte [], String> hello(final Table table, final String send, final byte [] start,
    final byte [] end)
throws ServiceException, Throwable {
  return table.coprocessorService(PingProtos.PingService.class,
      start, end,
      new Batch.Call<PingProtos.PingService, String>() {
        @Override
        public String call(PingProtos.PingService instance) throws IOException {
          BlockingRpcCallback<PingProtos.HelloResponse> rpcCallback =
            new BlockingRpcCallback<PingProtos.HelloResponse>();
          PingProtos.HelloRequest.Builder builder = PingProtos.HelloRequest.newBuilder();
          if (send != null) builder.setName(send);
          instance.hello(null, builder.build(), rpcCallback);
          PingProtos.HelloResponse r = rpcCallback.get();
          return r != null && r.hasResponse()? r.getResponse(): null;
        }
      });
}

protected HRegionInfo createRegion(Configuration conf, final Table htbl,
    byte[] startKey, byte[] endKey) throws IOException {
  Table meta = new HTable(conf, TableName.META_TABLE_NAME);
  HTableDescriptor htd = htbl.getTableDescriptor();
  HRegionInfo hri = new HRegionInfo(htbl.getName(), startKey, endKey);

  LOG.info("manually adding regioninfo and hdfs data: " + hri.toString());
  Path rootDir = FSUtils.getRootDir(conf);
  FileSystem fs = rootDir.getFileSystem(conf);
  Path p = new Path(FSUtils.getTableDir(rootDir, htbl.getName()),
      hri.getEncodedName());
  fs.mkdirs(p);
  Path riPath = new Path(p, HRegionFileSystem.REGION_INFO_FILE);
  FSDataOutputStream out = fs.create(riPath);
  out.write(hri.toDelimitedByteArray());
  out.close();

  // add to meta.
  MetaTableAccessor.addRegionToMeta(meta, hri);
  meta.close();
  return hri;
}

public static Table createTable(HBaseTestingUtility testUtil, TableName tableName,
    byte[][] families) throws Exception {
  HTableDescriptor htd = new HTableDescriptor(tableName);
  for (byte[] family : families) {
    HColumnDescriptor hcd = new HColumnDescriptor(family);
    htd.addFamily(hcd);
  }
  createTable(testUtil, testUtil.getHBaseAdmin(), htd);
  return testUtil.getConnection().getTable(htd.getTableName());
}

/**
 * delete record
 */
public static void delRecord(String tableName, String rowKey)
        throws IOException {
    Table table = connection.getTable(TableName.valueOf(tableName));
    List list = new ArrayList();
    Delete del = new Delete(rowKey.getBytes());
    list.add(del);
    table.delete(list);
    System.out.println("del recored " + rowKey + " ok.");
}

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

/**
 * @param t Table to use (will be closed when done).
 * @param g Get to run
 * @throws IOException
 */
private static Result get(final Table t, final Get g) throws IOException {
  try {
    return t.get(g);
  } finally {
    t.close();
  }
}

public synchronized void update(NamespaceDescriptor ns) throws IOException {
  Table table = getNamespaceTable();
  if (get(table, ns.getName()) == null) {
    throw new NamespaceNotFoundException(ns.getName());
  }
  upsert(table, ns);
}

boolean doCheckAndPut(final byte[] row, final byte[] family, final byte[] qualifier, final byte[] value,
        final Put put) {
    boolean result;
    final Table tableInterface = getTable();
    try {
        result = doCheckAndPut(tableInterface, row, family, qualifier, value, put);
    } finally {
        closeTable(tableInterface);
    }
    return result;
}

public static boolean doCheckAndDelete(final Table tableInterface, final byte[] row, final byte[] family,
        final byte[] qualifier, final byte[] value, final Delete delete) {
    boolean result;
    try {
        result = tableInterface.checkAndDelete(row, family, qualifier, value, delete);
    } catch (final Exception e) {
        closeTable(tableInterface);
        throw new HBaseException(e.getMessage(), e);
    }
    return result;
}

/**
 * Gets a Table for this table, does the get and closes the Table
 */
public Result doGet(final Get get) {
    Result result;
    final Table tableInterface = getTable();
    try {
        result = doGet(tableInterface, get);
    } finally {
        closeTable(tableInterface);
    }
    return result;
}