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

@Override
public HealthCheck.Result getHealth() {
    try {
        HBaseAdmin.checkHBaseAvailable(configuration);

        return HealthCheck.Result.builder()
                .healthy()
                .withMessage("HBase running on:")
                .withDetail("quorum", quorum)
                .withDetail("clientPort", clientPort)
                .withDetail("znodeParent", znodeParent)
                .build();
    } catch (Exception e) {
        return HealthCheck.Result.builder()
                .unhealthy(e)
                .build();
    }
}

/**
 * Create a hbaseBtable in case the table is not there Else prints table already exist conf is the
 * configuration file object which needs to be passed, hbase-site.xml
 * @method creatTable
 * @inputParameters: tablename as String, ColumnFamalies as String Array[]
 * @return NA as is a voild method 
 **/
public static void creatTable(String myHbaseBtableName, String[] cfamlies) throws Exception {
  HBaseAdmin hbaseBadmin = new HBaseAdmin(hbaseBconf);
  if (hbaseBadmin.tableExists(myHbaseBtableName)) {
    System.out.println("Ohh.. this table is already there");
  } else {
    @SuppressWarnings("deprecation")
    HTableDescriptor hbaseBtableDesc = new HTableDescriptor(myHbaseBtableName);
    for (int i = 0; i < cfamlies.length; i++) {
      hbaseBtableDesc.addFamily(new HColumnDescriptor(cfamlies[i]));
    }
    hbaseBadmin.createTable(hbaseBtableDesc);// creating a table 
    System.out.println("create table " + myHbaseBtableName + " Done.");
    // Table is created and printed with the name.
  }
}

@Test(timeout = 300000)
public void testDisableTableAndRestart() throws Exception {
  final TableName tableName = TableName.valueOf("testDisableTableAndRestart");
  final MiniHBaseCluster cluster = TEST_UTIL.getHBaseCluster();
  final HBaseAdmin admin = TEST_UTIL.getHBaseAdmin();
  final HTableDescriptor desc = new HTableDescriptor(tableName);
  desc.addFamily(new HColumnDescriptor(FAMILYNAME));
  admin.createTable(desc);
  admin.disableTable(tableName);
  TEST_UTIL.waitTableDisabled(tableName.getName());

  TEST_UTIL.getHBaseCluster().shutdown();
  TEST_UTIL.getHBaseCluster().waitUntilShutDown();

  TEST_UTIL.restartHBaseCluster(2);

  admin.enableTable(tableName);
  TEST_UTIL.waitTableEnabled(tableName);
}

public static void createTable(HBaseTestingUtility testUtil, HBaseAdmin admin,
  HTableDescriptor htd, byte [][] splitKeys)
throws Exception {
  // NOTE: We need a latch because admin is not sync,
  // so the postOp coprocessor method may be called after the admin operation returned.
  MasterSyncObserver observer = (MasterSyncObserver)testUtil.getHBaseCluster().getMaster()
    .getMasterCoprocessorHost().findCoprocessor(MasterSyncObserver.class.getName());
  observer.tableCreationLatch = new CountDownLatch(1);
  if (splitKeys != null) {
    admin.createTable(htd, splitKeys);
  } else {
    admin.createTable(htd);
  }
  observer.tableCreationLatch.await();
  observer.tableCreationLatch = null;
  testUtil.waitUntilAllRegionsAssigned(htd.getTableName());
}

public static void deleteTable(HBaseTestingUtility testUtil, HBaseAdmin admin,
  TableName tableName)
throws Exception {
  // NOTE: We need a latch because admin is not sync,
  // so the postOp coprocessor method may be called after the admin operation returned.
  MasterSyncObserver observer = (MasterSyncObserver)testUtil.getHBaseCluster().getMaster()
    .getMasterCoprocessorHost().findCoprocessor(MasterSyncObserver.class.getName());
  observer.tableDeletionLatch = new CountDownLatch(1);
  try {
    admin.disableTable(tableName);
  } catch (Exception e) {
    LOG.debug("Table: " + tableName + " already disabled, so just deleting it.");
  }
  admin.deleteTable(tableName);
  observer.tableDeletionLatch.await();
  observer.tableDeletionLatch = null;
}

