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

public static void main(String[] argc) throws Exception {
  Configuration conf = HBaseConfiguration.create();
  conf.addResource(new Path("file:///", System.getProperty("oozie.action.conf.xml")));

  if (System.getenv("HADOOP_TOKEN_FILE_LOCATION") != null) {
    conf.set("mapreduce.job.credentials.binary",
             System.getenv("HADOOP_TOKEN_FILE_LOCATION"));
  }

  Connection connection = ConnectionFactory.createConnection(conf);
  Admin admin = connection.getAdmin();

  System.out.println("Compacting table " + argc[0]);
  TableName tableName = TableName.valueOf(argc[0]);
  admin.majorCompact(tableName);
  while (admin.getCompactionState(tableName).toString() == "MAJOR") {
    TimeUnit.SECONDS.sleep(10);
    System.out.println("Compacting table " + argc[0]);
  }
  System.out.println("Done compacting table " + argc[0]);
}

private void verifyUserAllowedforCheckAndDelete(final User user, final byte[] row,
    final byte[] q1, final byte[] value) throws IOException, InterruptedException {
  user.runAs(new PrivilegedExceptionAction<Void>() {
    @Override
    public Void run() throws Exception {
      try (Connection connection = ConnectionFactory.createConnection(conf)) {
        try (Table t = connection.getTable(TEST_TABLE.getTableName())) {
          Delete d = new Delete(row);
          d.addColumn(TEST_FAMILY1, q1, 120);
          t.checkAndDelete(row, TEST_FAMILY1, q1, value, d);
        }
      }
      return null;
    }
  });
}

private void verifyUserDeniedForDeleteExactVersion(final User user, final byte[] row,
    final byte[] q1, final byte[] q2) throws IOException, InterruptedException {
  user.runAs(new PrivilegedExceptionAction<Void>() {
    @Override
    public Void run() throws Exception {
      try (Connection connection = ConnectionFactory.createConnection(conf)) {
        try (Table t = connection.getTable(TEST_TABLE.getTableName())) {
          Delete d = new Delete(row, 127);
          d.addColumns(TEST_FAMILY1, q1);
          d.addColumns(TEST_FAMILY1, q2);
          d.addFamily(TEST_FAMILY2, 129);
          t.delete(d);
          fail(user.getShortName() + " can not do the delete");
        } catch (Exception e) {

        }
      }
      return null;
    }
  });
}

/**
 * Note that all subclasses of this class must provide a public constructor
 * that has the exact same list of arguments.
 */
Test(final Connection con, final TestOptions options, final Status status) {
  this.connection = con;
  this.conf = con == null ? HBaseConfiguration.create() : this.connection.getConfiguration();
  this.opts = options;
  this.status = status;
  this.testName = this.getClass().getSimpleName();
  receiverHost = SpanReceiverHost.getInstance(conf);
  if (options.traceRate >= 1.0) {
    this.traceSampler = Sampler.ALWAYS;
  } else if (options.traceRate > 0.0) {
    conf.setDouble("hbase.sampler.fraction", options.traceRate);
    this.traceSampler = new ProbabilitySampler(new HBaseHTraceConfiguration(conf));
  } else {
    this.traceSampler = Sampler.NEVER;
  }
  everyN = (int) (opts.totalRows / (opts.totalRows * opts.sampleRate));
  if (options.isValueZipf()) {
    this.zipf = new RandomDistribution.Zipf(this.rand, 1, options.getValueSize(), 1.1);
  }
  LOG.info("Sampling 1 every " + everyN + " out of " + opts.perClientRunRows + " total rows.");
}

private void validateFromSnapshotFromMeta(HBaseTestingUtility util, TableName table,
    int numRegions, int numReplica, Connection connection) throws IOException {
  SnapshotOfRegionAssignmentFromMeta snapshot = new SnapshotOfRegionAssignmentFromMeta(
    connection);
  snapshot.initialize();
  Map<HRegionInfo, ServerName> regionToServerMap = snapshot.getRegionToRegionServerMap();
  assert(regionToServerMap.size() == numRegions * numReplica + 1); //'1' for the namespace
  Map<ServerName, List<HRegionInfo>> serverToRegionMap = snapshot.getRegionServerToRegionMap();
  for (Map.Entry<ServerName, List<HRegionInfo>> entry : serverToRegionMap.entrySet()) {
    if (entry.getKey().equals(util.getHBaseCluster().getMaster().getServerName())) {
      continue;
    }
    List<HRegionInfo> regions = entry.getValue();
    Set<byte[]> setOfStartKeys = new HashSet<byte[]>();
    for (HRegionInfo region : regions) {
      byte[] startKey = region.getStartKey();
      if (region.getTable().equals(table)) {
        setOfStartKeys.add(startKey); //ignore other tables
        LOG.info("--STARTKEY " + new String(startKey)+"--");
      }
    }
    // the number of startkeys will be equal to the number of regions hosted in each server
    // (each server will be hosting one replica of a region)
    assertEquals(numRegions, setOfStartKeys.size());
  }
}

