import org.apache.hadoop.hbase.zookeeper.ZKTable; //导入依赖的package包/类
/**
 * Constructs a new assignment manager.
 *
 * @param master
 * @param serverManager
 * @param catalogTracker
 * @param service
 * @throws KeeperException
 * @throws IOException 
 */
public AssignmentManager(Server master, ServerManager serverManager,
    CatalogTracker catalogTracker, final LoadBalancer balancer,
    final ExecutorService service) throws KeeperException, IOException {
  super(master.getZooKeeper());
  this.master = master;
  this.serverManager = serverManager;
  this.catalogTracker = catalogTracker;
  this.executorService = service;
  this.regionsToReopen = Collections.synchronizedMap
                         (new HashMap<String, HRegionInfo> ());
  Configuration conf = master.getConfiguration();
  this.timeoutMonitor = new TimeoutMonitor(
    conf.getInt("hbase.master.assignment.timeoutmonitor.period", 10000),
    master, serverManager,
    conf.getInt("hbase.master.assignment.timeoutmonitor.timeout", 1800000));
  this.timerUpdater = new TimerUpdater(conf.getInt(
      "hbase.master.assignment.timerupdater.period", 10000), master);
  Threads.setDaemonThreadRunning(timerUpdater.getThread(),
      master.getServerName() + ".timerUpdater");
  this.zkTable = new ZKTable(this.master.getZooKeeper());
  this.maximumAssignmentAttempts =
    this.master.getConfiguration().getInt("hbase.assignment.maximum.attempts", 10);
  this.balancer = balancer;
  this.threadPoolExecutorService = Executors.newCachedThreadPool();
}
 

import org.apache.hadoop.hbase.zookeeper.ZKTable; //导入依赖的package包/类
/**
 * Recover the tables that are not fully moved to ENABLED state. These tables
 * are in ENABLING state when the master restarted/switched
 *
 * @throws KeeperException
 * @throws org.apache.hadoop.hbase.TableNotFoundException
 * @throws IOException
 */
private void recoverTableInEnablingState()
    throws KeeperException, TableNotFoundException, IOException {
  Set<TableName> enablingTables = ZKTable.getEnablingTables(watcher);
  if (enablingTables.size() != 0) {
    for (TableName tableName : enablingTables) {
      // Recover by calling EnableTableHandler
      LOG.info("The table " + tableName
          + " is in ENABLING state.  Hence recovering by moving the table"
          + " to ENABLED state.");
      // enableTable in sync way during master startup,
      // no need to invoke coprocessor
      EnableTableHandler eth = new EnableTableHandler(this.server, tableName,
        catalogTracker, this, tableLockManager, true);
      try {
        eth.prepare();
      } catch (TableNotFoundException e) {
        LOG.warn("Table " + tableName + " not found in hbase:meta to recover.");
        continue;
      }
      eth.process();
    }
  }
}
 

import org.apache.hadoop.hbase.zookeeper.ZKTable; //导入依赖的package包/类
private boolean waitUntilTableDisabled(long timeout, TableName tableName, ZKTable zk) {
  long startTime = System.currentTimeMillis();
  long remaining = timeout;
  boolean disabled = false;
  while (!(disabled = zk.isDisabledTable(tableName)) && remaining > 0) {
    try {
      Thread.sleep(100);
    } catch (InterruptedException e) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Interrupted while waiting for table" + tableName + " set to DISABLED.");
      }
    }
    remaining = timeout - (System.currentTimeMillis() - startTime);
  }
  if (remaining <= 0) {
    return disabled;
  } else {
    return true;
  }
}
 

import org.apache.hadoop.hbase.zookeeper.ZKTable; //导入依赖的package包/类
private boolean waitUntilTableEnabled(long timeout, TableName tableName, ZKTable zk) {
  long startTime = System.currentTimeMillis();
  long remaining = timeout;
  boolean enabled = false;
  while (!(enabled = zk.isEnabledTable(tableName)) && remaining > 0) {
    try {
      Thread.sleep(100);
    } catch (InterruptedException e) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Interrupted while waiting for table " + tableName + "state set to ENABLED.");
      }
    }
    remaining = timeout - (System.currentTimeMillis() - startTime);
  }
  if (remaining <= 0) {
    return enabled;
  } else {
    return true;
  }
}
 

