Java 类org.apache.hadoop.hbase.regionserver.Region 实例源码

@Override
public DeleteTracker postInstantiateDeleteTracker(
    ObserverContext<RegionCoprocessorEnvironment> ctx, DeleteTracker delTracker)
    throws IOException {
  // Nothing to do if we are not filtering by visibility
  if (!authorizationEnabled) {
    return delTracker;
  }
  Region region = ctx.getEnvironment().getRegion();
  TableName table = region.getRegionInfo().getTable();
  if (table.isSystemTable()) {
    return delTracker;
  }
  // We are creating a new type of delete tracker here which is able to track
  // the timestamps and also the
  // visibility tags per cell. The covering cells are determined not only
  // based on the delete type and ts
  // but also on the visibility expression matching.
  return new VisibilityScanDeleteTracker();
}

@Test
// HBASE-3516: Test CP Class loading from local file system
public void testClassLoadingFromLocalFS() throws Exception {
  File jarFile = buildCoprocessorJar(cpName3);

  // create a table that references the jar
  HTableDescriptor htd = new HTableDescriptor(TableName.valueOf(cpName3));
  htd.addFamily(new HColumnDescriptor("test"));
  htd.setValue("COPROCESSOR$1", getLocalPath(jarFile) + "|" + cpName3 + "|" +
    Coprocessor.PRIORITY_USER);
  Admin admin = TEST_UTIL.getHBaseAdmin();
  admin.createTable(htd);
  waitForTable(htd.getTableName());

  // verify that the coprocessor was loaded
  boolean found = false;
  MiniHBaseCluster hbase = TEST_UTIL.getHBaseCluster();
  for (Region region: hbase.getRegionServer(0).getOnlineRegionsLocalContext()) {
    if (region.getRegionInfo().getRegionNameAsString().startsWith(cpName3)) {
      found = (region.getCoprocessorHost().findCoprocessor(cpName3) != null);
    }
  }
  assertTrue("Class " + cpName3 + " was missing on a region", found);
}

void initialize(RegionCoprocessorEnvironment e) throws IOException {
  final Region region = e.getRegion();
  Configuration conf = e.getConfiguration();
  Map<byte[], ListMultimap<String,TablePermission>> tables =
      AccessControlLists.loadAll(region);
  // For each table, write out the table's permissions to the respective
  // znode for that table.
  for (Map.Entry<byte[], ListMultimap<String,TablePermission>> t:
    tables.entrySet()) {
    byte[] entry = t.getKey();
    ListMultimap<String,TablePermission> perms = t.getValue();
    byte[] serialized = AccessControlLists.writePermissionsAsBytes(perms, conf);
    this.authManager.getZKPermissionWatcher().writeToZookeeper(entry, serialized);
  }
  initialized = true;
}

@Override
public void preOpen(ObserverContext<RegionCoprocessorEnvironment> e)
    throws IOException {
  RegionCoprocessorEnvironment env = e.getEnvironment();
  final Region region = env.getRegion();
  if (region == null) {
    LOG.error("NULL region from RegionCoprocessorEnvironment in preOpen()");
  } else {
    HRegionInfo regionInfo = region.getRegionInfo();
    if (regionInfo.getTable().isSystemTable()) {
      checkSystemOrSuperUser();
    } else {
      requirePermission("preOpen", Action.ADMIN);
    }
  }
}

/**
 * Tests that logs are deleted
 * @throws IOException
 * @throws org.apache.hadoop.hbase.regionserver.wal.FailedLogCloseException
 */
@Test
public void testLogRolling() throws Exception {
  this.tableName = getName();
    // TODO: Why does this write data take for ever?
    startAndWriteData();
  final WAL log = server.getWAL(null);
  LOG.info("after writing there are " + DefaultWALProvider.getNumRolledLogFiles(log) +
      " log files");

    // flush all regions
    for (Region r: server.getOnlineRegionsLocalContext()) {
      r.flush(true);
    }

    // Now roll the log
    log.rollWriter();

  int count = DefaultWALProvider.getNumRolledLogFiles(log);
  LOG.info("after flushing all regions and rolling logs there are " + count + " log files");
    assertTrue(("actual count: " + count), count <= 2);
}