/**
 * [ { "time": 1456293824385, "host": "09-201509070105", "ip": "127.0.0.1", "svrid":
 * "D:/UAV/apache-tomcat-6.0.41::D:/eclipseProject/.metadata/.plugins/org.eclipse.wst.server.core/tmp0", "tag": "L",
 * "frames": { "WebTest": [ { "content": "[CE] aaaaa" } ] } } ]
 */
@SuppressWarnings("unchecked")
public static void testInsertHBase() {

    // MongoDBHandler
    DataStoreMsg msg = new DataStoreMsg();
    String rawData = DataStoreUnitTest.getData(insertJson);
    msg.put(MonitorDataFrame.MessageType.Log.toString(), rawData);
    msg.put(DataStoreProtocol.HBASE_TABLE_NAME, HealthManagerConstants.HBASE_TABLE_LOGDATA);
    List<String> servers = DataConvertHelper.toList(zklist, ",");
    DataStoreConnection obj = new DataStoreConnection(null, null, null, servers, DataStoreType.HBASE);
    obj.putContext(DataStoreProtocol.HBASE_ZK_QUORUM, zklist);
    obj.putContext(DataStoreProtocol.HBASE_QUERY_CACHING, caching);
    obj.putContext(DataStoreProtocol.HBASE_QUERY_MAXRESULTSIZE, maxResultSize);

    obj.putContext(DataStoreProtocol.HBASE_QUERY_REVERSE, true);
    obj.putContext(DataStoreProtocol.HBASE_QUERY_PAGESIZE, 3000);

    AbstractDataStore<Connection> store = DataStoreFactory.getInstance().build(HealthManagerConstants.DataStore_Log,
            obj, new LogDataAdapter(), "");

    store.start();
    boolean rst = store.doInsert(msg);
    store.stop();
    DataStoreUnitTest.printTestResult("testInsertHBase", rst);
}

/**
 * { "starttime": 145629382438, "endtime": 145629382438, //optional "ip": "127.0.0.1", "svrid":
 * "D:/UAV/apache-tomcat-6.0.41::D:/eclipseProject/.metadata/.plugins/org.eclipse.wst.server.core/tmp0", "appid":
 * "sms" }
 */
@SuppressWarnings("unchecked")
public static void testQueryHBase() {

    DataStoreMsg msg = new DataStoreMsg();
    msg.put(DataStoreProtocol.HBASE_QUERY_JSON_KEY, queryJson);
    List<String> servers = DataConvertHelper.toList(zklist, ",");
    DataStoreConnection obj = new DataStoreConnection(null, null, null, servers, DataStoreType.HBASE);
    obj.putContext(DataStoreProtocol.HBASE_ZK_QUORUM, zklist);
    obj.putContext(DataStoreProtocol.HBASE_QUERY_CACHING, caching);
    obj.putContext(DataStoreProtocol.HBASE_QUERY_MAXRESULTSIZE, maxResultSize);
    AbstractDataStore<Connection> store = DataStoreFactory.getInstance().build(HealthManagerConstants.DataStore_Log,
            obj, new LogDataAdapter(), "");
    store.start();
    List<String> rst = store.doQuery(msg);
    store.stop();
    DataStoreUnitTest.printTestResult("testqueryHBase", rst, queryJson);
}

public Connection getConnection() throws IOException {
    Connection resultConn;
    if (usableSize != 0) {
        resultConn = conns.remove(0);
        usableSize --;
    } else if (currentSize < maxSize) {
        resultConn = ConnectionFactory.createConnection(conf);
        currentSize ++;
    } else {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            logger.error("获取HBASE连接出错了！\n"+e.getMessage());
        }
        return this.getConnection();
    }
    return resultConn;
}

@Test (timeout=180000)
public void testAppend() throws Exception {

  AccessTestAction appendAction = new AccessTestAction() {
    @Override
    public Object run() throws Exception {
      byte[] row = TEST_ROW;
      byte[] qualifier = TEST_QUALIFIER;
      Put put = new Put(row);
      put.add(TEST_FAMILY, qualifier, Bytes.toBytes(1));
      Append append = new Append(row);
      append.add(TEST_FAMILY, qualifier, Bytes.toBytes(2));
      try(Connection conn = ConnectionFactory.createConnection(conf);
          Table t = conn.getTable(TEST_TABLE)) {
        t.put(put);
        t.append(append);
      }
      return null;
    }
  };

  verifyAllowed(appendAction, SUPERUSER, USER_ADMIN, USER_OWNER, USER_CREATE, USER_RW,
    USER_GROUP_WRITE);
  verifyDenied(appendAction, USER_RO, USER_NONE, USER_GROUP_CREATE, USER_GROUP_READ,
    USER_GROUP_ADMIN);
}