protected void prepareForLoadTest() throws IOException {
  LOG.info("Starting load test: dataBlockEncoding=" + dataBlockEncoding +
      ", isMultiPut=" + isMultiPut);
  numKeys = numKeys();
  Admin admin = new HBaseAdmin(conf);
  while (admin.getClusterStatus().getServers().size() < NUM_RS) {
    LOG.info("Sleeping until " + NUM_RS + " RSs are online");
    Threads.sleepWithoutInterrupt(1000);
  }
  admin.close();

  HTableDescriptor htd = new HTableDescriptor(TABLE);
  HColumnDescriptor hcd = new HColumnDescriptor(CF)
    .setCompressionType(compression)
    .setDataBlockEncoding(dataBlockEncoding);
  createPreSplitLoadTestTable(htd, hcd);

  LoadTestDataGenerator dataGen = new MultiThreadedAction.DefaultDataGenerator(CF);
  writerThreads = prepareWriterThreads(dataGen, conf, TABLE);
  readerThreads = prepareReaderThreads(dataGen, conf, TABLE, 100);
}

/**
 * Get region info from local cluster.
 */
Map<ServerName, List<String>> getDeployedHRIs(final HBaseAdmin admin) throws IOException {
  ClusterStatus status = admin.getClusterStatus();
  Collection<ServerName> regionServers = status.getServers();
  Map<ServerName, List<String>> mm =
      new HashMap<ServerName, List<String>>();
  for (ServerName hsi : regionServers) {
    AdminProtos.AdminService.BlockingInterface server = ((HConnection) connection).getAdmin(hsi);

    // list all online regions from this region server
    List<HRegionInfo> regions = ProtobufUtil.getOnlineRegions(server);
    List<String> regionNames = new ArrayList<String>();
    for (HRegionInfo hri : regions) {
      regionNames.add(hri.getRegionNameAsString());
    }
    mm.put(hsi, regionNames);
  }
  return mm;
}

private Store prepareData() throws IOException {
  HBaseAdmin admin = TEST_UTIL.getHBaseAdmin();
  if (admin.tableExists(tableName)) {
    admin.disableTable(tableName);
    admin.deleteTable(tableName);
  }
  HTable table = TEST_UTIL.createTable(tableName, family);
  Random rand = new Random();
  for (int i = 0; i < 10; i++) {
    for (int j = 0; j < 10; j++) {
      byte[] value = new byte[128 * 1024];
      rand.nextBytes(value);
      table.put(new Put(Bytes.toBytes(i * 10 + j)).add(family, qualifier, value));
    }
    admin.flush(tableName);
  }
  return getStoreWithName(tableName);
}

private void compactAndWait() throws IOException, InterruptedException {
  LOG.debug("Compacting table " + tableName);
  HRegionServer rs = TEST_UTIL.getMiniHBaseCluster().getRegionServer(0);
  HBaseAdmin admin = TEST_UTIL.getHBaseAdmin();
  admin.majorCompact(tableName);

  // Waiting for the compaction to start, at least .5s.
  final long maxWaitime = System.currentTimeMillis() + 500;
  boolean cont;
  do {
    cont = rs.compactSplitThread.getCompactionQueueSize() == 0;
    Threads.sleep(1);
  } while (cont && System.currentTimeMillis() < maxWaitime);

  while (rs.compactSplitThread.getCompactionQueueSize() > 0) {
    Threads.sleep(1);
  }
  LOG.debug("Compaction queue size reached 0, continuing");
}

@Override
public int runTestFromCommandLine() throws Exception {
  IntegrationTestingUtility.setUseDistributedCluster(getConf());
  int numPresplits = getConf().getInt("loadmapper.numPresplits", 5);
  // create HTableDescriptor for specified table
  HTableDescriptor htd = new HTableDescriptor(getTablename());
  htd.addFamily(new HColumnDescriptor(TEST_FAMILY));

  Admin admin = new HBaseAdmin(getConf());
  try {
    admin.createTable(htd, Bytes.toBytes(0L), Bytes.toBytes(-1L), numPresplits);
  } finally {
    admin.close();
  }
  doLoad(getConf(), htd);
  doVerify(getConf(), htd);
  getTestingUtil(getConf()).deleteTable(htd.getName());
  return 0;
}

