import org.apache.hadoop.hbase.master.AssignmentManager.RegionState; //导入依赖的package包/类
public static void waitUntilAssigned(HBaseAdmin admin,
    HRegionInfo region) throws IOException, InterruptedException {
  long timeout = admin.getConfiguration().getLong("hbase.hbck.assign.timeout", 120000);
  long expiration = timeout + System.currentTimeMillis();
  while (System.currentTimeMillis() < expiration) {
    try {
      Map<String, RegionState> rits=
          admin.getClusterStatus().getRegionsInTransition();

      if (rits.keySet() != null && !rits.keySet().contains(region.getEncodedName())) {
        // yay! no longer RIT
        return;
      }
      // still in rit
      LOG.info("Region still in transition, waiting for "
          + "it to become assigned: " + region);
    } catch (IOException e) {
      LOG.warn("Exception when waiting for region to become assigned,"
          + " retrying", e);
    }
    Thread.sleep(1000);
  }
  throw new IOException("Region " + region + " failed to move out of " +
      "transition within timeout " + timeout + "ms");
}
 

import org.apache.hadoop.hbase.master.AssignmentManager.RegionState; //导入依赖的package包/类
public ClusterStatus(final String hbaseVersion, final String clusterid,
    final Map<ServerName, HServerLoad> servers,
    final Collection<ServerName> deadServers,
    final ServerName master,
    final Collection<ServerName> backupMasters,
    final Map<String, RegionState> rit,
    final String[] masterCoprocessors) {
  this.hbaseVersion = hbaseVersion;
  this.liveServers = servers;
  this.deadServers = deadServers;
  this.master = master;
  this.backupMasters = backupMasters;
  this.intransition = rit;
  this.clusterId = clusterid;
  this.masterCoprocessors = masterCoprocessors;
}
 

import org.apache.hadoop.hbase.master.AssignmentManager.RegionState; //导入依赖的package包/类
static void waitForRegionPendingOpenInRIT(AssignmentManager am, String encodedName)
  throws InterruptedException {
  // We used to do a check like this:
  //!Mocking.verifyRegionState(this.watcher, REGIONINFO, EventType.M_ZK_REGION_OFFLINE)) {
  // There is a race condition with this: because we may do the transition to
  // RS_ZK_REGION_OPENING before the RIT is internally updated. We need to wait for the
  // RIT to be as we need it to be instead. This cannot happen in a real cluster as we
  // update the RIT before sending the openRegion request.

  boolean wait = true;
  while (wait) {
    RegionState state = am.getRegionsInTransition().get(encodedName);
    if (state != null && state.isPendingOpen()){
      wait = false;
    } else {
      Thread.sleep(1);
    }
  }
}
 

import org.apache.hadoop.hbase.master.AssignmentManager.RegionState; //导入依赖的package包/类
@Override
public void nodeDeleted(final String path) {
  if (path.startsWith(this.watcher.assignmentZNode)) {
    String regionName = ZKAssign.getRegionName(this.master.getZooKeeper(), path);
    RegionState rs = this.regionsInTransition.get(regionName);
    if (rs != null) {
      HRegionInfo regionInfo = rs.getRegion();
      if (rs.isSplitting() || rs.isSplit()) {
        LOG.debug("Ephemeral node deleted, regionserver crashed?, " +
          "clearing from RIT; rs=" + rs);
        clearRegionFromTransition(rs.getRegion());
      } else {
        LOG.debug("The znode of region " + regionInfo.getRegionNameAsString()
            + " has been deleted.");
        if (rs.isOpened()) {
          makeRegionOnline(rs, regionInfo);
        }
      }
    }
  }
}
 