@Test
public void testManualHBaseInsertion() throws ServiceException, IOException {
  IgniteConfiguration cfg = prepareConfig(false);
  IgniteConfiguration cfg2 = new IgniteConfiguration(cfg);
  cfg.setGridName("first");
  cfg2.setGridName("second");
  String cacheName = "myCache";
  try (Ignite ignite = Ignition.getOrStart(cfg); Ignite ignite2 = Ignition.getOrStart(cfg2)) {
    IgniteCache<String, String> cache = ignite.getOrCreateCache(cacheName);
    cache.remove("Hello");
    assertNull(cache.get("Hello"));
    try (Connection conn = getHBaseConnection()) {
      TableName tableName = TableName.valueOf(TABLE_NAME);
      Table table = conn.getTable(tableName);
      Serializer<Object> serializer = ObjectSerializer.INSTANCE;
      Put put = new Put(serializer.serialize("Hello"));
      put.addColumn(cacheName.getBytes(), QUALIFIER, serializer.serialize("World"));
      table.put(put);
    }
    assertEquals("World", cache.get("Hello"));
  }
}

public static void checkTablePerms(HBaseTestingUtility testUtil, TableName table,
    Permission... perms) throws IOException {
  CheckPermissionsRequest.Builder request = CheckPermissionsRequest.newBuilder();
  for (Permission p : perms) {
    request.addPermission(ProtobufUtil.toPermission(p));
  }

  try(Connection conn = ConnectionFactory.createConnection(testUtil.getConfiguration());
      Table acl = conn.getTable(table)) {
    AccessControlService.BlockingInterface protocol =
      AccessControlService.newBlockingStub(acl.coprocessorService(new byte[0]));
    try {
      protocol.checkPermissions(null, request.build());
    } catch (ServiceException se) {
      ProtobufUtil.toIOException(se);
    }
  }
}

/**
 * Put the passed <code>ps</code> to the <code>hbase:meta</code> table.
 * @param connection connection we're using
 * @param ps Put to add to hbase:meta
 * @throws IOException
 */
public static void putsToMetaTable(final Connection connection, final List<Put> ps)
  throws IOException {
  Table t = getMetaHTable(connection);
  try {
    t.put(ps);
  } finally {
    t.close();
  }
}

@Override
public void configure(JobConf job) {
  try {
    Connection connection = ConnectionFactory.createConnection(HBaseConfiguration.create(job));
    TableName tableName = TableName.valueOf("exampleJobConfigurableTable");
    // mandatory
    initializeTable(connection, tableName);
    byte[][] inputColumns = new byte [][] { Bytes.toBytes("columnA"),
      Bytes.toBytes("columnB") };
    //optional
    Scan scan = new Scan();
    for (byte[] family : inputColumns) {
      scan.addFamily(family);
    }
    Filter exampleFilter = new RowFilter(CompareOp.EQUAL, new RegexStringComparator("aa.*"));
    scan.setFilter(exampleFilter);
    setScan(scan);
  } catch (IOException exception) {
    throw new RuntimeException("Failed to initialize.", exception);
  }
}

@Test(expected=RetriesExhaustedException.class)
public void testSocketClosed() throws IOException, InterruptedException {
  String tableName = "testSocketClosed";
  TableName name = TableName.valueOf(tableName);
  UTIL.createTable(name, fam1).close();

  Configuration conf = new Configuration(UTIL.getConfiguration());
  conf.set(RpcClientFactory.CUSTOM_RPC_CLIENT_IMPL_CONF_KEY,
    MyRpcClientImpl.class.getName());
  conf.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 2);
  Connection connection = ConnectionFactory.createConnection(conf);
  Table table = connection.getTable(TableName.valueOf(tableName));
  table.get(new Get("asd".getBytes()));
  connection.close();
  for (Socket socket : MyRpcClientImpl.savedSockets) {
    assertTrue("Socket + " +  socket + " is not closed", socket.isClosed());
  }
}

private void verifyUserDeniedForPutMultipleVersions(final User user, final byte[] row,
    final byte[] q1, final byte[] q2, final byte[] value) throws IOException,
    InterruptedException {
  user.runAs(new PrivilegedExceptionAction<Void>() {
    @Override
    public Void run() throws Exception {
      try (Connection connection = ConnectionFactory.createConnection(conf)) {
        try (Table t = connection.getTable(TEST_TABLE.getTableName())) {
          Put p = new Put(row);
          // column Q1 covers version at 123 fr which user2 do not have permission
          p.addColumn(TEST_FAMILY1, q1, 124, value);
          p.addColumn(TEST_FAMILY1, q2, value);
          t.put(p);
          fail(user.getShortName() + " cannot do the put.");
        } catch (Exception e) {

        }
      }
      return null;
    }
  });
}