public static void create() throws Exception {
    HBaseAdmin admin = new HBaseAdmin(cfg);
    if (admin.tableExists(tableName)) {
        System.out.println("[info]table has created!");
    } else {
        try {
            TableName table = TableName.valueOf(tableName);
            HTableDescriptor tableDescriptor = new HTableDescriptor(table);
            tableDescriptor.addFamily(new HColumnDescriptor(familyName));

            admin.createTable(tableDescriptor);
        } catch (TableExistsException e) {
            System.out.println("[warning] table exists!");
        }
    }
    System.out.println("[info]create table success!");
}

public static void create() throws Exception
{
    HBaseAdmin admin = new HBaseAdmin(cfg);
    if (admin.tableExists(tableName)) {
        System.out.println("[info]table has created!");
    } else {
        TableName table = TableName.valueOf(tableName);
        HTableDescriptor tableDescriptor = new HTableDescriptor(table);
        tableDescriptor.addFamily(new HColumnDescriptor(familyName));
        try{
            admin.createTable(tableDescriptor);
        }catch(TableExistsException e)
        {
            System.out.println("[warning] table exists!");
        }
    }
    System.out.println("[info]create table success!");
}

public static boolean delete(String tablename) throws IOException{

        HBaseAdmin admin=new HBaseAdmin(cfg);
        if(admin.tableExists(tablename)){
                try
                {
                        admin.disableTable(tablename);
                        admin.deleteTable(tablename);
                }catch(Exception ex){
                        ex.printStackTrace();
                        return false;
                }

        }
        return true;
}

@BeforeMethod
public void setUp() throws Exception {
    HBaseAdmin admin = testutil.getHBaseAdmin();

    if (!admin.tableExists(DEFAULT_TIMESTAMP_STORAGE_TABLE_NAME)) {
        HTableDescriptor desc = new HTableDescriptor(TABLE_NAME);
        HColumnDescriptor datafam = new HColumnDescriptor(DEFAULT_TIMESTAMP_STORAGE_CF_NAME);
        datafam.setMaxVersions(Integer.MAX_VALUE);
        desc.addFamily(datafam);

        admin.createTable(desc);
    }

    if (admin.isTableDisabled(DEFAULT_TIMESTAMP_STORAGE_TABLE_NAME)) {
        admin.enableTable(DEFAULT_TIMESTAMP_STORAGE_TABLE_NAME);
    }
    HTableDescriptor[] tables = admin.listTables();
    for (HTableDescriptor t : tables) {
        LOG.info(t.getNameAsString());
    }
}

private static void createTable(HBaseAdmin admin, String tableName, byte[][] families, byte[][] splitKeys,
                                int maxVersions)
        throws IOException {

    LOG.info("About to create Table named {} with {} splits", tableName, splitKeys.length);

    if (admin.tableExists(tableName)) {
        LOG.error("Table {} already exists. Table creation cancelled", tableName);
        return;
    }

    HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));

    for (byte[] family : families) {
        HColumnDescriptor colDescriptor = new HColumnDescriptor(family);
        colDescriptor.setMaxVersions(maxVersions);
        HBaseShims.addFamilyToHTableDescriptor(tableDescriptor, colDescriptor);
        LOG.info("\tAdding Family {}", colDescriptor);
    }

    admin.createTable(tableDescriptor, splitKeys);

    LOG.info("Table {} created. Regions: {}", tableName, admin.getTableRegions(Bytes.toBytes(tableName)).size());

}

