import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos; //导入依赖的package包/类
/**
 * Checks and sets table state in ZK. Sets no watches.
 * {@inheritDoc}
 */
@Override
public boolean setTableStateIfInStates(TableName tableName,
                                       ZooKeeperProtos.Table.State newState,
                                       ZooKeeperProtos.Table.State... states)
    throws CoordinatedStateException {
  synchronized (this.cache) {
    // Transition ENABLED->DISABLING has to be performed with a hack, because
    // we treat empty state as enabled in this case because 0.92- clusters.
    if (
        (newState == ZooKeeperProtos.Table.State.DISABLING) &&
             this.cache.get(tableName) != null && !isTableState(tableName, states) ||
        (newState != ZooKeeperProtos.Table.State.DISABLING &&
             !isTableState(tableName, states) )) {
      return false;
    }
    try {
      setTableStateInZK(tableName, newState);
    } catch (KeeperException e) {
      throw new CoordinatedStateException(e);
    }
    return true;
  }
}
 

import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos; //导入依赖的package包/类
/**
 * {@inheritDoc}
 */
@Override
public void checkAndRemoveTableState(TableName tableName, ZooKeeperProtos.Table.State states,
                                     boolean deletePermanentState)
    throws CoordinatedStateException {
  synchronized (this.cache) {
    if (isTableState(tableName, states)) {
      this.cache.remove(tableName);
      if (deletePermanentState) {
        try {
          ZKUtil.deleteNodeFailSilent(this.watcher,
              ZKUtil.joinZNode(this.watcher.tableZNode, tableName.getNameAsString()));
        } catch (KeeperException e) {
          throw new CoordinatedStateException(e);
        }
      }
    }
  }
}
 

import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos; //导入依赖的package包/类
/**
 * Gets a list of all the tables of specified states in zookeeper.
 * @return Set of tables of specified states, empty Set if none
 * @throws KeeperException
 */
Set<TableName> getAllTables(final ZooKeeperProtos.Table.State... states)
    throws KeeperException, InterruptedIOException {

  Set<TableName> allTables = new HashSet<TableName>();
  List<String> children =
    ZKUtil.listChildrenNoWatch(watcher, watcher.tableZNode);
  if(children == null) return allTables;
  for (String child: children) {
    TableName tableName = TableName.valueOf(child);
    ZooKeeperProtos.Table.State state;
    try {
      state = getTableState(watcher, tableName);
    } catch (InterruptedException e) {
      throw new InterruptedIOException();
    }
    for (ZooKeeperProtos.Table.State expectedState: states) {
      if (state == expectedState) {
        allTables.add(tableName);
        break;
      }
    }
  }
  return allTables;
}
 

import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos; //导入依赖的package包/类
@Override
public void handleMetadata(byte[] ownerMetadata) {
  if (!LOG.isDebugEnabled()) {
    return;
  }
  ZooKeeperProtos.TableLock data = fromBytes(ownerMetadata);
  if (data == null) {
    return;
  }
  LOG.debug("Table is locked by " +
      String.format("[tableName=%s:%s, lockOwner=%s, threadId=%s, " +
          "purpose=%s, isShared=%s, createTime=%s]",
          data.getTableName().getNamespace().toStringUtf8(),
          data.getTableName().getQualifier().toStringUtf8(),
          ProtobufUtil.toServerName(data.getLockOwner()), data.getThreadId(),
          data.getPurpose(), data.getIsShared(), data.getCreateTime()));
}
 

import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos; //导入依赖的package包/类
private InterProcessLock createTableLock() {
  String tableLockZNode = ZKUtil.joinZNode(zkWatcher.tableLockZNode,
      tableName.getNameAsString());

  ZooKeeperProtos.TableLock data = ZooKeeperProtos.TableLock.newBuilder()
    .setTableName(ProtobufUtil.toProtoTableName(tableName))
    .setLockOwner(ProtobufUtil.toServerName(serverName))
    .setThreadId(Thread.currentThread().getId())
    .setPurpose(purpose)
    .setIsShared(isShared)
    .setCreateTime(EnvironmentEdgeManager.currentTime()).build();
  byte[] lockMetadata = toBytes(data);

  InterProcessReadWriteLock lock = new ZKInterProcessReadWriteLock(zkWatcher, tableLockZNode,
    METADATA_HANDLER);
  return isShared ? lock.readLock(lockMetadata) : lock.writeLock(lockMetadata);
}
 