@Override public int run(String[] args) throws Exception {
  if (args.length < 1) {
    System.err.println("Usage: Clean <output dir>");
    return -1;
  }

  Path p = new Path(args[0]);
  Configuration conf = getConf();
  TableName tableName = getTableName(conf);
  try (FileSystem fs = HFileSystem.get(conf);
      Connection conn = ConnectionFactory.createConnection(conf);
      Admin admin = conn.getAdmin()) {
    if (admin.tableExists(tableName)) {
      admin.disableTable(tableName);
      admin.deleteTable(tableName);
    }

    if (fs.exists(p)) {
      fs.delete(p, true);
    }
  }

  return 0;
}

/**
 * Creates a table with given table name and specified number of column
 * families if the table does not already exist.
 */
private void setupTable(final Connection connection, TableName table, int cfs)
throws IOException {
  try {
    LOG.info("Creating table " + table);
    HTableDescriptor htd = new HTableDescriptor(table);
    for (int i = 0; i < cfs; i++) {
      htd.addFamily(new HColumnDescriptor(family(i)));
    }
    try (Admin admin = connection.getAdmin()) {
      admin.createTable(htd);
    }
  } catch (TableExistsException tee) {
    LOG.info("Table " + table + " already exists");
  }
}

private void bulkLoadHFile(
    TableName tableName,
    byte[] family,
    byte[] qualifier,
    byte[][][] hfileRanges,
    int numRowsPerRange) throws Exception {

  Path familyDir = new Path(loadPath, Bytes.toString(family));
  fs.mkdirs(familyDir);
  int hfileIdx = 0;
  for (byte[][] range : hfileRanges) {
    byte[] from = range[0];
    byte[] to = range[1];
    createHFile(new Path(familyDir, "hfile_"+(hfileIdx++)),
        family, qualifier, from, to, numRowsPerRange);
  }
  //set global read so RegionServer can move it
  setPermission(loadPath, FsPermission.valueOf("-rwxrwxrwx"));

  try (Connection conn = ConnectionFactory.createConnection(conf);
       HTable table = (HTable)conn.getTable(tableName)) {
    TEST_UTIL.waitUntilAllRegionsAssigned(tableName);
    LoadIncrementalHFiles loader = new LoadIncrementalHFiles(conf);
    loader.doBulkLoad(loadPath, table);
  }
}

/**
 * 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 testUseExistingToken() throws Exception {
  User user = User.createUserForTesting(TEST_UTIL.getConfiguration(), "testuser2",
      new String[]{"testgroup"});
  Token<AuthenticationTokenIdentifier> token =
      secretManager.generateToken(user.getName());
  assertNotNull(token);
  user.addToken(token);

  // make sure we got a token
  Token<AuthenticationTokenIdentifier> firstToken =
      new AuthenticationTokenSelector().selectToken(token.getService(), user.getTokens());
  assertNotNull(firstToken);
  assertEquals(token, firstToken);

  Connection conn = ConnectionFactory.createConnection(TEST_UTIL.getConfiguration());
  try {
    assertFalse(TokenUtil.addTokenIfMissing(conn, user));
    // make sure we still have the same token
    Token<AuthenticationTokenIdentifier> secondToken =
        new AuthenticationTokenSelector().selectToken(token.getService(), user.getTokens());
    assertEquals(firstToken, secondToken);
  } finally {
    conn.close();
  }
}

/**
 * Puts the specified HRegionInfo into META with replica related columns
 */
public static void fixMetaHoleOnlineAndAddReplicas(Configuration conf,
    HRegionInfo hri, Collection<ServerName> servers, int numReplicas) throws IOException {
  Connection conn = ConnectionFactory.createConnection(conf);
  Table meta = conn.getTable(TableName.META_TABLE_NAME);
  Put put = MetaTableAccessor.makePutFromRegionInfo(hri);
  if (numReplicas > 1) {
    Random r = new Random();
    ServerName[] serversArr = servers.toArray(new ServerName[servers.size()]);
    for (int i = 1; i < numReplicas; i++) {
      ServerName sn = serversArr[r.nextInt(serversArr.length)];
      // the column added here is just to make sure the master is able to
      // see the additional replicas when it is asked to assign. The
      // final value of these columns will be different and will be updated
      // by the actual regionservers that start hosting the respective replicas
      MetaTableAccessor.addLocation(put, sn, sn.getStartcode(), -1, i);
    }
  }
  meta.put(put);
  meta.close();
  conn.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);
}