/**
 * Check the quota for the current (rpc-context) user. Returns the OperationQuota used to get the
 * available quota and to report the data/usage of the operation.
 * @param region the region where the operation will be performed
 * @param numWrites number of writes to perform
 * @param numReads number of short-reads to perform
 * @param numScans number of scan to perform
 * @return the OperationQuota
 * @throws ThrottlingException if the operation cannot be executed due to quota exceeded.
 */
private OperationQuota checkQuota(final Region region, final int numWrites, final int numReads,
    final int numScans) throws IOException, ThrottlingException {
  User user = RpcServer.getRequestUser();
  UserGroupInformation ugi;
  if (user != null) {
    ugi = user.getUGI();
  } else {
    ugi = User.getCurrent().getUGI();
  }
  TableName table = region.getTableDesc().getTableName();

  OperationQuota quota = getQuota(ugi, table);
  try {
    quota.checkQuota(numWrites, numReads, numScans);
  } catch (ThrottlingException e) {
    LOG.debug("Throttling exception for user=" + ugi.getUserName() + " table=" + table
        + " numWrites=" + numWrites + " numReads=" + numReads + " numScans=" + numScans + ": "
        + e.getMessage());
    throw e;
  }
  return quota;
}

Region reopenRegion(final Region closedRegion, Class<?> ... implClasses)
    throws IOException {
  //HRegionInfo info = new HRegionInfo(tableName, null, null, false);
  Region r = HRegion.openHRegion(closedRegion, null);

  // this following piece is a hack. currently a coprocessorHost
  // is secretly loaded at OpenRegionHandler. we don't really
  // start a region server here, so just manually create cphost
  // and set it to region.
  Configuration conf = TEST_UTIL.getConfiguration();
  RegionCoprocessorHost host = new RegionCoprocessorHost(r, null, conf);
  ((HRegion)r).setCoprocessorHost(host);

  for (Class<?> implClass : implClasses) {
    host.load(implClass, Coprocessor.PRIORITY_USER, conf);
  }
  // we need to manually call pre- and postOpen here since the
  // above load() is not the real case for CP loading. A CP is
  // expected to be loaded by default from 1) configuration; or 2)
  // HTableDescriptor. If it's loaded after HRegion initialized,
  // the pre- and postOpen() won't be triggered automatically.
  // Here we have to call pre and postOpen explicitly.
  host.preOpen();
  host.postOpen();
  return r;
}

Region initHRegion(byte[] tableName, String callingMethod, Configuration conf,
    byte[]... families) throws IOException {
  HTableDescriptor htd = new HTableDescriptor(TableName.valueOf(tableName));
  for (byte[] family : families) {
    htd.addFamily(new HColumnDescriptor(family));
  }
  HRegionInfo info = new HRegionInfo(htd.getTableName(), null, null, false);
  Path path = new Path(DIR + callingMethod);
  HRegion r = HRegion.createHRegion(info, path, conf, htd);
  // this following piece is a hack. currently a coprocessorHost
  // is secretly loaded at OpenRegionHandler. we don't really
  // start a region server here, so just manually create cphost
  // and set it to region.
  RegionCoprocessorHost host = new RegionCoprocessorHost(r, null, conf);
  r.setCoprocessorHost(host);
  return r;
}

@Test (timeout = 240000)
public void testReplayCallable() throws Exception {
  // tests replaying the edits to a secondary region replica using the Callable directly
  openRegion(HTU, rs0, hriSecondary);
  ClusterConnection connection =
      (ClusterConnection) ConnectionFactory.createConnection(HTU.getConfiguration());

  //load some data to primary
  HTU.loadNumericRows(table, f, 0, 1000);

  Assert.assertEquals(1000, entries.size());
  // replay the edits to the secondary using replay callable
  replicateUsingCallable(connection, entries);

  Region region = rs0.getFromOnlineRegions(hriSecondary.getEncodedName());
  HTU.verifyNumericRows(region, f, 0, 1000);

  HTU.deleteNumericRows(table, f, 0, 1000);
  closeRegion(HTU, rs0, hriSecondary);
  connection.close();
}