@AfterMethod(alwaysRun = true, timeOut = 60_000)
public void cleanup() throws Exception {
    LOG.info("Cleanup");
    HBaseAdmin admin = hBaseUtils.getHBaseAdmin();
    deleteTable(admin, TableName.valueOf(DEFAULT_TIMESTAMP_STORAGE_TABLE_NAME));
    hBaseUtils.createTable(Bytes.toBytes(DEFAULT_TIMESTAMP_STORAGE_TABLE_NAME),
                           new byte[][]{DEFAULT_TIMESTAMP_STORAGE_CF_NAME.getBytes()},
                           Integer.MAX_VALUE);
    tso1.stopAndWait();
    TestUtils.waitForSocketNotListening("localhost", TSO1_PORT, 100);
    tso2.stopAndWait();
    TestUtils.waitForSocketNotListening("localhost", TSO2_PORT, 100);

    zkClient.delete().forPath(TSO_LEASE_PATH);
    LOG.info("ZKPath {} deleted", TSO_LEASE_PATH);
    zkClient.delete().forPath(CURRENT_TSO_PATH);
    LOG.info("ZKPaths {} deleted", CURRENT_TSO_PATH);

    zkClient.close();
}

@BeforeClass
public void setupTestSnapshotFilter() throws Exception {
    TSOServerConfig tsoConfig = new TSOServerConfig();
    tsoConfig.setPort(5678);
    tsoConfig.setConflictMapSize(1);
    injector = Guice.createInjector(new TSOForSnapshotFilterTestModule(tsoConfig));
    hbaseConf = injector.getInstance(Configuration.class);
    hbaseConf.setBoolean("omid.server.side.filter", true);
    hbaseConf.setInt("hbase.master.info.port", 16011);
    HBaseCommitTableConfig hBaseCommitTableConfig = injector.getInstance(HBaseCommitTableConfig.class);
    HBaseTimestampStorageConfig hBaseTimestampStorageConfig = injector.getInstance(HBaseTimestampStorageConfig.class);

    setupHBase();
    aggregationClient = new AggregationClient(hbaseConf);
    admin = new HBaseAdmin(hbaseConf);
    createRequiredHBaseTables(hBaseTimestampStorageConfig, hBaseCommitTableConfig);
    setupTSO();

    commitTable = injector.getInstance(CommitTable.class);
}

@BeforeClass
public void setupTestCompation() throws Exception {
    TSOServerConfig tsoConfig = new TSOServerConfig();
    tsoConfig.setPort(1234);
    tsoConfig.setConflictMapSize(1);
    injector = Guice.createInjector(new TSOForHBaseCompactorTestModule(tsoConfig));
    hbaseConf = injector.getInstance(Configuration.class);
    HBaseCommitTableConfig hBaseCommitTableConfig = injector.getInstance(HBaseCommitTableConfig.class);
    HBaseTimestampStorageConfig hBaseTimestampStorageConfig = injector.getInstance(HBaseTimestampStorageConfig.class);

    // settings required for #testDuplicateDeletes()
    hbaseConf.setInt("hbase.hstore.compaction.min", 2);
    hbaseConf.setInt("hbase.hstore.compaction.max", 2);

    setupHBase();
    aggregationClient = new AggregationClient(hbaseConf);
    admin = new HBaseAdmin(hbaseConf);
    createRequiredHBaseTables(hBaseTimestampStorageConfig, hBaseCommitTableConfig);
    setupTSO();

    commitTable = injector.getInstance(CommitTable.class);
}

/**
 * Creates the table with the column families
 *
 * @param tableName
 * @param columnFamilies
 * @return
 */