/**
 * 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();
  }
}

/**
 * Obtain an authentication token for the given user and add it to the
 * user's credentials.
 * @param conn The HBase cluster connection
 * @param user The user for whom to obtain the token
 * @throws IOException If making a remote call to the authentication service fails
 * @throws InterruptedException If executing as the given user is interrupted
 */
public static void obtainAndCacheToken(final Connection conn,
    User user)
    throws IOException, InterruptedException {
  try {
    Token<AuthenticationTokenIdentifier> token = obtainToken(conn, user);

    if (token == null) {
      throw new IOException("No token returned for user " + user.getName());
    }
    if (LOG.isDebugEnabled()) {
      LOG.debug("Obtained token " + token.getKind().toString() + " for user " +
          user.getName());
    }
    user.addToken(token);
  } catch (IOException ioe) {
    throw ioe;
  } catch (InterruptedException ie) {
    throw ie;
  } catch (RuntimeException re) {
    throw re;
  } catch (Exception e) {
    throw new UndeclaredThrowableException(e,
        "Unexpected exception obtaining token for user " + user.getName());
  }
}

/**
 * Grant permissions globally to 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 grantGlobal(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.grant(null, protocol, user, actions);
        }
      }
      return null;
    }
  });
}

private void verifyUserDeniedForIncrementMultipleVersions(final User user, final byte[] row,
    final byte[] q1) throws IOException, InterruptedException {
  user.runAs(new PrivilegedExceptionAction<Void>() {
    @Override
    public Void run() throws Exception {
      try (Connection connection = ConnectionFactory.createConnection(conf)) {
        try (Table t = connection.getTable(TEST_TABLE.getTableName())) {
          Increment inc = new Increment(row);
          inc.setTimeRange(0, 127);
          inc.addColumn(TEST_FAMILY1, q1, 2L);
          t.increment(inc);
          fail(user.getShortName() + " cannot do the increment.");
        } catch (Exception e) {

        }
      }
      return null;
    }
  });
}

/**
 * Obtain an authentication token, for the specified cluster, on behalf of the current user
 * and add it to the credentials for the given map reduce job.
 *
 * @param job The job that requires the permission.
 * @param conf The configuration to use in connecting to the peer cluster
 * @throws IOException When the authentication token cannot be obtained.
 */
public static void initCredentialsForCluster(Job job, Configuration conf)
    throws IOException {
  UserProvider userProvider = UserProvider.instantiate(job.getConfiguration());
  if (userProvider.isHBaseSecurityEnabled()) {
    try {
      Connection peerConn = ConnectionFactory.createConnection(conf);
      try {
        TokenUtil.addTokenForJob(peerConn, userProvider.getCurrent(), job);
      } finally {
        peerConn.close();
      }
    } catch (InterruptedException e) {
      LOG.info("Interrupted obtaining user authentication token");
      Thread.interrupted();
    }
  }
}

@BeforeClass
public static void setupBeforeClass() throws Exception {
  // setup configuration
  conf = TEST_UTIL.getConfiguration();
  VisibilityTestUtil.enableVisiblityLabels(conf);
  // Not setting any SLG class. This means to use the default behavior.
  conf.set("hbase.superuser", "admin");
  TEST_UTIL.startMiniCluster(1);
  SUPERUSER = User.createUserForTesting(conf, "admin", new String[] { "supergroup" });
  TESTUSER = User.createUserForTesting(conf, "test", new String[] { });

  // Wait for the labels table to become available
  TEST_UTIL.waitTableEnabled(LABELS_TABLE_NAME.getName(), 50000);

  // Set up for the test
  SUPERUSER.runAs(new PrivilegedExceptionAction<Void>() {
    public Void run() throws Exception {
      try (Connection conn = ConnectionFactory.createConnection(conf)) {
        VisibilityClient.addLabels(conn, new String[] { SECRET, CONFIDENTIAL });
        VisibilityClient.setAuths(conn, new String[] { CONFIDENTIAL }, TESTUSER.getShortName());
      } catch (Throwable t) {
        throw new IOException(t);
      }
      return null;
    }
  });
}

static void createPresplitTable(TableName tableName, SplitAlgorithm splitAlgo,
        String[] columnFamilies, Configuration conf)