import org.apache.hadoop.hbase.zookeeper.ZKTable; //导入依赖的package包/类
private boolean testTableOnlineState(byte [] tableName, boolean online)
throws IOException {
  if (Bytes.equals(tableName, HConstants.ROOT_TABLE_NAME)) {
    // The root region is always enabled
    return online;
  }
  String tableNameStr = Bytes.toString(tableName);
  try {
    if (online) {
      return ZKTable.isEnabledTable(this.zooKeeper, tableNameStr);
    }
    return ZKTable.isDisabledTable(this.zooKeeper, tableNameStr);
  } catch (KeeperException e) {
    throw new IOException("Enable/Disable failed", e);
  }
}
 

import org.apache.hadoop.hbase.zookeeper.ZKTable; //导入依赖的package包/类
/**
 * Constructs a new assignment manager.
 *
 * @param master
 * @param serverManager
 * @param catalogTracker
 * @param service
 * @throws KeeperException
 * @throws IOException 
 */
public AssignmentManager(Server master, ServerManager serverManager,
    CatalogTracker catalogTracker, final ExecutorService service)
throws KeeperException, IOException {
  super(master.getZooKeeper());
  this.master = master;
  this.serverManager = serverManager;
  this.catalogTracker = catalogTracker;
  this.executorService = service;
  this.regionsToReopen = Collections.synchronizedMap
                         (new HashMap<String, HRegionInfo> ());
  Configuration conf = master.getConfiguration();
  this.timeoutMonitor = new TimeoutMonitor(
    conf.getInt("hbase.master.assignment.timeoutmonitor.period", 10000),
    master, serverManager,
    conf.getInt("hbase.master.assignment.timeoutmonitor.timeout", 1800000));
  Threads.setDaemonThreadRunning(timeoutMonitor.getThread(),
    master.getServerName() + ".timeoutMonitor");
  this.zkTable = new ZKTable(this.master.getZooKeeper());
  this.maximumAssignmentAttempts =
    this.master.getConfiguration().getInt("hbase.assignment.maximum.attempts", 10);
  this.balancer = LoadBalancerFactory.getLoadBalancer(conf);
  this.threadPoolExecutorService = Executors.newCachedThreadPool();
}
 

import org.apache.hadoop.hbase.zookeeper.ZKTable; //导入依赖的package包/类
/**
 * Load the list of disabled tables in ZK into local set.
 * @throws ZooKeeperConnectionException
 * @throws IOException
 */
private void loadDisabledTables()
throws ZooKeeperConnectionException, IOException {
  HConnectionManager.execute(new HConnectable<Void>(conf) {
    @Override
    public Void connect(HConnection connection) throws IOException {
      ZooKeeperWatcher zkw = connection.getZooKeeperWatcher();
      try {
        for (String tableName : ZKTable.getDisabledOrDisablingTables(zkw)) {
          disabledTables.add(Bytes.toBytes(tableName));
        }
      } catch (KeeperException ke) {
        throw new IOException(ke);
      }
      return null;
    }
  });
}
 

import org.apache.hadoop.hbase.zookeeper.ZKTable; //导入依赖的package包/类
/**
 * Recover the tables that are not fully moved to ENABLED state. These tables
 * are in ENABLING state when the master restarted/switched
 *
 * @throws KeeperException
 * @throws org.apache.hadoop.hbase.TableNotFoundException
 * @throws IOException
 */
private void recoverTableInEnablingState()
    throws KeeperException, TableNotFoundException, IOException {
  Set<TableName> enablingTables = ZKTable.getEnablingTables(watcher);
  if (enablingTables.size() != 0) {
    for (TableName tableName : enablingTables) {
      // Recover by calling EnableTableHandler
      LOG.info("The table " + tableName
          + " is in ENABLING state.  Hence recovering by moving the table"
          + " to ENABLED state.");
      // enableTable in sync way during master startup,
      // no need to invoke coprocessor
      new EnableTableHandler(this.server, tableName,
          catalogTracker, this, tableLockManager, true).prepare().process();
    }
  }
}
 

import org.apache.hadoop.hbase.zookeeper.ZKTable; //导入依赖的package包/类
/**
 * Recover the tables that are not fully moved to ENABLED state. These tables
 * are in ENABLING state when the master restarted/switched
 *
 * @throws KeeperException
 * @throws TableNotFoundException
 * @throws IOException
 */
private void recoverTableInEnablingState()
    throws KeeperException, TableNotFoundException, IOException {
  Set<String> enablingTables = ZKTable.getEnablingTables(watcher);
  if (enablingTables.size() != 0) {
    for (String tableName : enablingTables) {
      // Recover by calling EnableTableHandler
      LOG.info("The table " + tableName
          + " is in ENABLING state.  Hence recovering by moving the table"
          + " to ENABLED state.");
      // enableTable in sync way during master startup,
      // no need to invoke coprocessor
      new EnableTableHandler(this.server, tableName.getBytes(),
          catalogTracker, this, true).process();
    }
  }
}
 

