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

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

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 verifyUserDeniedForCheckAndDelete(final User user, final byte[] row,
    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.addColumns(TEST_FAMILY1, TEST_Q1);
          t.checkAndDelete(row, TEST_FAMILY1, TEST_Q1, value, d);
          fail(user.getShortName() + " should not be allowed to do checkAndDelete");
        } catch (Exception e) {
        }
      }
      return null;
    }
  });
}

@Override
protected void initialize(JobContext job) throws IOException {
  Connection connection = ConnectionFactory.createConnection(HBaseConfiguration.create(
      job.getConfiguration()));
  TableName tableName = TableName.valueOf("exampleTable");
  // 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);
}

public static void addLabels() throws Exception {
  PrivilegedExceptionAction<VisibilityLabelsResponse> action =
      new PrivilegedExceptionAction<VisibilityLabelsResponse>() {
    public VisibilityLabelsResponse run() throws Exception {
      String[] labels = { SECRET, TOPSECRET, CONFIDENTIAL, PUBLIC, PRIVATE, COPYRIGHT, ACCENT,
          UNICODE_VIS_TAG, UC1, UC2 };
      try (Connection conn = ConnectionFactory.createConnection(conf)) {
        VisibilityClient.addLabels(conn, labels);
      } catch (Throwable t) {
        throw new IOException(t);
      }
      return null;
    }
  };
  SUPERUSER.runAs(action);
}

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

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

/**
 * Reads user permission assignments stored in the <code>l:</code> column
 * family of the first table row in <code>_acl_</code>.
 *
 * <p>
 * See {@link AccessControlLists class documentation} for the key structure
 * used for storage.
 * </p>
 */
static ListMultimap<String, TablePermission> getPermissions(Configuration conf,
    byte[] entryName) throws IOException {
  if (entryName == null) entryName = ACL_GLOBAL_NAME;

  // for normal user tables, we just read the table row from _acl_
  ListMultimap<String, TablePermission> perms = ArrayListMultimap.create();
  // 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)) {
      Get get = new Get(entryName);
      get.addFamily(ACL_LIST_FAMILY);
      Result row = table.get(get);
      if (!row.isEmpty()) {
        perms = parsePermissions(entryName, row);
      } else {
        LOG.info("No permissions found in " + ACL_TABLE_NAME + " for acl entry "
            + Bytes.toString(entryName));
      }
    }
  }

  return perms;
}

public static void checkGlobalPerms(HBaseTestingUtility testUtil, Permission.Action... actions)
    throws IOException {
  Permission[] perms = new Permission[actions.length];
  for (int i = 0; i < actions.length; i++) {
    perms[i] = new Permission(actions[i]);
  }
  CheckPermissionsRequest.Builder request = CheckPermissionsRequest.newBuilder();
  for (Action a : actions) {
    request.addPermission(AccessControlProtos.Permission.newBuilder()
        .setType(AccessControlProtos.Permission.Type.Global)
        .setGlobalPermission(
            AccessControlProtos.GlobalPermission.newBuilder()
                .addAction(ProtobufUtil.toPermissionAction(a)).build()));
  }
  try(Connection conn = ConnectionFactory.createConnection(testUtil.getConfiguration());
      Table acl = conn.getTable(AccessControlLists.ACL_TABLE_NAME)) {
    BlockingRpcChannel channel = acl.coprocessorService(new byte[0]);
    AccessControlService.BlockingInterface protocol =
      AccessControlService.newBlockingStub(channel);
    try {
      protocol.checkPermissions(null, request.build());
    } catch (ServiceException se) {
      ProtobufUtil.toIOException(se);
    }
  }
}

public static void addLabels() throws Exception {
  PrivilegedExceptionAction<VisibilityLabelsResponse> action =
      new PrivilegedExceptionAction<VisibilityLabelsResponse>() {
    @Override
    public VisibilityLabelsResponse run() throws Exception {
      String[] labels = { SECRET, TOPSECRET, CONFIDENTIAL, PUBLIC, PRIVATE };
      try (Connection conn = ConnectionFactory.createConnection(conf)) {
        VisibilityClient.addLabels(conn, labels);
      } catch (Throwable t) {
        throw new IOException(t);
      }
      return null;
    }
  };
  SUPERUSER.runAs(action);
}