public static void createTable(String tableName, String... columnFamilies) {
    HBaseTestingUtility testingUtility = getUtility();
    if (!status.get()) {
        throw new RuntimeException("The mini cluster hasn't started yet. " +
          " Call HBaseTestUtil#startMiniCluster() before creating a table");
    }
    final TableName name = TableName.valueOf(tableName);
    try (HBaseAdmin hBaseAdmin = testingUtility.getHBaseAdmin()) {
        final HTableDescriptor hTableDescriptor = new HTableDescriptor(name);
        for (String family : columnFamilies) {
            final HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(Bytes.toBytes(family));
            hTableDescriptor.addFamily(hColumnDescriptor);
        }

        hBaseAdmin.createTable(hTableDescriptor);
        testingUtility.waitUntilAllRegionsAssigned(name);
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
}

public SecondaryIndexTable(TransactionServiceClient transactionServiceClient, HTableInterface hTable,
                           byte[] secondaryIndex) {
  secondaryIndexTableName = TableName.valueOf(hTable.getName().getNameAsString() + ".idx");
  HTable secondaryIndexHTable = null;
  try (HBaseAdmin hBaseAdmin = new HBaseAdmin(hTable.getConfiguration())) {
    if (!hBaseAdmin.tableExists(secondaryIndexTableName)) {
      hBaseAdmin.createTable(new HTableDescriptor(secondaryIndexTableName));
    }
    secondaryIndexHTable = new HTable(hTable.getConfiguration(), secondaryIndexTableName);
  } catch (Exception e) {
    Throwables.propagate(e);
  }

  this.secondaryIndex = secondaryIndex;
  this.transactionAwareHTable = new TransactionAwareHTable(hTable);
  this.secondaryIndexTable = new TransactionAwareHTable(secondaryIndexHTable);
  this.transactionContext = new TransactionContext(transactionServiceClient, transactionAwareHTable,
                                                   secondaryIndexTable);
}

@Override
public void initialize(Configuration conf) throws IOException {
  this.conf = conf;
  this.hBaseAdmin = new HBaseAdmin(conf);
  this.connection = HConnectionManager.createConnection(conf);

  final TableName stateTable = TableName.valueOf(conf.get(TxConstants.TransactionPruning.PRUNE_STATE_TABLE,
                                                          TxConstants.TransactionPruning.DEFAULT_PRUNE_STATE_TABLE));
  LOG.info("Initializing plugin with state table {}:{}", stateTable.getNamespaceAsString(),
           stateTable.getNameAsString());
  createPruneTable(stateTable);
  this.dataJanitorState = new DataJanitorState(new DataJanitorState.TableSupplier() {
    @Override
    public HTableInterface get() throws IOException {
      return connection.getTable(stateTable);
    }
  });
}

public SecondaryIndexTable(TransactionServiceClient transactionServiceClient, HTableInterface hTable,
                           byte[] secondaryIndex) {
  secondaryIndexTableName = TableName.valueOf(hTable.getName().getNameAsString() + ".idx");
  HTable secondaryIndexHTable = null;
  try (HBaseAdmin hBaseAdmin = new HBaseAdmin(hTable.getConfiguration())) {
    if (!hBaseAdmin.tableExists(secondaryIndexTableName)) {
      hBaseAdmin.createTable(new HTableDescriptor(secondaryIndexTableName));
    }
    secondaryIndexHTable = new HTable(hTable.getConfiguration(), secondaryIndexTableName);
  } catch (Exception e) {
    Throwables.propagate(e);
  }

  this.secondaryIndex = secondaryIndex;
  this.transactionAwareHTable = new TransactionAwareHTable(hTable);
  this.secondaryIndexTable = new TransactionAwareHTable(secondaryIndexHTable);
  this.transactionContext = new TransactionContext(transactionServiceClient, transactionAwareHTable,
                                                   secondaryIndexTable);
}

@Override
public void initialize(Configuration conf) throws IOException {
  this.conf = conf;
  this.hBaseAdmin = new HBaseAdmin(conf);
  this.connection = HConnectionManager.createConnection(conf);

  final TableName stateTable = TableName.valueOf(conf.get(TxConstants.TransactionPruning.PRUNE_STATE_TABLE,
                                                          TxConstants.TransactionPruning.DEFAULT_PRUNE_STATE_TABLE));
  LOG.info("Initializing plugin with state table {}:{}", stateTable.getNamespaceAsString(),
           stateTable.getNameAsString());
  createPruneTable(stateTable);
  this.dataJanitorState = new DataJanitorState(new DataJanitorState.TableSupplier() {
    @Override
    public HTableInterface get() throws IOException {
      return connection.getTable(stateTable);
    }
  });
}

/**
 * delete table in preparation for next test
 *
 * @param tablename
 * @throws IOException
 */
void deleteTable(TableName tablename) throws IOException {
  HBaseAdmin admin = new HBaseAdmin(conf);
  admin.getConnection().clearRegionCache();
  if (admin.isTableEnabled(tablename)) {
    admin.disableTableAsync(tablename);
  }
  long totalWait = 0;
  long maxWait = 30*1000;
  long sleepTime = 250;
  while (!admin.isTableDisabled(tablename)) {
    try {
      Thread.sleep(sleepTime);
      totalWait += sleepTime;
      if (totalWait >= maxWait) {
        fail("Waited too long for table to be disabled + " + tablename);
      }
    } catch (InterruptedException e) {
      e.printStackTrace();
      fail("Interrupted when trying to disable table " + tablename);
    }
  }
  admin.deleteTable(tablename);
}

private HTableDescriptor ensureTableExists(String tableName, String initialCFName, int ttlInSeconds) throws BackendException {
    HBaseAdmin adm = null;

    HTableDescriptor desc;

    try { // Create our table, if necessary
        adm = getAdminInterface();
        /*
         * Some HBase versions/impls respond badly to attempts to create a
         * table without at least one CF. See #661. Creating a CF along with
         * the table avoids HBase carping.
         */
        if (adm.tableExists(tableName)) {
            desc = adm.getTableDescriptor(tableName.getBytes());
        } else {
            desc = createTable(tableName, initialCFName, ttlInSeconds, adm);
        }
    } catch (IOException e) {
        throw new TemporaryBackendException(e);
    } finally {
        IOUtils.closeQuietly(adm);
    }

    return desc;
}

public static void main(String[] argsParam) throws Exception {
    setLoggingThreshold("ERROR");

    SnapshotArgs args;
    try {
        args = new SnapshotArgs(argsParam);
    } catch (IllegalArgumentException e) {
        System.out.println(e.getMessage());
        System.out.println();
        System.out.println(usage());
        System.exit(1);
        throw e;
    }

    HBaseAdmin admin = HBaseClient.getAdmin(args);

    Snapshot app = new Snapshot(admin, args);
    app.run();
}

@Override
public void deleteTable(String tableName) {
    Util.checkEmptyString(tableName);

    try {
        HBaseAdmin hbaseAdmin = hbaseDataSource.getHBaseAdmin();
        // delete table if table exist.
        if (hbaseAdmin.tableExists(tableName)) {
            // disable table before delete it.
            if (!hbaseAdmin.isTableDisabled(tableName)) {
                hbaseAdmin.disableTable(tableName);
            }
            hbaseAdmin.deleteTable(tableName);
        }
    } catch (Exception e) {
        log.error(e);
        throw new SimpleHBaseException(e);
    }
}

@Override
public void perform() throws Exception {
  HBaseTestingUtility util = context.getHBaseIntegrationTestingUtility();
  HBaseAdmin admin = util.getHBaseAdmin();

  LOG.info("Performing action: Merge random adjacent regions of table " + tableName);
  List<HRegionInfo> regions = admin.getTableRegions(tableNameBytes);
  if (regions == null || regions.size() < 2) {
    LOG.info("Table " + tableName + " doesn't have enough regions to merge");
    return;
  }

  int i = RandomUtils.nextInt(regions.size() - 1);
  HRegionInfo a = regions.get(i++);
  HRegionInfo b = regions.get(i);
  LOG.debug("Merging " + a.getRegionNameAsString() + " and " + b.getRegionNameAsString());
  try {
    admin.mergeRegions(a.getEncodedNameAsBytes(), b.getEncodedNameAsBytes(), false);
  } catch (Exception ex) {
    LOG.warn("Merge failed, might be caused by other chaos: " + ex.getMessage());
  }
  if (sleepTime > 0) {
    Thread.sleep(sleepTime);
  }
}

public static void main(String[] argsParam) throws Exception {
    setLoggingThreshold("ERROR");

    SnapshotArgs args;
    try {
        args = new SnapshotArgs(argsParam);
    } catch (IllegalArgumentException e) {
        System.out.println(e.getMessage());
        System.out.println();
        System.out.println(usage());
        System.exit(1);
        throw e;
    }

    HBaseAdmin admin = HBaseClient.getAdmin(args);

    Snapshot app = new Snapshot(admin, args);
    app.run();
}

@BeforeClass
public static void setUpBeforeClass() throws Exception {
  conf = TEST_UTIL.getConfiguration();
  TEST_UTIL.startMiniCluster();
  REST_TEST_UTIL.startServletContainer(conf);
  context = JAXBContext.newInstance(
          CellModel.class,
          CellSetModel.class,
          RowModel.class);
  marshaller = context.createMarshaller();
  unmarshaller = context.createUnmarshaller();
  client = new Client(new Cluster().add("localhost", REST_TEST_UTIL.getServletPort()));
  HBaseAdmin admin = TEST_UTIL.getHBaseAdmin();
  if (admin.tableExists(TABLE)) {
    return;
  }
  HTableDescriptor htd = new HTableDescriptor(TABLE);
  htd.addFamily(new HColumnDescriptor(CFA));
  htd.addFamily(new HColumnDescriptor(CFB));
  admin.createTable(htd);
}

private void parseTables(HBaseAdmin admin, String[] tables, Set<String> tableSet) throws IOException {
    for (String table : tables) {
        final String tableName;
        if (table.contains(ENTRY_DELIMITER)) {
            String[] parts = table.split(ENTRY_DELIMITER);
            tableName = parts[0];
            tableKeepMap.put(tableName, Integer.valueOf(parts[1]));
            tableFlushMap.put(tableName, Boolean.valueOf(parts[2]));
        } else {
            tableName = table;
        }

        for (HTableDescriptor hTableDescriptor : admin.listTables(tableName)) {
            tableSet.add(hTableDescriptor.getNameAsString());
        }
    }
}

@BeforeClass
public static void setUpBeforeClass() throws Exception {
  conf = TEST_UTIL.getConfiguration();
  TEST_UTIL.startMiniCluster(3);
  REST_TEST_UTIL.startServletContainer(conf);
  context = JAXBContext.newInstance(
      CellModel.class,
      CellSetModel.class,
      RowModel.class);
  marshaller = context.createMarshaller();
  unmarshaller = context.createUnmarshaller();
  client = new Client(new Cluster().add("localhost", 
    REST_TEST_UTIL.getServletPort()));
  HBaseAdmin admin = TEST_UTIL.getHBaseAdmin();
  if (admin.tableExists(TABLE)) {
    return;
  }
  HTableDescriptor htd = new HTableDescriptor(TABLE);
  htd.addFamily(new HColumnDescriptor(CFA));
  htd.addFamily(new HColumnDescriptor(CFB));
  admin.createTable(htd);
}

/**
 * Get region info from local cluster.
 */
Map<ServerName, List<String>> getDeployedHRIs(
    final HBaseAdmin admin) throws IOException {
  ClusterStatus status = admin.getClusterStatus();
  Collection<ServerName> regionServers = status.getServers();
  Map<ServerName, List<String>> mm =
      new HashMap<ServerName, List<String>>();
  HConnection connection = admin.getConnection();
  for (ServerName hsi : regionServers) {
    AdminProtos.AdminService.BlockingInterface server = connection.getAdmin(hsi);

    // list all online regions from this region server
    List<HRegionInfo> regions = ProtobufUtil.getOnlineRegions(server);
    List<String> regionNames = new ArrayList<String>();
    for (HRegionInfo hri : regions) {
      regionNames.add(hri.getRegionNameAsString());
    }
    mm.put(hsi, regionNames);
  }
  return mm;
}

private void initIRIndex(HBaseAdmin admin) throws IOException {
  System.out.println("start init IRIndex");
  HTableDescriptor tableDesc = new HTableDescriptor(tableName);

  IndexDescriptor index1 = new IndexDescriptor(Bytes.toBytes("c3"), DataType.DOUBLE);
  IndexDescriptor index2 = new IndexDescriptor(Bytes.toBytes("c4"), DataType.STRING);
  IndexDescriptor index3 = new IndexDescriptor(Bytes.toBytes("c5"), DataType.STRING);

  IndexColumnDescriptor family = new IndexColumnDescriptor("f");
  family.addIndex(index1);
  family.addIndex(index2);
  family.addIndex(index3);

  tableDesc.addFamily(family);
  admin.createTable(tableDesc);
}

@Before
public void setUp() throws Exception {
  LOG.info(String.format("Initializing cluster with %d region servers.",
    REGION_SERVER_COUNT));
  util.initializeCluster(REGION_SERVER_COUNT);
  LOG.info("Cluster initialized");

  HBaseAdmin admin = util.getHBaseAdmin();
  if (admin.tableExists(TABLE_NAME)) {
    LOG.info(String.format("Deleting existing table %s.", TABLE_NAME));
    if (admin.isTableEnabled(TABLE_NAME)) admin.disableTable(TABLE_NAME);
    admin.deleteTable(TABLE_NAME);
    LOG.info(String.format("Existing table %s deleted.", TABLE_NAME));
  }
  LOG.info("Cluster ready");
}

public static void createTable(HBaseAdmin admin, String tableName, String[] families,
        boolean dataBlockEncoding, boolean compression, int remaining) throws IOException {
    HTableDescriptor table = new HTableDescriptor(TableName.valueOf(tableName));
    for (String f : families) {
        HColumnDescriptor colDesc = new HColumnDescriptor(f);
        // Compression
        if (compression) {
            colDesc.setCompressionType(Algorithm.SNAPPY);
        }

        // Alive time
        if (remaining > 0) {
            colDesc.setTimeToLive(remaining);
        }

        // Block encoding
        if (dataBlockEncoding) {
            colDesc.setDataBlockEncoding(DataBlockEncoding.FAST_DIFF);
        }

        table.addFamily(colDesc);
    }
    admin.createTable(table);
}

/**
 * Get region info from local cluster.
 */
Map<ServerName, List<String>> getDeployedHRIs(HBaseAdmin admin)
  throws IOException {
  ClusterStatus status = admin.getMaster().getClusterStatus();
  Collection<ServerName> regionServers = status.getServers();
  Map<ServerName, List<String>> mm =
      new HashMap<ServerName, List<String>>();
  HConnection connection = admin.getConnection();
  for (ServerName hsi : regionServers) {
    HRegionInterface server =
      connection.getHRegionConnection(hsi.getHostname(), hsi.getPort());

    // list all online regions from this region server
    List<HRegionInfo> regions = server.getOnlineRegions();
    List<String> regionNames = new ArrayList<String>();
    for (HRegionInfo hri : regions) {
      regionNames.add(hri.getRegionNameAsString());
    }
    mm.put(hsi, regionNames);
  }
  return mm;
}

public void init(String resourcePath) throws IOException {
  System.out.println("start init");
  Configuration conf = HBaseConfiguration.create();
  if (resourcePath != null) {
    conf.addResource(resourcePath);
  }
  HBaseAdmin admin = new HBaseAdmin(conf);

  if (admin.tableExists(tableName)) {
    admin.disableTable(tableName);
    admin.deleteTable(tableName);
    System.out.println("finish deleting existing table and index tables");
  }

  // for each kind
  initIRIndex(admin);
}

@Test
public void testLoadAndVerify() throws Exception {
  HTableDescriptor htd = new HTableDescriptor(TEST_NAME);
  htd.addFamily(new HColumnDescriptor(TEST_FAMILY));

  HBaseAdmin admin = getTestingUtil().getHBaseAdmin();
  int numPreCreate = 40;
  admin.createTable(htd, Bytes.toBytes(0L), Bytes.toBytes(-1L), numPreCreate);

  doLoad(getConf(), htd);
  doVerify(getConf(), htd);

  // Only disable and drop if we succeeded to verify - otherwise it's useful
  // to leave it around for post-mortem
  deleteTable(admin, htd);
}

/**
 * 连接HBase
 */
public static void open() {
    try {
        config = HBaseConfiguration.create();
        conn = HConnectionManager.createConnection(config);
        admin = new HBaseAdmin(conn);
        hbase_table = conn.getTable(ISAXIndex.TABLE_NAME);
    } catch (IOException e) {
        e.printStackTrace();
    }
}