import org.apache.hadoop.hbase.master.AssignmentManager.RegionState; //导入依赖的package包/类
/**
 * Touch timers for all regions in transition that have the passed
 * <code>sn</code> in common.
 * Call this method whenever a server checks in.  Doing so helps the case where
 * a new regionserver has joined the cluster and its been given 1k regions to
 * open.  If this method is tickled every time the region reports in a
 * successful open then the 1k-th region won't be timed out just because its
 * sitting behind the open of 999 other regions.  This method is NOT used
 * as part of bulk assign -- there we have a different mechanism for extending
 * the regions in transition timer (we turn it off temporarily -- because
 * there is no regionplan involved when bulk assigning.
 * @param sn
 */
private void updateTimers(final ServerName sn) {
  // This loop could be expensive.
  // First make a copy of current regionPlan rather than hold sync while
  // looping because holding sync can cause deadlock.  Its ok in this loop
  // if the Map we're going against is a little stale
  Map<String, RegionPlan> copy = new HashMap<String, RegionPlan>();
  synchronized(this.regionPlans) {
    copy.putAll(this.regionPlans);
  }
  for (Map.Entry<String, RegionPlan> e: copy.entrySet()) {
    if (e.getValue() == null || e.getValue().getDestination() == null) continue;
    if (!e.getValue().getDestination().equals(sn)) continue;
    RegionState rs = null;
    synchronized (this.regionsInTransition) {
      rs = this.regionsInTransition.get(e.getKey());
    }
    if (rs == null) continue;
    rs.updateTimestampToNow();
  }
}
 

import org.apache.hadoop.hbase.master.AssignmentManager.RegionState; //导入依赖的package包/类
/**
 * @param region
 * @param setOfflineInZK
 * @param forceNewPlan
 * @param hijack
 *          - true new assignment is needed, false otherwise
 */
public void assign(HRegionInfo region, boolean setOfflineInZK,
    boolean forceNewPlan, boolean hijack) {
  // If hijack is true do not call disableRegionIfInRIT as 
  // we have not yet moved the znode to OFFLINE state.
  if (!hijack && isDisabledorDisablingRegionInRIT(region)) {
    return;
  }
  if (this.serverManager.isClusterShutdown()) {
    LOG.info("Cluster shutdown is set; skipping assign of " +
      region.getRegionNameAsString());
    return;
  }
  RegionState state = addToRegionsInTransition(region,
      hijack);
  synchronized (state) {
    assign(region, state, setOfflineInZK, forceNewPlan, hijack);
  }
}
 

import org.apache.hadoop.hbase.master.AssignmentManager.RegionState; //导入依赖的package包/类
@Override
public void processResult(int rc, String path, Object ctx, String name) {
  if (rc != 0) {
    // Thisis resultcode.  If non-zero, need to resubmit.
    LOG.warn("rc != 0 for " + path + " -- retryable connectionloss -- " +
      "FIX see http://wiki.apache.org/hadoop/ZooKeeper/FAQ#A2");
    this.zkw.abort("Connectionloss writing unassigned at " + path +
      ", rc=" + rc, null);
    return;
  }
  LOG.debug("rs=" + (RegionState)ctx + ", server=" + this.destination.toString());
  // Async exists to set a watcher so we'll get triggered when
  // unassigned node changes.
  this.zkw.getRecoverableZooKeeper().getZooKeeper().exists(path, this.zkw,
    new ExistsUnassignedAsyncCallback(this.counter, destination), ctx);
}
 

import org.apache.hadoop.hbase.master.AssignmentManager.RegionState; //导入依赖的package包/类
/**
 * Set region as OFFLINED up in zookeeper asynchronously.
 * @param state
 * @return True if we succeeded, false otherwise (State was incorrect or failed
 * updating zk).
 */
boolean asyncSetOfflineInZooKeeper(final RegionState state,
    final AsyncCallback.StringCallback cb, final Object ctx) {
  if (!state.isClosed() && !state.isOffline()) {
      new RuntimeException("Unexpected state trying to OFFLINE; " + state);
    this.master.abort("Unexpected state trying to OFFLINE; " + state,
      new IllegalStateException());
    return false;
  }
  state.update(RegionState.State.OFFLINE);
  try {
    ZKAssign.asyncCreateNodeOffline(master.getZooKeeper(), state.getRegion(),
      this.master.getServerName(), cb, ctx);
  } catch (KeeperException e) {
    master.abort("Unexpected ZK exception creating/setting node OFFLINE", e);
    return false;
  }
  return true;
}
 