/**
 * Allows subclasses to set the {@link HTable}.
 *
 * Will attempt to reuse the underlying Connection for our own needs, including
 * retreiving an Admin interface to the HBase cluster.
 *
 * @param table  The table to get the data from.
 * @throws IOException 
 * @deprecated Use {@link #initializeTable(Connection, TableName)} instead.
 */
@Deprecated
protected void setHTable(HTable table) throws IOException {
  this.table = table;
  this.connection = table.getConnection();
  try {
    this.regionLocator = table.getRegionLocator();
    this.admin = this.connection.getAdmin();
  } catch (NeedUnmanagedConnectionException exception) {
    LOG.warn("You are using an HTable instance that relies on an HBase-managed Connection. " +
        "This is usually due to directly creating an HTable, which is deprecated. Instead, you " +
        "should create a Connection object and then request a Table instance from it. If you " +
        "don't need the Table instance for your own use, you should instead use the " +
        "TableInputFormatBase.initalizeTable method directly.");
    LOG.info("Creating an additional unmanaged connection because user provided one can't be " +
        "used for administrative actions. We'll close it when we close out the table.");
    LOG.debug("Details about our failure to request an administrative interface.", exception);
    // Do we need a "copy the settings from this Connection" method? are things like the User
    // properly maintained by just looking again at the Configuration?
    this.connection = ConnectionFactory.createConnection(this.connection.getConfiguration());
    this.regionLocator = this.connection.getRegionLocator(table.getName());
    this.admin = this.connection.getAdmin();
  }
}

public static void addLabels() throws Exception {
  PrivilegedExceptionAction<VisibilityLabelsResponse> action =
      new PrivilegedExceptionAction<VisibilityLabelsResponse>() {
    @Override
    public VisibilityLabelsResponse run() throws Exception {
      String[] labels = { TOPSECRET };
      try (Connection conn = ConnectionFactory.createConnection(conf)) {
        VisibilityClient.addLabels(conn, labels);
      } catch (Throwable t) {
        throw new IOException(t);
      }
      return null;
    }
  };
  SUPERUSER.runAs(action);
}

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

/**
 * Grant permissions on a namespace 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 grantOnNamespace(final HBaseTestingUtility util, final String user,
    final String namespace, final Permission.Action... actions) throws Exception {
  SecureTestUtil.updateACLs(util, new Callable<Void>() {
    @Override
    public Void call() throws Exception {
      try (Connection connection = ConnectionFactory.createConnection(util.getConfiguration())) {
        try (Table acl = connection.getTable(AccessControlLists.ACL_TABLE_NAME)) {
          BlockingRpcChannel service = acl.coprocessorService(HConstants.EMPTY_START_ROW);
          AccessControlService.BlockingInterface protocol =
              AccessControlService.newBlockingStub(service);
          ProtobufUtil.grant(null, protocol, user, namespace, actions);
        }
      }
      return null;
    }
  });
}

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

private static void createLabels() throws IOException, InterruptedException {
  PrivilegedExceptionAction<VisibilityLabelsResponse> action =
      new PrivilegedExceptionAction<VisibilityLabelsResponse>() {
    @Override
    public VisibilityLabelsResponse run() throws Exception {
      String[] labels = { SECRET, TOPSECRET, CONFIDENTIAL, PUBLIC, PRIVATE };
      try (Connection conn = ConnectionFactory.createConnection(conf)) {
        VisibilityClient.addLabels(conn, labels);
        LOG.info("Added labels ");
      } catch (Throwable t) {
        LOG.error("Error in adding labels" , t);
        throw new IOException(t);
      }
      return null;
    }
  };
  SUPERUSER.runAs(action);
}

/**
 * Get the cached connection for the current user.
 * If none or timed out, create a new one.
 */