import org.apache.hadoop.hbase.zookeeper.ZKTable; //导入依赖的package包/类
private boolean waitUntilTableDisabled(long timeout, String tableName, ZKTable zk) {
  long startTime = System.currentTimeMillis();
  long remaining = timeout;
  boolean disabled = false;
  while (!(disabled = zk.isDisabledTable(tableName)) && remaining > 0) {
    try {
      Thread.sleep(100);
    } catch (InterruptedException e) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Interrupted while waiting for table" + tableName + " set to DISABLED.");
      }
    }
    remaining = timeout - (System.currentTimeMillis() - startTime);
  }
  if (remaining <= 0) {
    return disabled;
  } else {
    return true;
  }
}
 

import org.apache.hadoop.hbase.zookeeper.ZKTable; //导入依赖的package包/类
private boolean waitUntilTableEnabled(long timeout, String tableName, ZKTable zk) {
  long startTime = System.currentTimeMillis();
  long remaining = timeout;
  boolean enabled = false;
  while (!(enabled = zk.isEnabledTable(tableName)) && remaining > 0) {
    try {
      Thread.sleep(100);
    } catch (InterruptedException e) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Interrupted while waiting for table " + tableName + "state set to ENABLED.");
      }
    }
    remaining = timeout - (System.currentTimeMillis() - startTime);
  }
  if (remaining <= 0) {
    return enabled;
  } else {
    return true;
  }

}
 

import org.apache.hadoop.hbase.zookeeper.ZKTable; //导入依赖的package包/类
/**
 * Assigns all user regions, if any exist.  Used during cluster startup.
 * <p>
 * This is a synchronous call and will return once every region has been
 * assigned.  If anything fails, an exception is thrown and the cluster
 * should be shutdown.
 * @throws InterruptedException
 * @throws IOException
 * @throws KeeperException
 */
private void assignAllUserRegions()
    throws IOException, InterruptedException, KeeperException {
  // Cleanup any existing ZK nodes and start watching
  ZKAssign.deleteAllNodes(watcher);
  ZKUtil.listChildrenAndWatchForNewChildren(this.watcher,
    this.watcher.assignmentZNode);
  failoverCleanupDone();

  // Skip assignment for regions of tables in DISABLING state because during clean cluster startup
  // no RS is alive and regions map also doesn't have any information about the regions.
  // See HBASE-6281.
  Set<TableName> disabledOrDisablingOrEnabling = ZKTable.getDisabledOrDisablingTables(watcher);
  disabledOrDisablingOrEnabling.addAll(ZKTable.getEnablingTables(watcher));
  // Scan hbase:meta for all user regions, skipping any disabled tables
  Map<HRegionInfo, ServerName> allRegions;
  SnapshotOfRegionAssignmentFromMeta snapshotOfRegionAssignment =
     new SnapshotOfRegionAssignmentFromMeta(catalogTracker, disabledOrDisablingOrEnabling, true);
  snapshotOfRegionAssignment.initialize();
  allRegions = snapshotOfRegionAssignment.getRegionToRegionServerMap();
  if (allRegions == null || allRegions.isEmpty()) return;

  // Determine what type of assignment to do on startup
  boolean retainAssignment = server.getConfiguration().
    getBoolean("hbase.master.startup.retainassign", true);

  if (retainAssignment) {
    assign(allRegions);
  } else {
    List<HRegionInfo> regions = new ArrayList<HRegionInfo>(allRegions.keySet());
    assign(regions);
  }

  for (HRegionInfo hri : allRegions.keySet()) {
    TableName tableName = hri.getTable();
    if (!zkTable.isEnabledTable(tableName)) {
      setEnabledTable(tableName);
    }
  }
}
 

import org.apache.hadoop.hbase.zookeeper.ZKTable; //导入依赖的package包/类
/**
 * Recover the tables that were not fully moved to DISABLED state. These
 * tables are in DISABLING state when the master restarted/switched.
 *
 * @throws KeeperException
 * @throws TableNotFoundException
 * @throws IOException
 */
private void recoverTableInDisablingState()
    throws KeeperException, TableNotFoundException, IOException {
  Set<TableName> disablingTables = ZKTable.getDisablingTables(watcher);
  if (disablingTables.size() != 0) {
    for (TableName tableName : disablingTables) {
      // Recover by calling DisableTableHandler
      LOG.info("The table " + tableName
          + " is in DISABLING state.  Hence recovering by moving the table"
          + " to DISABLED state.");
      new DisableTableHandler(this.server, tableName, catalogTracker,
          this, tableLockManager, true).prepare().process();
    }
  }
}
 