import org.apache.hadoop.hbase.master.AssignmentManager.RegionState; //导入依赖的package包/类
/**
 * Wait on region to clear regions-in-transition.
 * @param hri Region to wait on.
 * @throws IOException
 */
public void waitOnRegionToClearRegionsInTransition(final HRegionInfo hri)
throws IOException {
  if (isRegionInTransition(hri) == null) return;
  RegionState rs = null;
  // There is already a timeout monitor on regions in transition so I
  // should not have to have one here too?
  while(!this.master.isStopped() && (rs = isRegionInTransition(hri)) != null) {
    Threads.sleep(1000);
    LOG.info("Waiting on " + rs + " to clear regions-in-transition");
  }
  if (this.master.isStopped()) {
    LOG.info("Giving up wait on regions in " +
      "transition because stoppable.isStopped is set");
  }
}
 

import org.apache.hadoop.hbase.master.AssignmentManager.RegionState; //导入依赖的package包/类
@Override
protected void chore() {
  // If bulkAssign in progress, suspend checks
  if (this.bulkAssign) return;
  boolean allRSsOffline = this.serverManager.getOnlineServersList().
    isEmpty();

  synchronized (regionsInTransition) {
    // Iterate all regions in transition checking for time outs
    long now = System.currentTimeMillis();
    for (RegionState regionState : regionsInTransition.values()) {
      if (regionState.getStamp() + timeout <= now) {
       //decide on action upon timeout
        actOnTimeOut(regionState);
      } else if (this.allRegionServersOffline && !allRSsOffline) {
        // if some RSs just came back online, we can start the
        // the assignment right away
        actOnTimeOut(regionState);
      }
    }
  }
  setAllRegionServersOffline(allRSsOffline);
}
 

import org.apache.hadoop.hbase.master.AssignmentManager.RegionState; //导入依赖的package包/类
private boolean checkRegionInTransition(ObserverContext<MasterCoprocessorEnvironment> ctx,
    HRegionInfo hri) {
  MasterServices master = ctx.getEnvironment().getMasterServices();
  AssignmentManager am = master.getAssignmentManager();
  boolean isRegionInTransition = false;
  String tableName = hri.getTableNameAsString();
  if (false == IndexUtils.isIndexTable(tableName)) {
    NavigableMap<String, RegionState> regionsInTransition = am.getRegionsInTransition();
    RegionState regionState = regionsInTransition.get(hri.getEncodedName());
    if (regionState != null) {
      isRegionInTransition = true;
    } else {
      String indexTableName = IndexUtils.getIndexTableName(tableName);
      for (Entry<String, RegionState> region : regionsInTransition.entrySet()) {
        HRegionInfo regionInfo = region.getValue().getRegion();
        if (indexTableName.equals(regionInfo.getTableNameAsString())) {
          if (Bytes.compareTo(hri.getStartKey(), regionInfo.getStartKey()) == 0) {
            isRegionInTransition = true;
            break;
          }
        }
      }
    }
  }
  return isRegionInTransition;
}
 

import org.apache.hadoop.hbase.master.AssignmentManager.RegionState; //导入依赖的package包/类
private void dumpRIT(HMaster master, PrintWriter out) {
  NavigableMap<String, RegionState> regionsInTransition =
    master.getAssignmentManager().getRegionsInTransition();
  for (Map.Entry<String, RegionState> e : regionsInTransition.entrySet()) {
    String rid = e.getKey();
    RegionState rs = e.getValue();
    out.println("Region " + rid + ": " + rs.toDescriptiveString());
  }
}
 