ConnectionInfo getCurrentConnection() throws IOException {
  String userName = getEffectiveUser();
  ConnectionInfo connInfo = connections.get(userName);
  if (connInfo == null || !connInfo.updateAccessTime()) {
    Lock lock = locker.acquireLock(userName);
    try {
      connInfo = connections.get(userName);
      if (connInfo == null) {
        UserGroupInformation ugi = realUser;
        if (!userName.equals(realUserName)) {
          ugi = UserGroupInformation.createProxyUser(userName, realUser);
        }
        User user = userProvider.create(ugi);
        Connection conn = ConnectionFactory.createConnection(conf, user);
        connInfo = new ConnectionInfo(conn, userName);
        connections.put(userName, connInfo);
      }
    } finally {
      lock.unlock();
    }
  }
  return connInfo;
}

@Test (timeout=180000)
public void testCreateTableWithGroupPermissions() throws Exception {
  grantGlobal(TEST_UTIL, TESTGROUP_1_NAME, Action.CREATE);
  try {
    AccessTestAction createAction = new AccessTestAction() {
      @Override
      public Object run() throws Exception {
        HTableDescriptor desc = new HTableDescriptor(TEST_TABLE.getTableName());
        desc.addFamily(new HColumnDescriptor(TEST_FAMILY));
        try (Connection connection =
            ConnectionFactory.createConnection(TEST_UTIL.getConfiguration())) {
          try (Admin admin = connection.getAdmin()) {
            admin.createTable(desc);
          }
        }
        return null;
      }
    };
    verifyAllowed(createAction, TESTGROUP1_USER1);
    verifyDenied(createAction, TESTGROUP2_USER1);
  } finally {
    revokeGlobal(TEST_UTIL, TESTGROUP_1_NAME, Action.CREATE);
  }
}

@Override
protected void setup(Context context) throws IOException, InterruptedException {
  id = Bytes.toBytes("Job: "+context.getJobID() + " Task: " + context.getTaskAttemptID());
  Configuration conf = context.getConfiguration();
  connection = ConnectionFactory.createConnection(conf);
  instantiateHTable();
  this.width = context.getConfiguration().getInt(GENERATOR_WIDTH_KEY, WIDTH_DEFAULT);
  current = new byte[this.width][];
  int wrapMultiplier = context.getConfiguration().getInt(GENERATOR_WRAP_KEY, WRAP_DEFAULT);
  this.wrap = (long)wrapMultiplier * width;
  this.numNodes = context.getConfiguration().getLong(
      GENERATOR_NUM_ROWS_PER_MAP_KEY, (long)WIDTH_DEFAULT * WRAP_DEFAULT);
  if (this.numNodes < this.wrap) {
    this.wrap = this.numNodes;
  }
  this.multipleUnevenColumnFamilies = isMultiUnevenColumnFamilies(context.getConfiguration());
}

@Override
public void setup(Context context) throws IOException {
  conf = context.getConfiguration();
  recordsToWrite = conf.getLong(NUM_TO_WRITE_KEY, NUM_TO_WRITE_DEFAULT);
  String tableName = conf.get(TABLE_NAME_KEY, TABLE_NAME_DEFAULT);
  numBackReferencesPerRow = conf.getInt(NUM_BACKREFS_KEY, NUM_BACKREFS_DEFAULT);
  this.connection = ConnectionFactory.createConnection(conf);
  mutator = connection.getBufferedMutator(
      new BufferedMutatorParams(TableName.valueOf(tableName))
          .writeBufferSize(4 * 1024 * 1024));

  String taskId = conf.get("mapreduce.task.attempt.id");
  Matcher matcher = Pattern.compile(".+_m_(\\d+_\\d+)").matcher(taskId);
  if (!matcher.matches()) {
    throw new RuntimeException("Strange task ID: " + taskId);
  }
  shortTaskId = matcher.group(1);

  rowsWritten = context.getCounter(Counters.ROWS_WRITTEN);
  refsWritten = context.getCounter(Counters.REFERENCES_WRITTEN);
}