@Test
public void testRegionObserverScanTimeStacking() throws Exception {
  byte[] ROW = Bytes.toBytes("testRow");
  byte[] TABLE = Bytes.toBytes(getClass().getName());
  byte[] A = Bytes.toBytes("A");
  byte[][] FAMILIES = new byte[][] { A };

  Configuration conf = HBaseConfiguration.create();
  Region region = initHRegion(TABLE, getClass().getName(), conf, FAMILIES);
  RegionCoprocessorHost h = region.getCoprocessorHost();
  h.load(NoDataFromScan.class, Coprocessor.PRIORITY_HIGHEST, conf);
  h.load(EmptyRegionObsever.class, Coprocessor.PRIORITY_USER, conf);

  Put put = new Put(ROW);
  put.add(A, A, A);
  region.put(put);

  Get get = new Get(ROW);
  Result r = region.get(get);
  assertNull(
    "Got an unexpected number of rows - no data should be returned with the NoDataFromScan coprocessor. Found: "
        + r, r.listCells());
}

@Override
public ServerName getServerHoldingRegion(final TableName tn, byte[] regionName)
throws IOException {
  // Assume there is only one master thread which is the active master.
  // If there are multiple master threads, the backup master threads
  // should hold some regions. Please refer to #countServedRegions
  // to see how we find out all regions.
  HMaster master = getMaster();
  Region region = master.getOnlineRegion(regionName);
  if (region != null) {
    return master.getServerName();
  }
  int index = getServerWith(regionName);
  if (index < 0) {
    return null;
  }
  return getRegionServer(index).getServerName();
}

@Test (timeout=300000)
public void testWALRollWriting() throws Exception {
  setUpforLogRolling();
  String className = this.getClass().getName();
  StringBuilder v = new StringBuilder(className);
  while (v.length() < 1000) {
    v.append(className);
  }
  byte[] value = Bytes.toBytes(v.toString());
  HRegionServer regionServer = startAndWriteData(TableName.valueOf("TestLogRolling"), value);
  LOG.info("after writing there are "
      + DefaultWALProvider.getNumRolledLogFiles(regionServer.getWAL(null)) + " log files");

  // flush all regions
  for (Region r : regionServer.getOnlineRegionsLocalContext()) {
    r.flush(true);
  }
  admin.rollWALWriter(regionServer.getServerName());
  int count = DefaultWALProvider.getNumRolledLogFiles(regionServer.getWAL(null));
  LOG.info("after flushing all regions and rolling logs there are " +
      count + " log files");
  assertTrue(("actual count: " + count), count <= 2);
}

@BeforeClass
public static void setUpBeforeClass() throws Exception {
  cluster = TEST_UTIL.startMiniCluster(1, ServerNum);
  table = TEST_UTIL.createTable(tableName, FAMILY, HBaseTestingUtility.KEYS_FOR_HBA_CREATE_TABLE);
  TEST_UTIL.waitTableAvailable(tableName, 1000);
  TEST_UTIL.loadTable(table, FAMILY);

  for (int i = 0; i < ServerNum; i++) {
    HRegionServer server = cluster.getRegionServer(i);
    for (Region region : server.getOnlineRegions(tableName)) {
      region.flush(true);
    }
  }

  finder.setConf(TEST_UTIL.getConfiguration());
  finder.setServices(cluster.getMaster());
  finder.setClusterStatus(cluster.getMaster().getClusterStatus());
}

@Test
public void testInternalGetTopBlockLocation() throws Exception {
  for (int i = 0; i < ServerNum; i++) {
    HRegionServer server = cluster.getRegionServer(i);
    for (Region region : server.getOnlineRegions(tableName)) {
      // get region's hdfs block distribution by region and RegionLocationFinder, 
      // they should have same result
      HDFSBlocksDistribution blocksDistribution1 = region.getHDFSBlocksDistribution();
      HDFSBlocksDistribution blocksDistribution2 = finder.getBlockDistribution(region
          .getRegionInfo());
      assertEquals(blocksDistribution1.getUniqueBlocksTotalWeight(),
        blocksDistribution2.getUniqueBlocksTotalWeight());
      if (blocksDistribution1.getUniqueBlocksTotalWeight() != 0) {
        assertEquals(blocksDistribution1.getTopHosts().get(0), blocksDistribution2.getTopHosts()
            .get(0));
      }
    }
  }
}