import org.apache.hadoop.hbase.zookeeper.ZKTable; //导入依赖的package包/类
private void setTablesInZK() throws IOException, KeeperException {
  if (tablesToBeSetInZK != null && !tablesToBeSetInZK.isEmpty()) {
    ZKTable zkTable = new ZKTable(this.watcher);
    for (Pair<String, State> p : tablesToBeSetInZK) {
      setStateInZK(zkTable, p.getFirst(), p.getSecond());
    }
  }
}
 

import org.apache.hadoop.hbase.zookeeper.ZKTable; //导入依赖的package包/类
private void setStateInZK(ZKTable zkTable, String tableName, State state) throws IOException,
    KeeperException {
  if (state == State.ENABLED) {
    zkTable.setEnabledTable(TableName.valueOf(tableName));
  }
  if (state == State.DISABLED) {
    zkTable.setDisabledTable(TableName.valueOf(tableName));
  }
}
 

import org.apache.hadoop.hbase.zookeeper.ZKTable; //导入依赖的package包/类
/**
 * Constructs a new assignment manager.
 *
 * @param server
 * @param serverManager
 * @param catalogTracker
 * @param service
 * @throws KeeperException
 * @throws IOException
 */
public AssignmentManager(Server server, ServerManager serverManager,
    CatalogTracker catalogTracker, final LoadBalancer balancer,
    final ExecutorService service, MetricsMaster metricsMaster) throws KeeperException, IOException {
  super(server.getZooKeeper());
  this.server = server;
  this.serverManager = serverManager;
  this.catalogTracker = catalogTracker;
  this.executorService = service;
  this.regionsToReopen = Collections.synchronizedMap
                         (new HashMap<String, HRegionInfo> ());
  Configuration conf = server.getConfiguration();
  this.timeoutMonitor = new TimeoutMonitor(
    conf.getInt("hbase.master.assignment.timeoutmonitor.period", 30000),
    server, serverManager,
    conf.getInt("hbase.master.assignment.timeoutmonitor.timeout", 600000));
  this.timerUpdater = new TimerUpdater(conf.getInt(
    "hbase.master.assignment.timerupdater.period", 10000), server);
  Threads.setDaemonThreadRunning(timerUpdater.getThread(),
    server.getServerName() + ".timerUpdater");
  this.zkTable = new ZKTable(this.watcher);
  this.maximumAttempts =
    this.server.getConfiguration().getInt("hbase.assignment.maximum.attempts", 10);
  this.balancer = balancer;
  int maxThreads = conf.getInt("hbase.assignment.threads.max", 30);
  this.threadPoolExecutorService = Threads.getBoundedCachedThreadPool(
    maxThreads, 60L, TimeUnit.SECONDS, Threads.newDaemonThreadFactory("hbase-am"));
  this.metricsMaster = metricsMaster;// can be null only with tests.
  this.regionStates = new RegionStates(server, serverManager);

  int workers = conf.getInt("hbase.assignment.zkevent.workers", 20);
  ThreadFactory threadFactory = Threads.newDaemonThreadFactory("hbase-am-zkevent-worker");
  zkEventWorkers = Threads.getBoundedCachedThreadPool(workers, 60L,
          TimeUnit.SECONDS, threadFactory);
}
 

import org.apache.hadoop.hbase.zookeeper.ZKTable; //导入依赖的package包/类
/**
 * Assigns all user regions, if any exist.  Used during cluster startup.
 * <p>
 * This is a synchronous call and will return once every region has been
 * assigned.  If anything fails, an exception is thrown and the cluster
 * should be shutdown.
 * @throws InterruptedException
 * @throws IOException
 * @throws KeeperException
 */
private void assignAllUserRegions()
    throws IOException, InterruptedException, KeeperException {
  // Cleanup any existing ZK nodes and start watching
  ZKAssign.deleteAllNodes(watcher);
  ZKUtil.listChildrenAndWatchForNewChildren(this.watcher,
    this.watcher.assignmentZNode);
  failoverCleanupDone();

  // Skip assignment for regions of tables in DISABLING state because during clean cluster startup
  // no RS is alive and regions map also doesn't have any information about the regions.
  // See HBASE-6281.
  Set<String> disabledOrDisablingOrEnabling = ZKTable.getDisabledOrDisablingTables(watcher);
  disabledOrDisablingOrEnabling.addAll(ZKTable.getEnablingTables(watcher));
  // Scan META for all user regions, skipping any disabled tables
  Map<HRegionInfo, ServerName> allRegions = MetaReader.fullScan(
    catalogTracker, disabledOrDisablingOrEnabling, true);
  if (allRegions == null || allRegions.isEmpty()) return;

  // Determine what type of assignment to do on startup
  boolean retainAssignment = server.getConfiguration().
    getBoolean("hbase.master.startup.retainassign", true);

  if (retainAssignment) {
    assign(allRegions);
  } else {
    List<HRegionInfo> regions = new ArrayList<HRegionInfo>(allRegions.keySet());
    assign(regions);
  }

  for (HRegionInfo hri : allRegions.keySet()) {
    String tableName = hri.getTableNameAsString();
    if (!zkTable.isEnabledTable(tableName)) {
      setEnabledTable(tableName);
    }
  }
}
 