import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos; //导入依赖的package包/类
@Override
public void process() {
  LOG.debug("Handling CLOSED event for " + regionInfo.getEncodedName());
  // Check if this table is being disabled or not
  if (this.assignmentManager.getTableStateManager().isTableState(this.regionInfo.getTable(),
      ZooKeeperProtos.Table.State.DISABLED, ZooKeeperProtos.Table.State.DISABLING) ||
      assignmentManager.getReplicasToClose().contains(regionInfo)) {
    assignmentManager.offlineDisabledRegion(regionInfo);
    return;
  }
  // ZK Node is in CLOSED state, assign it.
  assignmentManager.getRegionStates().updateRegionState(
    regionInfo, RegionState.State.CLOSED);
  // This below has to do w/ online enable/disable of a table
  assignmentManager.removeClosedRegion(regionInfo);
  assignmentManager.assign(regionInfo, true);
}
 

import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos; //导入依赖的package包/类
private boolean prepareCreate(final MasterProcedureEnv env) throws IOException {
  final TableName tableName = getTableName();
  if (MetaTableAccessor.tableExists(env.getMasterServices().getConnection(), tableName)) {
    setFailure("master-create-table", new TableExistsException(getTableName()));
    return false;
  }
  // During master initialization, the ZK state could be inconsistent from failed DDL
  // in the past. If we fail here, it would prevent master to start.  We should force
  // setting the system table state regardless the table state.
  boolean skipTableStateCheck =
      !(env.getMasterServices().isInitialized()) && tableName.isSystemTable();
  if (!skipTableStateCheck) {
    TableStateManager tsm = env.getMasterServices().getAssignmentManager().getTableStateManager();
    if (tsm.isTableState(tableName, true, ZooKeeperProtos.Table.State.ENABLING,
        ZooKeeperProtos.Table.State.ENABLED)) {
      LOG.warn("The table " + tableName + " does not exist in meta but has a znode. " +
             "run hbck to fix inconsistencies.");
      setFailure("master-create-table", new TableExistsException(getTableName()));
      return false;
    }
  }
  return true;
}
 

import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos; //导入依赖的package包/类
protected static void assignRegions(final MasterProcedureEnv env,
    final TableName tableName, final List<HRegionInfo> regions)
    throws HBaseException, IOException {
  ProcedureSyncWait.waitRegionServers(env);

  final AssignmentManager assignmentManager = env.getMasterServices().getAssignmentManager();

  // Mark the table as Enabling
  assignmentManager.getTableStateManager().setTableState(tableName,
      ZooKeeperProtos.Table.State.ENABLING);

  // Trigger immediate assignment of the regions in round-robin fashion
  ModifyRegionUtils.assignRegions(assignmentManager, regions);

  // Enable table
  assignmentManager.getTableStateManager()
    .setTableState(tableName, ZooKeeperProtos.Table.State.ENABLED);
}
 

import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos; //导入依赖的package包/类
/**
 * Rollback of table state change in prepareDisable()
 * @param env MasterProcedureEnv
 */
@edu.umd.cs.findbugs.annotations.SuppressWarnings(value="REC_CATCH_EXCEPTION",
    justification="Intended")
private void undoTableStateChange(final MasterProcedureEnv env) {
  if (!skipTableStateCheck) {
    try {
      // If the state was changed, undo it.
      if (env.getMasterServices().getAssignmentManager().getTableStateManager().isTableState(
          tableName, ZooKeeperProtos.Table.State.DISABLING)) {
        EnableTableProcedure.setTableStateToEnabled(env, tableName);
      }
    } catch (Exception e) {
      // Ignore exception.
      LOG.trace(e.getMessage());
    }
  }
}
 

import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos; //导入依赖的package包/类
/**
 * @param bytes Content of a peer znode.
 * @return ClusterKey parsed from the passed bytes.
 * @throws DeserializationException
 */
private static ReplicationPeerConfig parsePeerFrom(final byte[] bytes)
    throws DeserializationException {
  if (ProtobufUtil.isPBMagicPrefix(bytes)) {
    int pblen = ProtobufUtil.lengthOfPBMagic();
    ZooKeeperProtos.ReplicationPeer.Builder builder =
        ZooKeeperProtos.ReplicationPeer.newBuilder();
    ZooKeeperProtos.ReplicationPeer peer;
    try {
      ProtobufUtil.mergeFrom(builder, bytes, pblen, bytes.length - pblen);
      peer = builder.build();
    } catch (IOException e) {
      throw new DeserializationException(e);
    }
    return convert(peer);
  } else {
    if (bytes.length > 0) {
      return new ReplicationPeerConfig().setClusterKey(Bytes.toString(bytes));
    }
    return new ReplicationPeerConfig().setClusterKey("");
  }
}
 

import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos; //导入依赖的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, IOException, CoordinatedStateException {
  Set<TableName> disablingTables =
    tableStateManager.getTablesInStates(ZooKeeperProtos.Table.State.DISABLING);
  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,
          this, tableLockManager, true).prepare().process();
    }
  }
}
 