throws IOException, InterruptedException {
  final int splitCount = conf.getInt("split.count", 0);
  Preconditions.checkArgument(splitCount > 1, "Split count must be > 1");

  Preconditions.checkArgument(columnFamilies.length > 0,
      "Must specify at least one column family. ");
  LOG.debug("Creating table " + tableName + " with " + columnFamilies.length
      + " column families.  Presplitting to " + splitCount + " regions");

  HTableDescriptor desc = new HTableDescriptor(tableName);
  for (String cf : columnFamilies) {
    desc.addFamily(new HColumnDescriptor(Bytes.toBytes(cf)));
  }
  try (Connection connection = ConnectionFactory.createConnection(conf)) {
    Admin admin = connection.getAdmin();
    try {
      Preconditions.checkArgument(!admin.tableExists(tableName),
        "Table already exists: " + tableName);
      admin.createTable(desc, splitAlgo.split(splitCount));
    } finally {
      admin.close();
    }
    LOG.debug("Table created!  Waiting for regions to show online in META...");
    if (!conf.getBoolean("split.verify", true)) {
      // NOTE: createTable is synchronous on the table, but not on the regions
      int onlineRegions = 0;
      while (onlineRegions < splitCount) {
        onlineRegions = MetaTableAccessor.getRegionCount(connection, tableName);
        LOG.debug(onlineRegions + " of " + splitCount + " regions online...");
        if (onlineRegions < splitCount) {
          Thread.sleep(10 * 1000); // sleep
        }
      }
    }
    LOG.debug("Finished creating table with " + splitCount + " regions");
  }
}

@Override
protected Connection initSourceConnect() throws IOException, ServiceException {

    // 目前只有zklist转成serverlist和dbname
    Configuration config = HBaseConfiguration.create();
    String address = connection.toString(",");
    config.set(DataStoreProtocol.HBASE_ZK_QUORUM, address);
    config.set("hbase.client.scanner.caching",
            (String) connection.getContext(DataStoreProtocol.HBASE_QUERY_CACHING));
    config.set("hbase.client.scanner.max.result.size",
            (String) connection.getContext(DataStoreProtocol.HBASE_QUERY_MAXRESULTSIZE));
    config.set("zookeeper.recovery.retry", String.valueOf(connection.getRetryTimes()));

    // Failed to replace a bad datanode exception protection configuration
    config.set("dfs.client.block.write.replace-datanode-on-failure.policy", "NEVER");
    config.set("dfs.client.block.write.replace-datanode-on-failure.enable", "true");

    HBaseAdmin.checkHBaseAvailable(config);
    conn = ConnectionFactory.createConnection(config);
    // hbase.client.retries.number = 1 and zookeeper.recovery.retry = 1.
    return conn;
}

/**
 * Tests the case where killing a primary region with unflushed data recovers
 */
@Test (timeout = 120000)
public void testPrimaryRegionKill() throws Exception {
  try (Connection connection = ConnectionFactory.createConnection(HTU.getConfiguration());
      Table table = connection.getTable(htd.getTableName())) {

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

    // wal replication is async, we have to wait until the replication catches up, or we timeout
    verifyNumericRowsWithTimeout(table, fam, 0, 1000, 1, 30000);
    verifyNumericRowsWithTimeout(table, fam, 0, 1000, 2, 30000);

    // we should not have flushed files now, but data in memstores of primary and secondary
    // kill the primary region replica now, and ensure that when it comes back up, we can still
    // read from it the same data from primary and secondaries
    boolean aborted = false;
    for (RegionServerThread rs : HTU.getMiniHBaseCluster().getRegionServerThreads()) {
      for (Region r : rs.getRegionServer().getOnlineRegions(htd.getTableName())) {
        if (r.getRegionInfo().getReplicaId() == 0) {
          LOG.info("Aborting region server hosting primary region replica");
          rs.getRegionServer().abort("for test");
          aborted = true;
        }
      }
    }
    assertTrue(aborted);

    // wal replication is async, we have to wait until the replication catches up, or we timeout
    verifyNumericRowsWithTimeout(table, fam, 0, 1000, 0, 30000);
    verifyNumericRowsWithTimeout(table, fam, 0, 1000, 1, 30000);
    verifyNumericRowsWithTimeout(table, fam, 0, 1000, 2, 30000);
  }

  // restart the region server
  HTU.getMiniHBaseCluster().startRegionServer();
}

private static void setAuths() throws Exception {
  String[] labels = { SECRET, CONFIDENTIAL, PRIVATE, PUBLIC, TOPSECRET };
  try (Connection conn = ConnectionFactory.createConnection(conf)) {
    VisibilityClient.setAuths(conn, labels, User.getCurrent().getShortName());
  } catch (Throwable t) {
    throw new IOException(t);
  }
}

/**
 * Splits the region into two in an atomic operation. Offlines the parent
 * region with the information that it is split into two, and also adds
 * the daughter regions. Does not add the location information to the daughter
 * regions since they are not open yet.
 * @param connection connection we're using
 * @param parent the parent region which is split
 * @param splitA Split daughter region A
 * @param splitB Split daughter region A
 * @param sn the location of the region
 */