@Test
public void testGetTopBlockLocations() throws Exception {
  for (int i = 0; i < ServerNum; i++) {
    HRegionServer server = cluster.getRegionServer(i);
    for (Region region : server.getOnlineRegions(tableName)) {
      List<ServerName> servers = finder.getTopBlockLocations(region.getRegionInfo());
      // test table may have empty region
      if (region.getHDFSBlocksDistribution().getUniqueBlocksTotalWeight() == 0) {
        continue;
      }
      List<String> topHosts = region.getHDFSBlocksDistribution().getTopHosts();
      // rs and datanode may have different host in local machine test
      if (!topHosts.contains(server.getServerName().getHostname())) {
        continue;
      }
      for (int j = 0; j < ServerNum; j++) {
        ServerName serverName = cluster.getRegionServer(j).getServerName();
        assertTrue(servers.contains(serverName));
      }
    }
  }
}

Region initHRegion (TableName tableName, String callingMethod,
    Configuration conf, Class<?> [] implClasses, byte [][] families)
    throws IOException {
  HTableDescriptor htd = new HTableDescriptor(tableName);
  for(byte [] family : families) {
    htd.addFamily(new HColumnDescriptor(family));
  }
  HRegionInfo info = new HRegionInfo(tableName, null, null, false);
  Path path = new Path(DIR + callingMethod);
  HRegion r = HRegion.createHRegion(info, path, conf, htd);

  // this following piece is a hack.
  RegionCoprocessorHost host = new RegionCoprocessorHost(r, null, conf);
  r.setCoprocessorHost(host);

  for (Class<?> implClass : implClasses) {
    host.load(implClass, Coprocessor.PRIORITY_USER, conf);
    Coprocessor c = host.findCoprocessor(implClass.getName());
    assertNotNull(c);
  }

  // Here we have to call pre and postOpen explicitly.
  host.preOpen();
  host.postOpen();
  return r;
}

@Override
public RegionScanner preScannerOpen(ObserverContext<RegionCoprocessorEnvironment> e, Scan scan,
    RegionScanner s) throws IOException {
  if (!initialized) {
    throw new VisibilityControllerNotReadyException("VisibilityController not yet initialized!");
  }
  // Nothing to do if authorization is not enabled
  if (!authorizationEnabled) {
    return s;
  }
  Region region = e.getEnvironment().getRegion();
  Authorizations authorizations = null;
  try {
    authorizations = scan.getAuthorizations();
  } catch (DeserializationException de) {
    throw new IOException(de);
  }
  if (authorizations == null) {
    // No Authorizations present for this scan/Get!
    // In case of system tables other than "labels" just scan with out visibility check and
    // filtering. Checking visibility labels for META and NAMESPACE table is not needed.
    TableName table = region.getRegionInfo().getTable();
    if (table.isSystemTable() && !table.equals(LABELS_TABLE_NAME)) {
      return s;
    }
  }

  Filter visibilityLabelFilter = VisibilityUtils.createVisibilityLabelFilter(region,
      authorizations);
  if (visibilityLabelFilter != null) {
    Filter filter = scan.getFilter();
    if (filter != null) {
      scan.setFilter(new FilterList(filter, visibilityLabelFilter));
    } else {
      scan.setFilter(visibilityLabelFilter);
    }
  }
  return s;
}

public static Filter createVisibilityLabelFilter(Region region, Authorizations authorizations)
    throws IOException {
  Map<ByteRange, Integer> cfVsMaxVersions = new HashMap<ByteRange, Integer>();
  for (HColumnDescriptor hcd : region.getTableDesc().getFamilies()) {
    cfVsMaxVersions.put(new SimpleMutableByteRange(hcd.getName()), hcd.getMaxVersions());
  }
  VisibilityLabelService vls = VisibilityLabelServiceManager.getInstance()
      .getVisibilityLabelService();
  Filter visibilityLabelFilter = new VisibilityLabelFilter(
      vls.getVisibilityExpEvaluator(authorizations), cfVsMaxVersions);
  return visibilityLabelFilter;
}

protected void addSystemLabel(Region region, Map<String, Integer> labels,
    Map<String, List<Integer>> userAuths) throws IOException {
  if (!labels.containsKey(SYSTEM_LABEL)) {
    Put p = new Put(Bytes.toBytes(SYSTEM_LABEL_ORDINAL));
    p.addImmutable(LABELS_TABLE_FAMILY, LABEL_QUALIFIER, Bytes.toBytes(SYSTEM_LABEL));
    region.put(p);
    labels.put(SYSTEM_LABEL, SYSTEM_LABEL_ORDINAL);
  }
}