import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos; //导入依赖的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, IOException, CoordinatedStateException {
  Set<TableName> enablingTables = tableStateManager.
    getTablesInStates(ZooKeeperProtos.Table.State.ENABLING);
  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,
        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.protobuf.generated.ZooKeeperProtos; //导入依赖的package包/类
/**
 * @param bytes - Content of a WAL position znode.
 * @return long - The current WAL position.
 * @throws DeserializationException
 */
public static long parseWALPositionFrom(final byte[] bytes) throws DeserializationException {
  if (bytes == null) {
    throw new DeserializationException("Unable to parse null WAL position.");
  }
  if (ProtobufUtil.isPBMagicPrefix(bytes)) {
    int pblen = ProtobufUtil.lengthOfPBMagic();
    ZooKeeperProtos.ReplicationHLogPosition.Builder builder =
        ZooKeeperProtos.ReplicationHLogPosition.newBuilder();
    ZooKeeperProtos.ReplicationHLogPosition position;
    try {
      ProtobufUtil.mergeFrom(builder, bytes, pblen, bytes.length - pblen);
      position = builder.build();
    } catch (IOException e) {
      throw new DeserializationException(e);
    }
    return position.getPosition();
  } else {
    if (bytes.length > 0) {
      return Bytes.toLong(bytes);
    }
    return 0;
  }
}
 

import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos; //导入依赖的package包/类
private void onRegionOpen(final HRegionInfo hri, final ServerName sn, long openSeqNum) {
  regionOnline(hri, sn, openSeqNum);
  if (useZKForAssignment) {
    try {
      // Delete the ZNode if exists
      ZKAssign.deleteNodeFailSilent(watcher, hri);
    } catch (KeeperException ke) {
      server.abort("Unexpected ZK exception deleting node " + hri, ke);
    }
  }

  // reset the count, if any
  failedOpenTracker.remove(hri.getEncodedName());
  if (getTableStateManager().isTableState(hri.getTable(),
      ZooKeeperProtos.Table.State.DISABLED, ZooKeeperProtos.Table.State.DISABLING)) {
    invokeUnAssign(hri);
  }
}
 

import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos; //导入依赖的package包/类
private static void appendPeerState(ZooKeeperWatcher zkw, String znodeToProcess,
    StringBuilder sb) throws KeeperException, InvalidProtocolBufferException {
  String peerState = zkw.getConfiguration().get("zookeeper.znode.replication.peers.state",
    "peer-state");
  int pblen = ProtobufUtil.lengthOfPBMagic();
  for (String child : ZKUtil.listChildrenNoWatch(zkw, znodeToProcess)) {
    if (!child.equals(peerState)) continue;
    String peerStateZnode = ZKUtil.joinZNode(znodeToProcess, child);
    sb.append("\n").append(peerStateZnode).append(": ");
    byte[] peerStateData;
    try {
      peerStateData = ZKUtil.getData(zkw, peerStateZnode);
      ZooKeeperProtos.ReplicationState.Builder builder =
          ZooKeeperProtos.ReplicationState.newBuilder();
      ProtobufUtil.mergeFrom(builder, peerStateData, pblen, peerStateData.length - pblen);
      sb.append(builder.getState().name());
    } catch (IOException ipbe) {
      LOG.warn("Got Exception while parsing peer: " + znodeToProcess, ipbe);
    } catch (InterruptedException e) {
      zkw.interruptedException(e);
      return;
    }
  }
}
 

import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos; //导入依赖的package包/类
private void checkAndMigrateTableStatesToPB(ZooKeeperWatcher zkw) throws KeeperException,
    InterruptedException {
  List<String> tables = ZKUtil.listChildrenNoWatch(zkw, zkw.tableZNode);
  if (tables == null) {
    LOG.info("No table present to migrate table state to PB. returning..");
    return;
  }
  for (String table : tables) {
    String znode = ZKUtil.joinZNode(zkw.tableZNode, table);
    // Delete -ROOT- table state znode since its no longer present in 0.95.0
    // onwards.
    if (table.equals("-ROOT-") || table.equals(".META.")) {
      ZKUtil.deleteNode(zkw, znode);
      continue;
    }
    byte[] data = ZKUtil.getData(zkw, znode);
    if (ProtobufUtil.isPBMagicPrefix(data)) continue;
    ZooKeeperProtos.Table.Builder builder = ZooKeeperProtos.Table.newBuilder();
    builder.setState(ZooKeeperProtos.Table.State.valueOf(Bytes.toString(data)));
    data = ProtobufUtil.prependPBMagic(builder.build().toByteArray());
    ZKUtil.setData(zkw, znode, data);
  }
}
 

import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos; //导入依赖的package包/类
/**
 * Gets a list of tables that are set as one of the passing in states in zookeeper.
 * @param zkw ZooKeeperWatcher instance to use
 * @param states the list of states that a table could be in
 * @return Set of tables in one of the states, empty Set if none
 * @throws KeeperException
 * @throws InterruptedException
 */
