Java 类org.apache.hadoop.hbase.util.JVMClusterUtil.RegionServerThread 实例源码

@SuppressWarnings("unchecked")
public JVMClusterUtil.RegionServerThread addRegionServer(
    Configuration config, final int index)
throws IOException {
  // Create each regionserver with its own Configuration instance so each has
  // its HConnection instance rather than share (see HBASE_INSTANCES down in
  // the guts of HConnectionManager.

  // Also, create separate CoordinatedStateManager instance per Server.
  // This is special case when we have to have more than 1 CoordinatedStateManager
  // within 1 process.
  CoordinatedStateManager cp = CoordinatedStateManagerFactory.getCoordinatedStateManager(conf);

  JVMClusterUtil.RegionServerThread rst =
      JVMClusterUtil.createRegionServerThread(config, cp, (Class<? extends HRegionServer>) conf
          .getClass(HConstants.REGION_SERVER_IMPL, this.regionServerClass), index);

  this.regionThreads.add(rst);
  return rst;
}

/**
 * Wait for the specified region server to stop
 * Removes this thread from list of running threads.
 * @param rst
 * @return Name of region server that just went down.
 */
public String waitOnRegionServer(JVMClusterUtil.RegionServerThread rst) {
  while (rst.isAlive()) {
    try {
      LOG.info("Waiting on " +
        rst.getRegionServer().toString());
      rst.join();
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
  for (int i=0;i<regionThreads.size();i++) {
    if (regionThreads.get(i) == rst) {
      regionThreads.remove(i);
      break;
    }
  }
  return rst.getName();
}

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

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

/**
 * Starts a region server thread running
 *
 * @throws IOException
 * @return New RegionServerThread
 */
public JVMClusterUtil.RegionServerThread startRegionServer()
    throws IOException {
  final Configuration newConf = HBaseConfiguration.create(conf);
  User rsUser =
      HBaseTestingUtility.getDifferentUser(newConf, ".hfs."+index++);
  JVMClusterUtil.RegionServerThread t =  null;
  try {
    t = hbaseCluster.addRegionServer(
        newConf, hbaseCluster.getRegionServers().size(), rsUser);
    t.start();
    t.waitForServerOnline();
  } catch (InterruptedException ie) {
    throw new IOException("Interrupted adding regionserver to cluster", ie);
  }
  return t;
}

/**
 * Make sure that at least the specified number of region servers
 * are running. We don't count the ones that are currently stopping or are
 * stopped.
 * @param num minimum number of region servers that should be running
 * @return true if we started some servers
 * @throws IOException
 */
public boolean ensureSomeNonStoppedRegionServersAvailable(final int num)
  throws IOException {
  boolean startedServer = ensureSomeRegionServersAvailable(num);

  int nonStoppedServers = 0;
  for (JVMClusterUtil.RegionServerThread rst :
    getMiniHBaseCluster().getRegionServerThreads()) {

    HRegionServer hrs = rst.getRegionServer();
    if (hrs.isStopping() || hrs.isStopped()) {
      LOG.info("A region server is stopped or stopping:"+hrs);
    } else {
      nonStoppedServers++;
    }
  }
  for (int i=nonStoppedServers; i<num; ++i) {
    LOG.info("Started new server=" + getMiniHBaseCluster().startRegionServer());
    startedServer = true;
  }
  return startedServer;
}

void populateDataInTable(int nrows, String fname) throws Exception {
  byte [] family = Bytes.toBytes(fname);

  List<RegionServerThread> rsts = cluster.getLiveRegionServerThreads();
  assertEquals(NUM_RS, rsts.size());

  for (RegionServerThread rst : rsts) {
    HRegionServer hrs = rst.getRegionServer();
    List<HRegionInfo> hris = hrs.getOnlineRegions();
    for (HRegionInfo hri : hris) {
      if (hri.isMetaTable()) {
        continue;
      }
      LOG.debug("adding data to rs = " + rst.getName() +
          " region = "+ hri.getRegionNameAsString());
      HRegion region = hrs.getOnlineRegion(hri.getRegionName());
      assertTrue(region != null);
      putData(region, hri.getStartKey(), nrows, Bytes.toBytes("q"), family);
    }
  }
}

public JVMClusterUtil.RegionServerThread addRegionServer(
    Configuration config, final int index)
throws IOException {
  // Create each regionserver with its own Configuration instance so each has
  // its HConnection instance rather than share (see HBASE_INSTANCES down in
  // the guts of HConnectionManager.

  // Also, create separate CoordinatedStateManager instance per Server.
  // This is special case when we have to have more than 1 CoordinatedStateManager
  // within 1 process.
  CoordinatedStateManager cp = CoordinatedStateManagerFactory.getCoordinatedStateManager(conf);

  JVMClusterUtil.RegionServerThread rst =
    JVMClusterUtil.createRegionServerThread(config, cp,
        this.regionServerClass, index);
  this.regionThreads.add(rst);
  return rst;
}

/**
 * Starts a region server thread running
 *
 * @throws IOException
 * @return New RegionServerThread
 */
public JVMClusterUtil.RegionServerThread startRegionServer()
    throws IOException {
  final Configuration newConf = HBaseConfiguration.create(conf);
  User rsUser =
      HBaseTestingUtility.getDifferentUser(newConf, ".hfs."+index++);
  JVMClusterUtil.RegionServerThread t =  null;
  try {
    t = hbaseCluster.addRegionServer(
        newConf, hbaseCluster.getRegionServers().size(), rsUser);
    t.start();
    t.waitForServerOnline();
  } catch (InterruptedException ie) {
    throw new IOException("Interrupted adding regionserver to cluster", ie);
  }
  return t;
}

void populateDataInTable(int nrows, String fname) throws Exception {
  byte [] family = Bytes.toBytes(fname);

  List<RegionServerThread> rsts = cluster.getLiveRegionServerThreads();
  assertEquals(NUM_RS, rsts.size());

  for (RegionServerThread rst : rsts) {
    HRegionServer hrs = rst.getRegionServer();
    List<HRegionInfo> hris = ProtobufUtil.getOnlineRegions(hrs);
    for (HRegionInfo hri : hris) {
      if (hri.getTable().isSystemTable()) {
        continue;
      }
      LOG.debug("adding data to rs = " + rst.getName() +
          " region = "+ hri.getRegionNameAsString());
      HRegion region = hrs.getOnlineRegion(hri.getRegionName());
      assertTrue(region != null);
      putData(region, hri.getStartKey(), nrows, Bytes.toBytes("q"), family);
    }
  }
}

/**
 * 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();
    HRegion metaRegion =
      hrs.getOnlineRegion(regionName);
    if (metaRegion != null) {
      index = count;
      break;
    }
    count++;
  }
  return index;
}

/**
 * Make sure that at least the specified number of region servers
 * are running. We don't count the ones that are currently stopping or are
 * stopped.
 * @param num minimum number of region servers that should be running
 * @return true if we started some servers
 * @throws IOException
 */
public boolean ensureSomeNonStoppedRegionServersAvailable(final int num)
  throws IOException {
  boolean startedServer = ensureSomeRegionServersAvailable(num);

  int nonStoppedServers = 0;
  for (JVMClusterUtil.RegionServerThread rst :
    getMiniHBaseCluster().getRegionServerThreads()) {

    HRegionServer hrs = rst.getRegionServer();
    if (hrs.isStopping() || hrs.isStopped()) {
      LOG.info("A region server is stopped or stopping:"+hrs);
    } else {
      nonStoppedServers++;
    }
  }
  for (int i=nonStoppedServers; i<num; ++i) {
    LOG.info("Started new server=" + getMiniHBaseCluster().startRegionServer());
    startedServer = true;
  }
  return startedServer;
}

void populateDataInTable(int nrows, String fname) throws Exception {
  byte [] family = Bytes.toBytes(fname);

  List<RegionServerThread> rsts = cluster.getLiveRegionServerThreads();
  assertEquals(NUM_RS, rsts.size());

  for (RegionServerThread rst : rsts) {
    HRegionServer hrs = rst.getRegionServer();
    List<HRegionInfo> hris = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
    for (HRegionInfo hri : hris) {
      if (hri.getTable().isSystemTable()) {
        continue;
      }
      LOG.debug("adding data to rs = " + rst.getName() +
          " region = "+ hri.getRegionNameAsString());
      HRegion region = hrs.getOnlineRegion(hri.getRegionName());
      assertTrue(region != null);
      putData(region, hri.getStartKey(), nrows, Bytes.toBytes("q"), family);
    }
  }
}

/**
 * Wait for the specified region server to stop
 * Removes this thread from list of running threads.
 * @param rst
 * @return Name of region server that just went down.
 */
public String waitOnRegionServer(JVMClusterUtil.RegionServerThread rst) {
  while (rst.isAlive()) {
    try {
      LOG.info("Waiting on " +
        rst.getRegionServer().toString());
      rst.join();
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
  for (int i=0;i<regionThreads.size();i++) {
    if (regionThreads.get(i) == rst) {
      regionThreads.remove(i);
      break;
    }
  }
  return rst.getName();
}

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

/**
 * Starts a region server thread running
 *
 * @throws IOException
 * @return New RegionServerThread
 */
public JVMClusterUtil.RegionServerThread startRegionServer()
    throws IOException {
  final Configuration newConf = HBaseConfiguration.create(conf);
  User rsUser =
      HBaseTestingUtility.getDifferentUser(newConf, ".hfs."+index++);
  JVMClusterUtil.RegionServerThread t =  null;
  try {
    t = hbaseCluster.addRegionServer(
        newConf, hbaseCluster.getRegionServers().size(), rsUser);
    t.start();
    t.waitForServerOnline();
  } catch (InterruptedException ie) {
    throw new IOException("Interrupted adding regionserver to cluster", ie);
  }
  return t;
}

/**
 * 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();
    HRegion metaRegion =
      hrs.getOnlineRegion(regionName);
    if (metaRegion != null) {
      index = count;
      break;
    }
    count++;
  }
  return index;
}

public JVMClusterUtil.RegionServerThread addRegionServer(
    final Configuration config, final int index, User user)
throws IOException, InterruptedException {
  return user.runAs(
      new PrivilegedExceptionAction<JVMClusterUtil.RegionServerThread>() {
        public JVMClusterUtil.RegionServerThread run() throws Exception {
          return addRegionServer(config, index);
        }
      });
}

/**
 * @return List of running servers (Some servers may have been killed or
 * aborted during lifetime of cluster; these servers are not included in this
 * list).
 */
public List<JVMClusterUtil.RegionServerThread> getLiveRegionServers() {
  List<JVMClusterUtil.RegionServerThread> liveServers =
    new ArrayList<JVMClusterUtil.RegionServerThread>();
  List<RegionServerThread> list = getRegionServers();
  for (JVMClusterUtil.RegionServerThread rst: list) {
    if (rst.isAlive()) liveServers.add(rst);
    else LOG.info("Not alive " + rst.getName());
  }
  return liveServers;
}

/**
 * Wait for the specified region server to stop
 * Removes this thread from list of running threads.
 * @param serverNumber
 * @return Name of region server that just went down.
 */
public String waitOnRegionServer(int serverNumber) {
  JVMClusterUtil.RegionServerThread regionServerThread =
    this.regionThreads.remove(serverNumber);
  while (regionServerThread.isAlive()) {
    try {
      LOG.info("Waiting on " +
        regionServerThread.getRegionServer().toString());
      regionServerThread.join();
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
  return regionServerThread.getName();
}

private static List<AccessController> getAccessControllers(MiniHBaseCluster cluster) {
  List<AccessController> result = Lists.newArrayList();
  for (RegionServerThread t: cluster.getLiveRegionServerThreads()) {
    for (Region region: t.getRegionServer().getOnlineRegionsLocalContext()) {
      Coprocessor cp = region.getCoprocessorHost()
        .findCoprocessor(AccessController.class.getName());
      if (cp != null) {
        result.add((AccessController)cp);
      }
    }
  }
  return result;
}

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

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

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

/**
 * Counts the total numbers of regions being served by the currently online
 * region servers by asking each how many regions they have.  Does not look
 * at hbase:meta at all.  Count includes catalog tables.
 * @return number of regions being served by all region servers
 */
public long countServedRegions() {
  long count = 0;
  for (JVMClusterUtil.RegionServerThread rst : getLiveRegionServerThreads()) {
    count += rst.getRegionServer().getNumberOfOnlineRegions();
  }
  for (JVMClusterUtil.MasterThread mt : getLiveMasterThreads()) {
    count += mt.getMaster().getNumberOfOnlineRegions();
  }
  return count;
}

/**
 * Do a simulated kill all masters and regionservers. Useful when it is
 * impossible to bring the mini-cluster back for clean shutdown.
 */
public void killAll() {
  for (RegionServerThread rst : getRegionServerThreads()) {
    rst.getRegionServer().abort("killAll");
  }
  for (MasterThread masterThread : getMasterThreads()) {
    masterThread.getMaster().abort("killAll", new Throwable());
  }
}

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

protected int getRegionServerIndex(ServerName serverName) {
  //we have a small number of region servers, this should be fine for now.
  List<RegionServerThread> servers = getRegionServerThreads();
  for (int i=0; i < servers.size(); i++) {
    if (servers.get(i).getRegionServer().getServerName().equals(serverName)) {
      return i;
    }
  }
  return -1;
}

/**
 * Check to make sure the region is open on the specified
 * region server, but not on any other one.
 */
public void assertRegionOnlyOnServer(
    final HRegionInfo hri, final ServerName server,
    final long timeout) throws IOException, InterruptedException {
  long timeoutTime = System.currentTimeMillis() + timeout;
  while (true) {
    List<HRegionInfo> regions = getHBaseAdmin().getOnlineRegions(server);
    if (regions.contains(hri)) {
      List<JVMClusterUtil.RegionServerThread> rsThreads =
        getHBaseCluster().getLiveRegionServerThreads();
      for (JVMClusterUtil.RegionServerThread rsThread: rsThreads) {
        HRegionServer rs = rsThread.getRegionServer();
        if (server.equals(rs.getServerName())) {
          continue;
        }
        Collection<Region> hrs = rs.getOnlineRegionsLocalContext();
        for (Region r: hrs) {
          assertTrue("Region should not be double assigned",
            r.getRegionInfo().getRegionId() != hri.getRegionId());
        }
      }
      return; // good, we are happy
    }
    long now = System.currentTimeMillis();
    if (now > timeoutTime) break;
    Thread.sleep(10);
  }
  fail("Could not find region " + hri.getRegionNameAsString()
    + " on server " + server);
}

@After
public void tearDown() throws Exception {
  for (RegionServerThread rst : TEST_UTIL.getMiniHBaseCluster().getRegionServerThreads()) {
    RegionServerQuotaManager quotaManager = rst.getRegionServer().getRegionServerQuotaManager();
    QuotaCache quotaCache = quotaManager.getQuotaCache();
    quotaCache.getNamespaceQuotaCache().clear();
    quotaCache.getTableQuotaCache().clear();
    quotaCache.getUserQuotaCache().clear();
  }
}

private int getNumberOfOnlineRegions(MiniHBaseCluster cluster) {
  int numFound = 0;
  for (RegionServerThread rst : cluster.getLiveRegionServerThreads()) {
    numFound += rst.getRegionServer().getNumberOfOnlineRegions();
  }
  for (MasterThread mt : cluster.getMasterThreads()) {
    numFound += mt.getMaster().getNumberOfOnlineRegions();
  }
  return numFound;
}

private NavigableSet<String> getDoubleAssignedRegions(
    MiniHBaseCluster cluster) throws IOException {
  NavigableSet<String> online = new TreeSet<String>();
  NavigableSet<String> doubled = new TreeSet<String>();
  for (RegionServerThread rst : cluster.getLiveRegionServerThreads()) {
    for (HRegionInfo region : ProtobufUtil.getOnlineRegions(
        rst.getRegionServer().getRSRpcServices())) {
      if(!online.add(region.getRegionNameAsString())) {
        doubled.add(region.getRegionNameAsString());
      }
    }
  }
  return doubled;
}

/** Flush the given region in the mini cluster. Since no master, we cannot use HBaseAdmin.flush() */
public static void flushRegion(HBaseTestingUtility HTU, HRegionInfo regionInfo) throws IOException {
  for (RegionServerThread rst : HTU.getMiniHBaseCluster().getRegionServerThreads()) {
    Region region = rst.getRegionServer().getRegionByEncodedName(regionInfo.getEncodedName());
    if (region != null) {
      region.flush(true);
      return;
    }
  }
  throw new IOException("Region to flush cannot be found");
}

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

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

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

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

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

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

/**
 * Tests the case where killing a secondary region with unflushed data recovers, and the replica
 * becomes available to read again shortly.
 */
@Test (timeout = 120000)
public void testSecondaryRegionKill() throws Exception {
  try (Connection connection = ConnectionFactory.createConnection(HTU.getConfiguration());
      Table table = connection.getTable(htd.getTableName())) {
    HTU.loadNumericRows(table, fam, 0, 1000);

    // wait for some time to ensure that async wal replication does it's magic
    verifyNumericRowsWithTimeout(table, fam, 0, 1000, 1, 30000);
    verifyNumericRowsWithTimeout(table, fam, 0, 1000, 2, 30000);

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

    Threads.sleep(5000);

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

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

/**
 * Find regionserver other than the one passed.
 * Can't rely on indexes into list of regionservers since crashed servers
 * occupy an index.
 * @param cluster
 * @param notThisOne
 * @return A regionserver that is not <code>notThisOne</code> or null if none
 * found
 */
private HRegionServer getOtherRegionServer(final MiniHBaseCluster cluster,
    final HRegionServer notThisOne) {
  for (RegionServerThread rst: cluster.getRegionServerThreads()) {
    HRegionServer hrs = rst.getRegionServer();
    if (hrs.getServerName().equals(notThisOne.getServerName())) continue;
    if (hrs.isStopping() || hrs.isStopped()) continue;
    return hrs;
  }
  return null;
}

private void evictHFileCache(final Path hfile) throws Exception {
  for (RegionServerThread rst: UTIL.getMiniHBaseCluster().getRegionServerThreads()) {
    HRegionServer rs = rst.getRegionServer();
    rs.getCacheConfig().getBlockCache().evictBlocksByHfileName(hfile.getName());
  }
  Thread.sleep(6000);
}