@Test
// HBASE-6308: Test CP classloader is the CoprocessorClassLoader
public void testPrivateClassLoader() throws Exception {
  File jarFile = buildCoprocessorJar(cpName4);

  // create a table that references the jar
  HTableDescriptor htd = new HTableDescriptor(TableName.valueOf(cpName4));
  htd.addFamily(new HColumnDescriptor("test"));
  htd.setValue("COPROCESSOR$1", getLocalPath(jarFile) + "|" + cpName4 + "|" +
    Coprocessor.PRIORITY_USER);
  Admin admin = TEST_UTIL.getHBaseAdmin();
  admin.createTable(htd);
  waitForTable(htd.getTableName());

  // verify that the coprocessor was loaded correctly
  boolean found = false;
  MiniHBaseCluster hbase = TEST_UTIL.getHBaseCluster();
  for (Region region: hbase.getRegionServer(0).getOnlineRegionsLocalContext()) {
    if (region.getRegionInfo().getRegionNameAsString().startsWith(cpName4)) {
      Coprocessor cp = region.getCoprocessorHost().findCoprocessor(cpName4);
      if (cp != null) {
        found = true;
        assertEquals("Class " + cpName4 + " was not loaded by CoprocessorClassLoader",
          cp.getClass().getClassLoader().getClass(), CoprocessorClassLoader.class);
      }
    }
  }
  assertTrue("Class " + cpName4 + " was missing on a region", found);
}

private TableName getTableName(RegionCoprocessorEnvironment e) {
  Region region = e.getRegion();
  if (region != null) {
    return getTableName(region);
  }
  return null;
}

/**
 * Check the quota for the current (rpc-context) user. Returns the OperationQuota used to get the
 * available quota and to report the data/usage of the operation.
 * @param region the region where the operation will be performed
 * @param type the operation type
 * @return the OperationQuota
 * @throws ThrottlingException if the operation cannot be executed due to quota exceeded.
 */
public OperationQuota checkQuota(final Region region, final OperationQuota.OperationType type)
    throws IOException, ThrottlingException {
  switch (type) {
  case SCAN:
    return checkQuota(region, 0, 0, 1);
  case GET:
    return checkQuota(region, 0, 1, 0);
  case MUTATE:
    return checkQuota(region, 1, 0, 0);
  default:
    throw new RuntimeException("Invalid operation type: " + type);
  }
}

/**
 * Check the quota for the current (rpc-context) user. Returns the OperationQuota used to get the
 * available quota and to report the data/usage of the operation.
 * @param region the region where the operation will be performed
 * @param actions the "multi" actions to perform
 * @return the OperationQuota
 * @throws ThrottlingException if the operation cannot be executed due to quota exceeded.
 */
public OperationQuota checkQuota(final Region region, final List<ClientProtos.Action> actions)
    throws IOException, ThrottlingException {
  int numWrites = 0;
  int numReads = 0;
  for (final ClientProtos.Action action : actions) {
    if (action.hasMutation()) {
      numWrites++;
    } else if (action.hasGet()) {
      numReads++;
    }
  }
  return checkQuota(region, numWrites, numReads, 0);
}

public FlushTableSubprocedure(ProcedureMember member,
    ForeignExceptionDispatcher errorListener, long wakeFrequency, long timeout,
    List<Region> regions, String table,
    FlushTableSubprocedurePool taskManager) {
  super(member, table, errorListener, wakeFrequency, timeout);
  this.table = table;
  this.regions = regions;
  this.taskManager = taskManager;
}

private List<Path> findStorefilePaths(TableName tableName) throws Exception {
  List<Path> paths = new ArrayList<Path>();
  for (Region region:
      TEST_UTIL.getRSForFirstRegionInTable(tableName).getOnlineRegions(htd.getTableName())) {
    for (Store store: region.getStores()) {
      for (StoreFile storefile: store.getStorefiles()) {
        paths.add(storefile.getPath());
      }
    }
  }
  return paths;
}