import org.apache.hadoop.hbase.master.AssignmentManager.RegionState; //导入依赖的package包/类
@Override
public void process() {
  // Code to defend against case where we get SPLIT before region open
  // processing completes; temporary till we make SPLITs go via zk -- 0.92.
  RegionState regionState = this.assignmentManager.isRegionInTransition(regionInfo);
  boolean openedNodeDeleted = false;
  if (regionState != null
      && regionState.getState().equals(RegionState.State.OPEN)) {
    openedNodeDeleted = deleteOpenedNode(expectedVersion);
    if (!openedNodeDeleted) {
      LOG.error("The znode of region " + regionInfo.getRegionNameAsString()
          + " could not be deleted.");
    }
  } else {
    LOG.warn("Skipping the onlining of " + regionInfo.getRegionNameAsString() +
      " because regions is NOT in RIT -- presuming this is because it SPLIT");
  }
  if (!openedNodeDeleted) {
    if (this.assignmentManager.getZKTable().isDisablingOrDisabledTable(
        regionInfo.getTableNameAsString())) {
      debugLog(regionInfo, "Opened region "
          + regionInfo.getRegionNameAsString() + " but "
          + "this table is disabled, triggering close of region");
      assignmentManager.unassign(regionInfo);
    }
  }
}
 

import org.apache.hadoop.hbase.master.AssignmentManager.RegionState; //导入依赖的package包/类
private void removeRITsOfRregionInDisablingOrDisabledTables(List<HRegionInfo> toAssign,
    RegionState rit, AssignmentManager assignmentManager, HRegionInfo hri) {

  if (!assignmentManager.getZKTable().isDisablingOrDisabledTable(hri.getTableNameAsString())) {
    return;
  }

  // To avoid region assignment if table is in disabling or disabled state.
  toAssign.remove(hri);

  if (rit != null) {
    assignmentManager.deleteNodeAndOfflineRegion(hri);
  }
}
 

import org.apache.hadoop.hbase.master.AssignmentManager.RegionState; //导入依赖的package包/类
private void testSSHWhenSourceRSandDestRSInRegionPlanGoneDown(boolean regionInOffline)
    throws IOException, KeeperException, ServiceException {
  // We need a mocked catalog tracker.
  CatalogTracker ct = Mockito.mock(CatalogTracker.class);
  // Create an AM.
  AssignmentManagerWithExtrasForTesting am =
      setUpMockedAssignmentManager(this.server, this.serverManager);
  // adding region in pending open.
  if (regionInOffline) {
    ServerName MASTER_SERVERNAME = new ServerName("example.org", 1111, 1111);
    am.regionsInTransition.put(REGIONINFO.getEncodedName(), new RegionState(REGIONINFO,
        State.OFFLINE, System.currentTimeMillis(), MASTER_SERVERNAME));
  } else {
    am.regionsInTransition.put(REGIONINFO.getEncodedName(), new RegionState(REGIONINFO,
        State.OPENING, System.currentTimeMillis(), SERVERNAME_B));
  }
  // adding region plan
  am.regionPlans.put(REGIONINFO.getEncodedName(), new RegionPlan(REGIONINFO, SERVERNAME_A, SERVERNAME_B));
  am.getZKTable().setEnabledTable(REGIONINFO.getTableNameAsString());

  try {
    processServerShutdownHandler(ct, am, false, SERVERNAME_A);
    processServerShutdownHandler(ct, am, false, SERVERNAME_B);
    if(regionInOffline){
      assertFalse("Assign should not be invoked.", am.assignInvoked);
    } else {
      assertTrue("Assign should be invoked.", am.assignInvoked);
    }
  } finally {
    am.regionsInTransition.remove(REGIONINFO.getEncodedName());
    am.regionPlans.remove(REGIONINFO.getEncodedName());
  }
}
 