public static void splitRegion(final Connection connection,
                               HRegionInfo parent, HRegionInfo splitA, HRegionInfo splitB,
                               ServerName sn, int regionReplication) throws IOException {
  Table meta = getMetaHTable(connection);
  try {
    HRegionInfo copyOfParent = new HRegionInfo(parent);
    copyOfParent.setOffline(true);
    copyOfParent.setSplit(true);

    //Put for parent
    Put putParent = makePutFromRegionInfo(copyOfParent);
    addDaughtersToPut(putParent, splitA, splitB);

    //Puts for daughters
    Put putA = makePutFromRegionInfo(splitA);
    Put putB = makePutFromRegionInfo(splitB);

    addLocation(putA, sn, 1, -1, splitA.getReplicaId()); //new regions, openSeqNum = 1 is fine.
    addLocation(putB, sn, 1, -1, splitB.getReplicaId());

    // Add empty locations for region replicas of daughters so that number of replicas can be
    // cached whenever the primary region is looked up from meta
    for (int i = 1; i < regionReplication; i++) {
      addEmptyLocation(putA, i);
      addEmptyLocation(putB, i);
    }

    byte[] tableRow = Bytes.toBytes(parent.getRegionNameAsString() + HConstants.DELIMITER);
    multiMutate(meta, tableRow, putParent, putA, putB);
  } finally {
    meta.close();
  }
}

@Test (timeout = 240000)
public void testRegionReplicaWithoutMemstoreReplication() throws Exception {
  int regionReplication = 3;
  TableName tableName = TableName.valueOf("testRegionReplicaWithoutMemstoreReplication");
  HTableDescriptor htd = HTU.createTableDescriptor(tableName.toString());
  htd.setRegionReplication(regionReplication);
  htd.setRegionMemstoreReplication(false);
  HTU.getHBaseAdmin().createTable(htd);

  Connection connection = ConnectionFactory.createConnection(HTU.getConfiguration());
  Table table = connection.getTable(tableName);
  try {
    // write data to the primary. The replicas should not receive the data
    final int STEP = 100;
    for (int i = 0; i < 3; ++i) {
      final int startRow = i * STEP;
      final int endRow = (i + 1) * STEP;
      LOG.info("Writing data from " + startRow + " to " + endRow);
      HTU.loadNumericRows(table, HBaseTestingUtility.fam1, startRow, endRow);
      verifyReplication(tableName, regionReplication, startRow, endRow, false);

      // Flush the table, now the data should show up in the replicas
      LOG.info("flushing table");
      HTU.flush(tableName);
      verifyReplication(tableName, regionReplication, 0, endRow, true);
    }
  } finally {
    table.close();
    connection.close();
  }
}

public static void generateHBaseDatasetDoubleOB(Connection conn, Admin admin, TableName tableName, int numberRegions) throws Exception {
  if (admin.tableExists(tableName)) {
    admin.disableTable(tableName);
    admin.deleteTable(tableName);
  }

  HTableDescriptor desc = new HTableDescriptor(tableName);
  desc.addFamily(new HColumnDescriptor(FAMILY_F));

  if (numberRegions > 1) {
    admin.createTable(desc, Arrays.copyOfRange(SPLIT_KEYS, 0, numberRegions-1));
  } else {
    admin.createTable(desc);
  }

  BufferedMutator table = conn.getBufferedMutator(tableName);

  for (double i = 0.5; i <= 100.00; i += 0.75) {
    byte[] bytes = new byte[9];
    PositionedByteRange br = new SimplePositionedMutableByteRange(bytes, 0, 9);
    OrderedBytes.encodeFloat64(br, i, Order.ASCENDING);
    Put p = new Put(bytes);
    p.addColumn(FAMILY_F, COLUMN_C, String.format("value %03f", i).getBytes());
    table.mutate(p);
  }

  table.close();

  admin.flush(tableName);
}

private void warmUpConnectionCache(Connection connection, TableName tn) throws IOException {
  try (RegionLocator locator = connection.getRegionLocator(tn)) {
    LOG.info(
        "Warmed up region location cache for " + tn
            + " got " + locator.getAllRegionLocations().size());
  }
}

private void verfifyUserDeniedForWrite(final User user, final byte[] value) throws Exception {
  verifyDenied(new AccessTestAction() {
    @Override
    public Object run() throws Exception {
      try (Connection connection = ConnectionFactory.createConnection(conf);
          Table t = connection.getTable(TEST_TABLE.getTableName())) {
        Put p;
        p = new Put(TEST_ROW).addColumn(TEST_FAMILY, TEST_Q1, value);
        t.put(p);
      }
      return null;
    }
  }, user);
}