private void flushRegions() throws ForeignException {
  if (regions.isEmpty()) {
    // No regions on this RS, we are basically done.
    return;
  }

  monitor.rethrowException();

  // assert that the taskManager is empty.
  if (taskManager.hasTasks()) {
    throw new IllegalStateException("Attempting to flush "
        + table + " but we currently have outstanding tasks");
  }

  // Add all hfiles already existing in region.
  for (Region region : regions) {
    // submit one task per region for parallelize by region.
    taskManager.submitTask(new RegionFlushTask(region));
    monitor.rethrowException();
  }

  // wait for everything to complete.
  LOG.debug("Flush region tasks submitted for " + regions.size() + " regions");
  try {
    taskManager.waitForOutstandingTasks();
  } catch (InterruptedException e) {
    throw new ForeignException(getMemberName(), e);
  }
}

/**
 * If in a running state, creates the specified subprocedure to flush table regions.
 *
 * Because this gets the local list of regions to flush and not the set the master had,
 * there is a possibility of a race where regions may be missed.
 *
 * @param table
 * @return Subprocedure to submit to the ProcedureMemeber.
 */
public Subprocedure buildSubprocedure(String table) {

  // don't run the subprocedure if the parent is stop(ping)
  if (rss.isStopping() || rss.isStopped()) {
    throw new IllegalStateException("Can't start flush region subprocedure on RS: "
        + rss.getServerName() + ", because stopping/stopped!");
  }

  // check to see if this server is hosting any regions for the table
  List<Region> involvedRegions;
  try {
    involvedRegions = getRegionsToFlush(table);
  } catch (IOException e1) {
    throw new IllegalStateException("Failed to figure out if there is region to flush.", e1);
  }

  // We need to run the subprocedure even if we have no relevant regions.  The coordinator
  // expects participation in the procedure and without sending message the master procedure
  // will hang and fail.

  LOG.debug("Launching subprocedure to flush regions for " + table);
  ForeignExceptionDispatcher exnDispatcher = new ForeignExceptionDispatcher(table);
  Configuration conf = rss.getConfiguration();
  long timeoutMillis = conf.getLong(FLUSH_TIMEOUT_MILLIS_KEY,
      FLUSH_TIMEOUT_MILLIS_DEFAULT);
  long wakeMillis = conf.getLong(FLUSH_REQUEST_WAKE_MILLIS_KEY,
      FLUSH_REQUEST_WAKE_MILLIS_DEFAULT);

  FlushTableSubprocedurePool taskManager =
      new FlushTableSubprocedurePool(rss.getServerName().toString(), conf, rss);
  return new FlushTableSubprocedure(member, exnDispatcher, wakeMillis,
    timeoutMillis, involvedRegions, table, taskManager);
}

private void doPuts(Region region) throws IOException{
  LoadTestKVGenerator dataGenerator = new LoadTestKVGenerator(MIN_VALUE_SIZE, MAX_VALUE_SIZE);
   for (int i = 0; i < NUM_ROWS; ++i) {
    byte[] key = LoadTestKVGenerator.md5PrefixedKey(i).getBytes();
    for (int j = 0; j < NUM_COLS_PER_ROW; ++j) {
      Put put = new Put(key);
      put.setDurability(Durability.ASYNC_WAL);
      byte[] col = Bytes.toBytes(String.valueOf(j));
      byte[] value = dataGenerator.generateRandomSizeValue(key, col);
      if (includeTags) {
        Tag[] tag = new Tag[1];
        tag[0] = new Tag((byte) 1, "Visibility");
        KeyValue kv = new KeyValue(key, CF_BYTES, col, HConstants.LATEST_TIMESTAMP, value, tag);
        put.add(kv);
      } else {
        put.add(CF_BYTES, col, value);
      }
      if(VERBOSE){
        KeyValue kvPut = new KeyValue(key, CF_BYTES, col, value);
        System.err.println(Strings.padFront(i+"", ' ', 4)+" "+kvPut);
      }
      region.put(put);
    }
    if (i % NUM_ROWS_PER_FLUSH == 0) {
      region.flush(true);
    }
  }
}

private void writeTestData(Region region) throws IOException {
  for (int i = 0; i < NUM_ROWS; ++i) {
    Put put = new Put(Bytes.toBytes("row" + i));
    for (int j = 0; j < NUM_COLS_PER_ROW; ++j) {
      for (long ts = 1; ts < NUM_TIMESTAMPS_PER_COL; ++ts) {
        put.add(CF_BYTES, Bytes.toBytes("col" + j), ts,
            Bytes.toBytes("value" + i + "_" + j + "_" + ts));
      }
    }
    region.put(put);
    if ((i + 1) % ROWS_PER_HFILE == 0) {
      region.flush(true);
    }
  }
}