import org.apache.hadoop.hbase.master.AssignmentManager.RegionState; //导入依赖的package包/类
static void waitForRegionOfflineInRIT(AssignmentManager am, String encodedName)
    throws InterruptedException {
  boolean wait = true;
  while (wait) {
    RegionState state = am.getRegionsInTransition().get(encodedName);
    if (state != null && state.isOffline()) {
      wait = false;
    } else {
      Thread.sleep(1);
    }
  }
}
 

import org.apache.hadoop.hbase.master.AssignmentManager.RegionState; //导入依赖的package包/类
@Before
public void setupBasicMocks() {
  conf = HBaseConfiguration.create();
  
  master = Mockito.mock(HMaster.class);
  Mockito.doReturn(FAKE_HOST).when(master).getServerName();
  Mockito.doReturn(conf).when(master).getConfiguration();
  
  // Fake ActiveMasterManager
  ActiveMasterManager amm = Mockito.mock(ActiveMasterManager.class);
  Mockito.doReturn(amm).when(master).getActiveMasterManager();
  Mockito.doReturn(FAKE_HOST).when(amm).getActiveMaster();

  // Fake serverManager
  ServerManager serverManager = Mockito.mock(ServerManager.class);
  Mockito.doReturn(1.0).when(serverManager).getAverageLoad();
  Mockito.doReturn(serverManager).when(master).getServerManager();

  // Fake AssignmentManager and RIT
  AssignmentManager am = Mockito.mock(AssignmentManager.class);
  NavigableMap<String, RegionState> regionsInTransition =
    Maps.newTreeMap();
  regionsInTransition.put("r1",
      new RegionState(FAKE_HRI, RegionState.State.CLOSING, 12345L, FAKE_HOST));
  Mockito.doReturn(regionsInTransition).when(am).getRegionsInTransition();
  Mockito.doReturn(am).when(master).getAssignmentManager();
  
  // Fake ZKW
  ZooKeeperWatcher zkw = Mockito.mock(ZooKeeperWatcher.class);
  Mockito.doReturn("fakequorum").when(zkw).getQuorum();
  Mockito.doReturn(zkw).when(master).getZooKeeperWatcher();

  // Mock admin
  admin = Mockito.mock(HBaseAdmin.class); 
  Mockito.when(admin.getConfiguration()).thenReturn(conf);
}
 

import org.apache.hadoop.hbase.master.AssignmentManager.RegionState; //导入依赖的package包/类
@Test
public void testAssignmentManagerTruncatedList() throws IOException {
  AssignmentManager am = Mockito.mock(AssignmentManager.class);
  
  // Add 100 regions as in-transition
  NavigableMap<String, RegionState> regionsInTransition =
    Maps.newTreeMap();
  for (byte i = 0; i < 100; i++) {
    HRegionInfo hri = new HRegionInfo(FAKE_TABLE.getName(),
        new byte[]{i}, new byte[]{(byte) (i+1)});
    regionsInTransition.put(hri.getEncodedName(),
        new RegionState(hri, RegionState.State.CLOSING, 12345L, FAKE_HOST));
  }
  // Add META in transition as well
  regionsInTransition.put(
      HRegionInfo.FIRST_META_REGIONINFO.getEncodedName(),
      new RegionState(HRegionInfo.FIRST_META_REGIONINFO,
                      RegionState.State.CLOSING, 12345L, FAKE_HOST));
  Mockito.doReturn(regionsInTransition).when(am).getRegionsInTransition();

  // Render to a string
  StringWriter sw = new StringWriter();
  new AssignmentManagerStatusTmpl()
    .setLimit(50)
    .render(sw, am);
  String result = sw.toString();

  // Should always include META
  assertTrue(result.contains(HRegionInfo.FIRST_META_REGIONINFO.getEncodedName()));
  
  // Make sure we only see 50 of them
  Matcher matcher = Pattern.compile("CLOSING").matcher(result);
  int count = 0;
  while (matcher.find()) {
    count++;
  }
  assertEquals(50, count);
}