import org.apache.hadoop.hbase.zookeeper.ZKTable; //导入依赖的package包/类
/**
 * Recover the tables that were not fully moved to DISABLED state. These
 * tables are in DISABLING state when the master restarted/switched.
 *
 * @throws KeeperException
 * @throws TableNotFoundException
 * @throws IOException
 */
private void recoverTableInDisablingState()
    throws KeeperException, TableNotFoundException, IOException {
  Set<String> disablingTables = ZKTable.getDisablingTables(watcher);
  if (disablingTables.size() != 0) {
    for (String tableName : disablingTables) {
      // Recover by calling DisableTableHandler
      LOG.info("The table " + tableName
          + " is in DISABLING state.  Hence recovering by moving the table"
          + " to DISABLED state.");
      new DisableTableHandler(this.server, tableName.getBytes(),
          catalogTracker, this, true).process();
    }
  }
}
 

import org.apache.hadoop.hbase.zookeeper.ZKTable; //导入依赖的package包/类
/**
 * @return Instance of ZKTable.
 */
public ZKTable getZKTable() {
  // These are 'expensive' to make involving trip to zk ensemble so allow
  // sharing.
  return this.zkTable;
}
 

import org.apache.hadoop.hbase.zookeeper.ZKTable; //导入依赖的package包/类
@Test
public void testShouldNotCompeleteOpenedRegionSuccessfullyIfVersionMismatches()
    throws Exception {
  HRegion region = null;
  try {
    int testIndex = 0;
    TEST_UTIL.startMiniZKCluster();
    final Server server = new MockServer(TEST_UTIL);
    HTableDescriptor htd = new HTableDescriptor(
        "testShouldNotCompeleteOpenedRegionSuccessfullyIfVersionMismatches");
    HRegionInfo hri = new HRegionInfo(htd.getName(),
        Bytes.toBytes(testIndex), Bytes.toBytes(testIndex + 1));
    region = HRegion.createHRegion(hri, TEST_UTIL.getDataTestDir(), TEST_UTIL.getConfiguration(), htd);
    assertNotNull(region);
    AssignmentManager am = Mockito.mock(AssignmentManager.class);
    when(am.isRegionInTransition(hri)).thenReturn(
        new RegionState(region.getRegionInfo(), RegionState.State.OPEN,
            System.currentTimeMillis(), server.getServerName()));
    // create a node with OPENED state
    zkw = HBaseTestingUtility.createAndForceNodeToOpenedState(TEST_UTIL,
        region, server.getServerName());
    when(am.getZKTable()).thenReturn(new ZKTable(zkw));
    Stat stat = new Stat();
    String nodeName = ZKAssign.getNodeName(zkw, region.getRegionInfo()
        .getEncodedName());
    ZKUtil.getDataAndWatch(zkw, nodeName, stat);

    // use the version for the OpenedRegionHandler
    OpenedRegionHandler handler = new OpenedRegionHandler(server, am, region
        .getRegionInfo(), server.getServerName(), stat.getVersion());
    // Once again overwrite the same znode so that the version changes.
    ZKAssign.transitionNode(zkw, region.getRegionInfo(), server
        .getServerName(), EventType.RS_ZK_REGION_OPENED,
        EventType.RS_ZK_REGION_OPENED, stat.getVersion());

    // Should not invoke assignmentmanager.regionOnline. If it is 
    // invoked as per current mocking it will throw null pointer exception.
    boolean expectedException = false;
    try {
      handler.process();
    } catch (Exception e) {
      expectedException = true;
    }
    assertFalse("The process method should not throw any exception.",
        expectedException);
    List<String> znodes = ZKUtil.listChildrenAndWatchForNewChildren(zkw,
        zkw.assignmentZNode);
    String regionName = znodes.get(0);
    assertEquals("The region should not be opened successfully.", regionName,
        region.getRegionInfo().getEncodedName());
  } finally {
    region.close();
    region.getLog().closeAndDelete();
    TEST_UTIL.shutdownMiniZKCluster();
  }
}