private void flushSnapshot() throws ForeignException {
  if (regions.isEmpty()) {
    // No regions on this RS, we are basically done.
    return;
  }

  monitor.rethrowException();

  // assert that the taskManager is empty.
  if (taskManager.hasTasks()) {
    throw new IllegalStateException("Attempting to take snapshot "
        + ClientSnapshotDescriptionUtils.toString(snapshot)
        + " but we currently have outstanding tasks");
  }

  // Add all hfiles already existing in region.
  for (Region region : regions) {
    // submit one task per region for parallelize by region.
    taskManager.submitTask(new RegionSnapshotTask(region));
    monitor.rethrowException();
  }

  // wait for everything to complete.
  LOG.debug("Flush Snapshot Tasks submitted for " + regions.size() + " regions");
  try {
    taskManager.waitForOutstandingTasks();
  } catch (InterruptedException e) {
    LOG.error("got interrupted exception for " + getMemberName());
    throw new ForeignException(getMemberName(), e);
  }
}

@Test (timeout = 240000)
public void testReplayCallableWithRegionMove() throws Exception {
  // tests replaying the edits to a secondary region replica using the Callable directly while
  // the region is moved to another location.It tests handling of RME.
  openRegion(HTU, rs0, hriSecondary);
  ClusterConnection connection =
      (ClusterConnection) ConnectionFactory.createConnection(HTU.getConfiguration());
  //load some data to primary
  HTU.loadNumericRows(table, f, 0, 1000);

  Assert.assertEquals(1000, entries.size());
  // replay the edits to the secondary using replay callable
  replicateUsingCallable(connection, entries);

  Region region = rs0.getFromOnlineRegions(hriSecondary.getEncodedName());
  HTU.verifyNumericRows(region, f, 0, 1000);

  HTU.loadNumericRows(table, f, 1000, 2000); // load some more data to primary

  // move the secondary region from RS0 to RS1
  closeRegion(HTU, rs0, hriSecondary);
  openRegion(HTU, rs1, hriSecondary);

  // replicate the new data
  replicateUsingCallable(connection, entries);

  region = rs1.getFromOnlineRegions(hriSecondary.getEncodedName());
  // verify the new data. old data may or may not be there
  HTU.verifyNumericRows(region, f, 1000, 2000);

  HTU.deleteNumericRows(table, f, 0, 2000);
  closeRegion(HTU, rs1, hriSecondary);
  connection.close();
}

private Region [] split(final Region r, final byte [] splitRow) throws IOException {
  Region[] regions = new Region[2];

  SplitTransaction st = new SplitTransactionFactory(TEST_UTIL.getConfiguration())
    .create(r, splitRow);
  int i = 0;

  if (!st.prepare()) {
    // test fails.
    assertTrue(false);
  }
  try {
    Server mockServer = Mockito.mock(Server.class);
    when(mockServer.getConfiguration()).thenReturn(TEST_UTIL.getConfiguration());
    PairOfSameType<Region> daughters = st.execute(mockServer, null);
    for (Region each_daughter: daughters) {
      regions[i] = each_daughter;
      i++;
    }
  } catch (IOException ioe) {
    LOG.info("Split transaction of " + r.getRegionInfo().getRegionNameAsString() +
        " failed:" + ioe.getMessage());
    assertTrue(false);
  } catch (RuntimeException e) {
    LOG.info("Failed rollback of failed split of " +
        r.getRegionInfo().getRegionNameAsString() + e.getMessage());
  }

  assertTrue(i == 2);
  return regions;
}