public static void checkTablePerms(Configuration conf, TableName table, Permission... perms)
throws IOException {
  CheckPermissionsRequest.Builder request = CheckPermissionsRequest.newBuilder();
  for (Permission p : perms) {
    request.addPermission(ProtobufUtil.toPermission(p));
  }
  try (Connection connection = ConnectionFactory.createConnection(conf)) {
    try (Table acl = connection.getTable(table)) {
      AccessControlService.BlockingInterface protocol =
          AccessControlService.newBlockingStub(acl.coprocessorService(new byte[0]));
      try {
        protocol.checkPermissions(null, request.build());
      } catch (ServiceException se) {
        ProtobufUtil.toIOException(se);
      }
    }
  }
}

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

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

@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 setAuths() throws IOException, InterruptedException {
  PrivilegedExceptionAction<VisibilityLabelsResponse> action =
      new PrivilegedExceptionAction<VisibilityLabelsResponse>() {
    @Override
    public VisibilityLabelsResponse run() throws Exception {
      try (Connection conn = ConnectionFactory.createConnection(conf)) {
        return VisibilityClient.setAuths(conn, new String[] { CONFIDENTIAL,
          PRIVATE, SECRET,
            TOPSECRET }, SUPERUSER.getShortName());
      } catch (Throwable e) {
      }
      return null;
    }
  };
  SUPERUSER.runAs(action);
}

public HBaseAdminTemplate(Configuration configuration) {
    try {
        connection = ConnectionFactory.createConnection(configuration);
        admin = connection.getAdmin();
    } catch (Exception e) {
        throw new HbaseSystemException(e);
    }
}

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

/**
 * Create HBase connection using the given configuration. Verifies whether HBase is available.
 *
 * @param config configuration to use
 * @return HBase connection
 * @throws IOException if an error occurs connecting to HBase
 */
public static Connection createConnection(Configuration config) throws IOException {
  try {
    HBaseAdmin.checkHBaseAvailable(config);
  } catch (ServiceException e) {
    throw new IOException("No HBase connection", e);
  }
  Connection conn = ConnectionFactory.createConnection(config);
  log.info("Created HBase connection");
  return conn;
}

public TestGetRow() throws IOException, InterruptedException {
    Configuration config = HBaseConfiguration.create();

    Connection connection = ConnectionFactory.createConnection(config);
    familyName = config.get("hbase.client.tablestore.family");

    TableName tableName = TableName.valueOf(config.get("hbase.client.tablestore.table"));
    if (!connection.getAdmin().tableExists(tableName)) {
        HTableDescriptor descriptor = new HTableDescriptor(tableName);
        connection.getAdmin().createTable(descriptor);
        TimeUnit.SECONDS.sleep(1);
    }
    table = connection.getTable(tableName);
}

private void init() throws IOException {
  conf = HBaseConfiguration.create();
  HRegionServer.loadWinterConf(conf, confFile);
  conn = ConnectionFactory.createConnection(conf);
  hBaseAdmin = conn.getAdmin();
  relation = new IndexTableRelation(tableName, indexType);
  for (String col : indexColumnNames) {
    relation.addIndexColumn(familyName, Bytes.toBytes(col));
    relation.addColumn(familyName, Bytes.toBytes(col));
  }
  for (int i = 0; i < dataColumnNumber; i++) {
    relation.addColumn(familyName, Bytes.toBytes(i));
  }
  admin = new IndexTableAdmin(conf, conn, relation);
}

protected HTableDescriptor[] getTables(final Configuration configuration) throws IOException {
  HTableDescriptor[] htbls = null;
  try (Connection connection = ConnectionFactory.createConnection(configuration)) {
    try (Admin admin = connection.getAdmin()) {
      htbls = admin.listTables();
    }
  }
  return htbls;
}

public HFileTest() throws IOException {
  conf = HBaseConfiguration.create();
  HRegionServer.loadWinterConf(conf, null);
  conn = ConnectionFactory.createConnection(conf);
  desc = conn.getTable(tableName).getTableDescriptor();
  relation = IndexTableRelation.getIndexTableRelation(desc);
}

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