private static Set<TableName> getTablesInStates(
  ZooKeeperWatcher zkw,
  ZooKeeperProtos.Table.State... states)
    throws KeeperException, InterruptedException {
  Set<TableName> tableNameSet = new HashSet<TableName>();
  List<String> children = ZKUtil.listChildrenNoWatch(zkw, zkw.tableZNode);
  TableName tableName;
  ZooKeeperProtos.Table.State tableState;
  for (String child: children) {
    tableName = TableName.valueOf(child);
    tableState = getTableState(zkw, tableName);
    for (ZooKeeperProtos.Table.State state : states) {
       if (tableState == state) {
         tableNameSet.add(tableName);
         break;
       }
    }
  }
  return tableNameSet;
}
 

import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos; //导入依赖的package包/类
/**
 * Sets znodes used in 0.94 version. Only table and replication znodes will be upgraded to PB,
 * others would be deleted.
 * @throws KeeperException
 */
private static void setUp94Znodes() throws IOException, KeeperException {
  // add some old znodes, which would be deleted after upgrade.
  String rootRegionServerZnode = ZKUtil.joinZNode(zkw.baseZNode, "root-region-server");
  ZKUtil.createWithParents(zkw, rootRegionServerZnode);
  ZKUtil.createWithParents(zkw, zkw.backupMasterAddressesZNode);
  // add table znode, data of its children would be protobuffized
  tableAZnode = ZKUtil.joinZNode(zkw.tableZNode, "a");
  ZKUtil.createWithParents(zkw, tableAZnode,
    Bytes.toBytes(ZooKeeperProtos.Table.State.ENABLED.toString()));
  // add replication znodes, data of its children would be protobuffized
  String replicationZnode = ZKUtil.joinZNode(zkw.baseZNode, "replication");
  replicationPeerZnode = ZKUtil.joinZNode(replicationZnode, "peers");
  peer1Znode = ZKUtil.joinZNode(replicationPeerZnode, "1");
  peer1 = ReplicationPeer.newBuilder().setClusterkey("abc:123:/hbase").build();
  ZKUtil.createWithParents(zkw, peer1Znode, Bytes.toBytes(peer1.getClusterkey()));
}
 

import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos; //导入依赖的package包/类
@Test
public void testMasterRestartAfterNameSpaceEnablingNodeIsCreated() throws Exception {
  // Step 1: start mini zk cluster.
  MiniZooKeeperCluster zkCluster;
  zkCluster = TEST_UTIL.startMiniZKCluster();
  // Step 2: add an orphaned system table ZNODE
  TableName tableName = TableName.valueOf("hbase:namespace");
  ZooKeeperWatcher zkw = TEST_UTIL.getZooKeeperWatcher();
  String znode = ZKUtil.joinZNode(zkw.tableZNode, tableName.getNameAsString());
  ZooKeeperProtos.Table.Builder builder = ZooKeeperProtos.Table.newBuilder();
  builder.setState(ZooKeeperProtos.Table.State.ENABLED);
  byte [] data = ProtobufUtil.prependPBMagic(builder.build().toByteArray());
  ZKUtil.createSetData(zkw, znode, data);
  LOG.info("Create an orphaned Znode " + znode + " with data " + data);
  // Step 3: link the zk cluster to hbase cluster
  TEST_UTIL.setZkCluster(zkCluster);
  // Step 4: start hbase cluster and expect master to start successfully.
  TEST_UTIL.startMiniCluster();
  assertTrue(TEST_UTIL.getHBaseCluster().getLiveMasterThreads().size() == 1);
}
 

import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos; //导入依赖的package包/类
/**
 * Update the state znode of a peer cluster.
 * @param id
 * @param state
 */
private void changePeerState(String id, ZooKeeperProtos.ReplicationState.State state)
    throws ReplicationException {
  try {
    if (!peerExists(id)) {
      throw new IllegalArgumentException("Cannot enable/disable peer because id=" + id
          + " does not exist.");
    }
    String peerStateZNode = getPeerStateNode(id);
    byte[] stateBytes =
        (state == ZooKeeperProtos.ReplicationState.State.ENABLED) ? ENABLED_ZNODE_BYTES
            : DISABLED_ZNODE_BYTES;
    if (ZKUtil.checkExists(this.zookeeper, peerStateZNode) != -1) {
      ZKUtil.setData(this.zookeeper, peerStateZNode, stateBytes);
    } else {
      ZKUtil.createAndWatch(this.zookeeper, peerStateZNode, stateBytes);
    }
    LOG.info("Peer with id= " + id + " is now " + state.name());
  } catch (KeeperException e) {
    throw new ReplicationException("Unable to change state of the peer with id=" + id, e);
  }
}