/** Test region server observer */
@Test
public void testPassiveRegionServerOperations() throws Exception {
  // preStopRegionServer
  verifyAllowed(new AccessTestAction() {
    @Override
    public Object run() throws Exception {
      ACCESS_CONTROLLER.preStopRegionServer(ObserverContext.createAndPrepare(RSCP_ENV, null));
      return null;
    }
  }, SUPERUSER, USER_ADMIN, USER_RW, USER_RO, USER_OWNER, USER_CREATE, USER_QUAL, USER_NONE);

  // preMerge
  verifyAllowed(new AccessTestAction() {
    @Override
    public Object run() throws Exception {
      HTableDescriptor htd = new HTableDescriptor(TEST_TABLE.getTableName());
      Region region_a = mock(Region.class);
      when(region_a.getTableDesc()).thenReturn(htd);
      Region region_b = mock(Region.class);
      when(region_b.getTableDesc()).thenReturn(htd);
      ACCESS_CONTROLLER.preMerge(ObserverContext.createAndPrepare(RSCP_ENV, null), region_a,
        region_b);
      return null;
    }
  }, SUPERUSER, USER_ADMIN, USER_RW, USER_RO, USER_OWNER, USER_CREATE, USER_QUAL, USER_NONE);

  // preRollWALWriterRequest
  verifyAllowed(new AccessTestAction() {
    @Override
    public Object run() throws Exception {
      ACCESS_CONTROLLER.preRollWALWriterRequest(ObserverContext.createAndPrepare(RSCP_ENV,
        null));
      return null;
    }
  }, SUPERUSER, USER_ADMIN, USER_RW, USER_RO, USER_OWNER, USER_CREATE, USER_QUAL, USER_NONE);

}

/**
 * Create a new hfile in the passed region
 * @param region region to operate on
 * @param columnFamily family for which to add data
 * @throws IOException
 */
private void createHFileInRegion(Region region, byte[] columnFamily) throws IOException {
  // put one row in the region
  Put p = new Put(Bytes.toBytes("row"));
  p.add(columnFamily, Bytes.toBytes("Qual"), Bytes.toBytes("v1"));
  region.put(p);
  // flush the region to make a store file
  region.flush(true);
}

/**
 * Call flushCache on all regions on all participating regionservers.
 * @throws IOException
 */
public void flushcache() throws IOException {
  for (JVMClusterUtil.RegionServerThread t:
      this.hbaseCluster.getRegionServers()) {
    for(Region r: t.getRegionServer().getOnlineRegionsLocalContext()) {
      r.flush(true);
    }
  }
}

/**
 * Call flushCache on all regions of the specified table.
 * @throws IOException
 */
public void flushcache(TableName tableName) throws IOException {
  for (JVMClusterUtil.RegionServerThread t:
      this.hbaseCluster.getRegionServers()) {
    for(Region r: t.getRegionServer().getOnlineRegionsLocalContext()) {
      if(r.getTableDesc().getTableName().equals(tableName)) {
        r.flush(true);
      }
    }
  }
}

public static void waitForTableToBeOnline(final HBaseTestingUtility util,
                                          final TableName tableName)
    throws IOException, InterruptedException {
  HRegionServer rs = util.getRSForFirstRegionInTable(tableName);
  List<Region> onlineRegions = rs.getOnlineRegions(tableName);
  for (Region region : onlineRegions) {
    region.waitForFlushesAndCompactions();
  }
  // Wait up to 60 seconds for a table to be available.
  util.waitFor(60000, util.predicateTableAvailable(tableName));
}

/**
 * Call flushCache on all regions of the specified table.
 * @throws IOException
 */
public void compact(TableName tableName, boolean major) throws IOException {
  for (JVMClusterUtil.RegionServerThread t:
      this.hbaseCluster.getRegionServers()) {
    for(Region r: t.getRegionServer().getOnlineRegionsLocalContext()) {
      if(r.getTableDesc().getTableName().equals(tableName)) {
        r.compact(major);
      }
    }
  }
}

public List<HRegion> getRegions(TableName tableName) {
  List<HRegion> ret = new ArrayList<HRegion>();
  for (JVMClusterUtil.RegionServerThread rst : getRegionServerThreads()) {
    HRegionServer hrs = rst.getRegionServer();
    for (Region region : hrs.getOnlineRegionsLocalContext()) {
      if (region.getTableDesc().getTableName().equals(tableName)) {
        ret.add((HRegion)region);
      }
    }
  }
  return ret;
}

/**
 * Get the location of the specified region
 * @param regionName Name of the region in bytes
 * @return Index into List of {@link MiniHBaseCluster#getRegionServerThreads()}
 * of HRS carrying hbase:meta. Returns -1 if none found.
 */
public int getServerWith(byte[] regionName) {
  int index = -1;
  int count = 0;
  for (JVMClusterUtil.RegionServerThread rst: getRegionServerThreads()) {
    HRegionServer hrs = rst.getRegionServer();
    Region region = hrs.getOnlineRegion(regionName);
    if (region != null) {
      index = count;
      break;
    }
    count++;
  }
  return index;
}