@Test
public void testListLabels() throws Throwable {
  PrivilegedExceptionAction<ListLabelsResponse> action =
      new PrivilegedExceptionAction<ListLabelsResponse>() {
    public ListLabelsResponse run() throws Exception {
      ListLabelsResponse response = null;
      try (Connection conn = ConnectionFactory.createConnection(conf)) {
        response = VisibilityClient.listLabels(conn, null);
      } catch (Throwable e) {
        fail("Should not have thrown exception");
      }
      // The addLabels() in setup added:
      // { SECRET, TOPSECRET, CONFIDENTIAL, PUBLIC, PRIVATE, COPYRIGHT, ACCENT,
      //  UNICODE_VIS_TAG, UC1, UC2 };
      // The previous tests added 2 more labels: ABC, XYZ
      // The 'system' label is excluded.
      List<ByteString> labels = response.getLabelList();
      assertEquals(12, labels.size());
      assertTrue(labels.contains(ByteString.copyFrom(SECRET.getBytes())));
      assertTrue(labels.contains(ByteString.copyFrom(TOPSECRET.getBytes())));
      assertTrue(labels.contains(ByteString.copyFrom(CONFIDENTIAL.getBytes())));
      assertTrue(labels.contains(ByteString.copyFrom("ABC".getBytes())));
      assertTrue(labels.contains(ByteString.copyFrom("XYZ".getBytes())));
      assertFalse(labels.contains(ByteString.copyFrom(SYSTEM_LABEL.getBytes())));
      return null;
    }
  };
  SUPERUSER.runAs(action);
}

public static void generateHBaseDatasetIntOBDesc(Connection conn, Admin admin, TableName tableName, int numberRegions) throws Exception {
  if (admin.tableExists(tableName)) {
    admin.disableTable(tableName);
    admin.deleteTable(tableName);
  }

  HTableDescriptor desc = new HTableDescriptor(tableName);
  desc.addFamily(new HColumnDescriptor(FAMILY_F));

  if (numberRegions > 1) {
    admin.createTable(desc, Arrays.copyOfRange(SPLIT_KEYS, 0, numberRegions-1));
  } else {
    admin.createTable(desc);
  }

  BufferedMutator table = conn.getBufferedMutator(tableName);

  for (int i = -49; i <= 100; i ++) {
    byte[] bytes = new byte[5];
    PositionedByteRange br = new SimplePositionedMutableByteRange(bytes, 0, 5);
    OrderedBytes.encodeInt32(br, i, Order.DESCENDING);
    Put p = new Put(bytes);
    p.addColumn(FAMILY_F, COLUMN_C, String.format("value %d", i).getBytes());
    table.mutate(p);
  }

  table.close();

  admin.flush(tableName);
}

private void runTest(String testName, HTableDescriptor htd, BloomType bloomType,
    boolean preCreateTable, byte[][] tableSplitKeys, byte[][][] hfileRanges) throws Exception {

  for (boolean managed : new boolean[] { true, false }) {
    Path dir = util.getDataTestDirOnTestFS(testName);
    FileSystem fs = util.getTestFileSystem();
    dir = dir.makeQualified(fs);
    Path familyDir = new Path(dir, Bytes.toString(FAMILY));

    int hfileIdx = 0;
    for (byte[][] range : hfileRanges) {
      byte[] from = range[0];
      byte[] to = range[1];
      HFileTestUtil.createHFile(util.getConfiguration(), fs, new Path(familyDir, "hfile_"
          + hfileIdx++), FAMILY, QUALIFIER, from, to, 1000);
    }
    int expectedRows = hfileIdx * 1000;

    if (preCreateTable) {
      util.getHBaseAdmin().createTable(htd, tableSplitKeys);
    }

    final TableName tableName = htd.getTableName();
    if (!util.getHBaseAdmin().tableExists(tableName)) {
      util.getHBaseAdmin().createTable(htd);
    }
    LoadIncrementalHFiles loader = new LoadIncrementalHFiles(util.getConfiguration());

    if (managed) {
      try (HTable table = new HTable(util.getConfiguration(), tableName)) {
        loader.doBulkLoad(dir, table);
        assertEquals(expectedRows, util.countRows(table));
      }
    } else {
      try (Connection conn = ConnectionFactory.createConnection(util.getConfiguration());
          HTable table = (HTable) conn.getTable(tableName)) {
        loader.doBulkLoad(dir, table);
      }
    }

    // verify staging folder has been cleaned up
    Path stagingBasePath = SecureBulkLoadUtil.getBaseStagingDir(util.getConfiguration());
    if (fs.exists(stagingBasePath)) {
      FileStatus[] files = fs.listStatus(stagingBasePath);
      for (FileStatus file : files) {
        assertTrue("Folder=" + file.getPath() + " is not cleaned up.",
            file.getPath().getName() != "DONOTERASE");
      }
    }

    util.deleteTable(tableName);
  }
}