public DITBScanBase(TableName tableName, IndexType indexType, Configuration conf)
    throws IOException {
  this.tableName = tableName;
  this.indexType = indexType;
  this.conf = conf;
  conn = ConnectionFactory.createConnection(conf);
}

@Test (timeout=180000)
public void testCreateWithCorrectOwner() throws Exception {
  // Create a test user
  final User testUser = User.createUserForTesting(TEST_UTIL.getConfiguration(), "TestUser",
    new String[0]);
  // Grant the test user the ability to create tables
  SecureTestUtil.grantGlobal(TEST_UTIL, testUser.getShortName(), Action.CREATE);
  verifyAllowed(new AccessTestAction() {
    @Override
    public Object run() throws Exception {
      HTableDescriptor desc = new HTableDescriptor(TEST_TABLE.getTableName());
      desc.addFamily(new HColumnDescriptor(TEST_FAMILY));
      try (Connection connection =
          ConnectionFactory.createConnection(TEST_UTIL.getConfiguration(), testUser)) {
        try (Admin admin = connection.getAdmin()) {
          createTable(TEST_UTIL, admin, desc);
        }
      }
      return null;
    }
  }, testUser);
  TEST_UTIL.waitTableAvailable(TEST_TABLE.getTableName());
  // Verify that owner permissions have been granted to the test user on the
  // table just created
  List<TablePermission> perms =
    AccessControlLists.getTablePermissions(conf, TEST_TABLE.getTableName())
     .get(testUser.getShortName());
  assertNotNull(perms);
  assertFalse(perms.isEmpty());
  // Should be RWXCA
  assertTrue(perms.get(0).implies(Permission.Action.READ));
  assertTrue(perms.get(0).implies(Permission.Action.WRITE));
  assertTrue(perms.get(0).implies(Permission.Action.EXEC));
  assertTrue(perms.get(0).implies(Permission.Action.CREATE));
  assertTrue(perms.get(0).implies(Permission.Action.ADMIN));
}

public PerfNormalInserter(String loadDataDir, int processId, int threadNum, String statFilePath,
    ConcurrentLinkedQueue<String> reportQueue, AbstractWorkload workload) throws IOException {
  super(workload.getHBaseConfiguration(), givenTableName, loadDataDir, processId, threadNum,
      statFilePath, reportQueue, workload);
  conn = ConnectionFactory.createConnection(conf);
  indexTableAdmin = new IndexTableAdmin(conf, conn, relation);
  conn = ConnectionFactory.createConnection(conf);
}

public PerfMDInserter(String loadDataDir, int processId, int threadNum, String statFilePath,
    ConcurrentLinkedQueue<String> reportQueue, AbstractWorkload workload) throws IOException {
  super(workload.getHBaseConfiguration(), givenTableName, loadDataDir, processId, threadNum,
      statFilePath, reportQueue, workload);
  conn = ConnectionFactory.createConnection(conf);
  admin = conn.getAdmin();
}

public DITBNormalInserter(Configuration conf, TableName tableName, String loadDataDir,
    int processId, int threadNum, String statFilePath, ConcurrentLinkedQueue<String> reportQueue,
    IndexType indexType, AbstractWorkload workload) throws IOException {
  super(conf, tableName, loadDataDir, processId, threadNum, statFilePath, reportQueue, workload);
  conn = ConnectionFactory.createConnection(conf);
  IndexTableRelation relation = workload.getTableRelation(tableName, indexType);
  indexTableAdmin = new IndexTableAdmin(conf, conn, relation);
  this.indexType = indexType;
  this.statFilePath = statFilePath;
}

public DITBMDInserter(Configuration conf, TableName tableName, String loadDataDir, int processId,
    int threadNum, String statFilePath, ConcurrentLinkedQueue<String> reportQueue,
    IndexType indexType, AbstractWorkload workload) throws IOException {
  super(conf, tableName, loadDataDir, processId, threadNum, statFilePath, reportQueue, workload);
  conn = ConnectionFactory.createConnection(conf);
  admin = conn.getAdmin();
  mdAdmin = new MDHBaseAdmin(conf, tableName, workload.getMDBucketThreshold(),
      workload.getMDDimensions());
}