Java 类org.apache.hadoop.hbase.catalog.CatalogTracker 实例源码

/**
 * If the input is a region name, it is returned as is. If it's an
 * encoded region name, the corresponding region is found from meta
 * and its region name is returned. If we can't find any region in
 * meta matching the input as either region name or encoded region
 * name, the input is returned as is. We don't throw unknown
 * region exception.
 */
private byte[] getRegionName(
    final byte[] regionNameOrEncodedRegionName) throws IOException {
  if (Bytes.equals(regionNameOrEncodedRegionName,
      HRegionInfo.FIRST_META_REGIONINFO.getRegionName())
        || Bytes.equals(regionNameOrEncodedRegionName,
          HRegionInfo.FIRST_META_REGIONINFO.getEncodedNameAsBytes())) {
    return HRegionInfo.FIRST_META_REGIONINFO.getRegionName();
  }
  CatalogTracker ct = getCatalogTracker();
  byte[] tmp = regionNameOrEncodedRegionName;
  try {
    Pair<HRegionInfo, ServerName> regionServerPair
      = getRegion(regionNameOrEncodedRegionName, ct);
    if (regionServerPair != null && regionServerPair.getFirst() != null) {
      tmp = regionServerPair.getFirst().getRegionName();
    }
  } finally {
    cleanupCatalogTracker(ct);
  }
  return tmp;
}

/**
 * Check individual daughter is up in .META.; fixup if its not.
 * @param result The contents of the parent row in .META.
 * @param qualifier Which daughter to check for.
 * @return 1 if the daughter is missing and fixed. Otherwise 0
 * @throws IOException
 */
static int fixupDaughter(final Result result, final byte [] qualifier,
    final AssignmentManager assignmentManager,
    final CatalogTracker catalogTracker)
throws IOException {
  HRegionInfo daughter =
    MetaReader.parseHRegionInfoFromCatalogResult(result, qualifier);
  if (daughter == null) return 0;
  if (isDaughterMissing(catalogTracker, daughter)) {
    LOG.info("Fixup; missing daughter " + daughter.getRegionNameAsString());
    MetaEditor.addDaughter(catalogTracker, daughter, null);

    // TODO: Log WARN if the regiondir does not exist in the fs.  If its not
    // there then something wonky about the split -- things will keep going
    // but could be missing references to parent region.

    // And assign it.
    assignmentManager.assign(daughter, true);
    return 1;
  } else {
    LOG.debug("Daughter " + daughter.getRegionNameAsString() + " present");
  }
  return 0;
}

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

@Test
public void testGetRegion() throws Exception {
  final String name = "testGetRegion";
  LOG.info("Started " + name);
  final byte [] nameBytes = Bytes.toBytes(name);
  HTable t = TEST_UTIL.createTable(nameBytes, HConstants.CATALOG_FAMILY);
  TEST_UTIL.createMultiRegions(t, HConstants.CATALOG_FAMILY);
  CatalogTracker ct = new CatalogTracker(TEST_UTIL.getConfiguration());
  ct.start();
  try {
    HRegionLocation regionLocation = t.getRegionLocation("mmm");
    HRegionInfo region = regionLocation.getRegionInfo();
    byte[] regionName = region.getRegionName();
    Pair<HRegionInfo, ServerName> pair = admin.getRegion(regionName, ct);
    assertTrue(Bytes.equals(regionName, pair.getFirst().getRegionName()));
    pair = admin.getRegion(region.getEncodedNameAsBytes(), ct);
    assertTrue(Bytes.equals(regionName, pair.getFirst().getRegionName()));
  } finally {
    ct.stop();
  }
}

MockServer(final HBaseTestingUtility htu)
throws NotAllMetaRegionsOnlineException, IOException, InterruptedException {
  this.c = htu.getConfiguration();
  // Mock an HConnection and a HRegionInterface implementation.  Have the
  // HConnection return the HRI.  Have the HRI return a few mocked up responses
  // to make our test work.
  this.connection =
    HConnectionTestingUtility.getMockedConnectionAndDecorate(this.c,
      Mockito.mock(HRegionInterface.class),
      new ServerName("example.org,12345,6789"),
      HRegionInfo.FIRST_META_REGIONINFO);
  // Set hbase.rootdir into test dir.
  FileSystem fs = FileSystem.get(this.c);
  Path rootdir = fs.makeQualified(new Path(this.c.get(HConstants.HBASE_DIR)));
  this.c.set(HConstants.HBASE_DIR, rootdir.toString());
  this.ct = Mockito.mock(CatalogTracker.class);
  HRegionInterface hri = Mockito.mock(HRegionInterface.class);
  Mockito.when(this.ct.getConnection()).thenReturn(this.connection);
  Mockito.when(ct.waitForMetaServerConnection(Mockito.anyLong())).thenReturn(hri);
}

/**
 * Run a simple server shutdown handler.
 * @throws KeeperException
 * @throws IOException
 */
@Test
public void testShutdownHandler() throws KeeperException, IOException {
  // Create and startup an executor.  This is used by AssignmentManager
  // handling zk callbacks.
  ExecutorService executor = startupMasterExecutor("testShutdownHandler");

  // We need a mocked catalog tracker.
  CatalogTracker ct = Mockito.mock(CatalogTracker.class);
  LoadBalancer balancer = LoadBalancerFactory.getLoadBalancer(server
      .getConfiguration());
  // Create an AM.
  AssignmentManager am =
    new AssignmentManager(this.server, this.serverManager, ct, balancer, executor);
  try {
    processServerShutdownHandler(ct, am, false, null);
  } finally {
    executor.shutdown();
    am.shutdown();
    // Clean up all znodes
    ZKAssign.deleteAllNodes(this.watcher);
  }
}

private void offlineParentInMetaAndputMetaEntries(CatalogTracker catalogTracker,
    HRegionInfo parent, HRegionInfo splitA, HRegionInfo splitB,
    ServerName serverName, List<Mutation> metaEntries) throws IOException {
  List<Mutation> mutations = metaEntries;
  HRegionInfo copyOfParent = new HRegionInfo(parent);
  copyOfParent.setOffline(true);
  copyOfParent.setSplit(true);

  //Put for parent
  Put putParent = MetaEditor.makePutFromRegionInfo(copyOfParent);
  MetaEditor.addDaughtersToPut(putParent, splitA, splitB);
  mutations.add(putParent);

  //Puts for daughters
  Put putA = MetaEditor.makePutFromRegionInfo(splitA);
  Put putB = MetaEditor.makePutFromRegionInfo(splitB);

  addLocation(putA, serverName, 1); //these are new regions, openSeqNum = 1 is fine.
  addLocation(putB, serverName, 1);
  mutations.add(putA);
  mutations.add(putB);
  MetaEditor.mutateMetaTable(catalogTracker, mutations);
}

@Test (timeout=300000)
public void testGetRegion() throws Exception {
  final String name = "testGetRegion";
  LOG.info("Started " + name);
  final byte [] nameBytes = Bytes.toBytes(name);
  HTable t = TEST_UTIL.createTable(nameBytes, HConstants.CATALOG_FAMILY);
  TEST_UTIL.createMultiRegions(t, HConstants.CATALOG_FAMILY);
  CatalogTracker ct = new CatalogTracker(TEST_UTIL.getConfiguration());
  ct.start();
  try {
    HRegionLocation regionLocation = t.getRegionLocation("mmm");
    HRegionInfo region = regionLocation.getRegionInfo();
    byte[] regionName = region.getRegionName();
    Pair<HRegionInfo, ServerName> pair = admin.getRegion(regionName, ct);
    assertTrue(Bytes.equals(regionName, pair.getFirst().getRegionName()));
    pair = admin.getRegion(region.getEncodedNameAsBytes(), ct);
    assertTrue(Bytes.equals(regionName, pair.getFirst().getRegionName()));
  } finally {
    ct.stop();
  }
}

/**
 * Run a simple server shutdown handler.
 * @throws KeeperException
 * @throws IOException
 */
@Test
public void testShutdownHandler()
    throws KeeperException, IOException, ServiceException {
  // Create and startup an executor.  This is used by AssignmentManager
  // handling zk callbacks.
  ExecutorService executor = startupMasterExecutor("testShutdownHandler");

  // We need a mocked catalog tracker.
  CatalogTracker ct = Mockito.mock(CatalogTracker.class);
  // Create an AM.
  AssignmentManagerWithExtrasForTesting am = setUpMockedAssignmentManager(
      this.server, this.serverManager);
  try {
    processServerShutdownHandler(ct, am, false);
  } finally {
    executor.shutdown();
    am.shutdown();
    // Clean up all znodes
    ZKAssign.deleteAllNodes(this.watcher);
  }
}

/**
 * Initialize the restore helper, based on the snapshot and table information provided.
 */
private RestoreSnapshotHelper getRestoreHelper(final Path rootDir, final Path snapshotDir,
    final String sourceTableName, final HTableDescriptor htdClone) throws IOException {
  CatalogTracker catalogTracker = Mockito.mock(CatalogTracker.class);
  HTableDescriptor tableDescriptor = Mockito.mock(HTableDescriptor.class);
  ForeignExceptionDispatcher monitor = Mockito.mock(ForeignExceptionDispatcher.class);
  MonitoredTask status = Mockito.mock(MonitoredTask.class);

  SnapshotDescription sd = SnapshotDescription.newBuilder()
    .setName("snapshot")
    .setTable(sourceTableName)
    .build();

  return new RestoreSnapshotHelper(conf, fs, sd, snapshotDir,
    htdClone, rootDir, monitor, status);
}

/**
 * If a table is deleted, we should not be able to balance it anymore.
 * Otherwise, the region will be brought back.
 * @throws Exception
 */
@Test
public void testBalanceRegionOfDeletedTable() throws Exception {
  CatalogTracker ct = Mockito.mock(CatalogTracker.class);
  AssignmentManager am = new AssignmentManager(this.server, this.serverManager,
    ct, balancer, null, null, master.getTableLockManager());
  RegionStates regionStates = am.getRegionStates();
  HRegionInfo hri = REGIONINFO;
  regionStates.createRegionState(hri);
  assertFalse(regionStates.isRegionInTransition(hri));
  RegionPlan plan = new RegionPlan(hri, SERVERNAME_A, SERVERNAME_B);
  // Fake table is deleted
  regionStates.tableDeleted(hri.getTable());
  am.balance(plan);
  assertFalse("The region should not in transition",
    regionStates.isRegionInTransition(hri));
}

/**
 * @param tableName Table to check.
 * @return True if table exists already.
 * @throws IOException
 */
public boolean tableExists(final String tableName) throws IOException {
  boolean b = false;
  CatalogTracker ct = getCatalogTracker();
  try {
    b = MetaReader.tableExists(ct, tableName);
  } finally {
    cleanupCatalogTracker(ct);
  }
  return b;
}

/**
 * Compact a table or an individual region. Asynchronous operation.
 * @param tableNameOrRegionName table or region to compact
 * @param columnFamily column family within a table or region
 * @param major True if we are to do a major compaction.
 * @throws IOException if a remote or network exception occurs
 * @throws InterruptedException
 */
private void compact(final byte[] tableNameOrRegionName, final byte[] columnFamily,
    final boolean major) throws IOException, InterruptedException {
  CatalogTracker ct = getCatalogTracker();
  try {
    Pair<HRegionInfo, ServerName> regionServerPair = getRegion(tableNameOrRegionName, ct);
    if (regionServerPair != null) {
      if (regionServerPair.getSecond() == null) {
        throw new NoServerForRegionException(Bytes.toStringBinary(tableNameOrRegionName));
      } else {
        compact(regionServerPair.getSecond(), regionServerPair.getFirst(), major, columnFamily);
      }
    } else {
      final String tableName = tableNameString(tableNameOrRegionName, ct);
      List<Pair<HRegionInfo, ServerName>> pairs = MetaReader.getTableRegionsAndLocations(ct,
        tableName);
      for (Pair<HRegionInfo, ServerName> pair : pairs) {
        if (pair.getFirst().isOffline()) continue;
        if (pair.getSecond() == null) continue;
        try {
          compact(pair.getSecond(), pair.getFirst(), major, columnFamily);
        } catch (NotServingRegionException e) {
          if (LOG.isDebugEnabled()) {
            LOG.debug("Trying to" + (major ? " major" : "") + " compact " + pair.getFirst()
                + ": " + StringUtils.stringifyException(e));
          }
        }
      }
    }
  } finally {
    cleanupCatalogTracker(ct);
  }
}

/**
 * Split a table or an individual region. Asynchronous operation.
 * @param tableNameOrRegionName table to region to split
 * @param splitPoint the explicit position to split on
 * @throws IOException if a remote or network exception occurs
 * @throws InterruptedException interrupt exception occurred
 */
public void split(final byte[] tableNameOrRegionName, final byte[] splitPoint)
    throws IOException, InterruptedException {
  CatalogTracker ct = getCatalogTracker();
  try {
    Pair<HRegionInfo, ServerName> regionServerPair = getRegion(tableNameOrRegionName, ct);
    if (regionServerPair != null) {
      if (regionServerPair.getSecond() == null) {
        throw new NoServerForRegionException(Bytes.toStringBinary(tableNameOrRegionName));
      } else {
        split(regionServerPair.getSecond(), regionServerPair.getFirst(), splitPoint);
      }
    } else {
      final String tableName = tableNameString(tableNameOrRegionName, ct);
      List<Pair<HRegionInfo, ServerName>> pairs = MetaReader.getTableRegionsAndLocations(ct,
        tableName);
      for (Pair<HRegionInfo, ServerName> pair : pairs) {
        // May not be a server for a particular row
        if (pair.getSecond() == null) continue;
        HRegionInfo r = pair.getFirst();
        // check for parents
        if (r.isSplitParent()) continue;
        // if a split point given, only split that particular region
        if (splitPoint != null && !r.containsRow(splitPoint)) continue;
        // call out to region server to do split now
        split(pair.getSecond(), pair.getFirst(), splitPoint);
      }
    }
  } finally {
    cleanupCatalogTracker(ct);
  }
}

/**
 * @param tableNameOrRegionName Name of a table or name of a region.
 * @param ct A {@link CatalogTracker} instance (caller of this method usually has one).
 * @return a pair of HRegionInfo and ServerName if <code>tableNameOrRegionName</code> is a
 *         verified region name (we call {@link MetaReader#getRegion(CatalogTracker, byte[])} else
 *         null. Throw an exception if <code>tableNameOrRegionName</code> is null.
 * @throws IOException
 */
Pair<HRegionInfo, ServerName> getRegion(final byte[] tableNameOrRegionName,
    final CatalogTracker ct) throws IOException {
  if (tableNameOrRegionName == null) {
    throw new IllegalArgumentException("Pass a table name or region name");
  }
  Pair<HRegionInfo, ServerName> pair = MetaReader.getRegion(ct, tableNameOrRegionName);
  if (pair == null) {
    final AtomicReference<Pair<HRegionInfo, ServerName>> result = new AtomicReference<Pair<HRegionInfo, ServerName>>(
        null);
    final String encodedName = Bytes.toString(tableNameOrRegionName);
    MetaScannerVisitor visitor = new MetaScannerVisitorBase() {
      @Override
      public boolean processRow(Result data) throws IOException {
        if (data == null || data.size() <= 0) {
          return true;
        }
        HRegionInfo info = MetaReader.parseHRegionInfoFromCatalogResult(data,
          HConstants.REGIONINFO_QUALIFIER);
        if (info == null) {
          LOG.warn("No serialized HRegionInfo in " + data);
          return true;
        }
        if (!encodedName.equals(info.getEncodedName())) return true;
        ServerName sn = MetaReader.getServerNameFromCatalogResult(data);
        result.set(new Pair<HRegionInfo, ServerName>(info, sn));
        return false; // found the region, stop
      }
    };

    MetaScanner.metaScan(conf, connection, visitor, null);
    pair = result.get();
  }
  return pair;
}

/**
 * get the regions of a given table.
 * @param tableName the name of the table
 * @return Ordered list of {@link HRegionInfo}.
 * @throws IOException
 */
public List<HRegionInfo> getTableRegions(final byte[] tableName) throws IOException {
  CatalogTracker ct = getCatalogTracker();
  List<HRegionInfo> Regions = null;
  try {
    Regions = MetaReader.getTableRegions(ct, tableName, true);
  } finally {
    cleanupCatalogTracker(ct);
  }
  return Regions;
}

/**
 * Process a dead region from a dead RS. Checks if the region is disabled or
 * disabling or if the region has a partially completed split.
 * @param hri
 * @param result
 * @param assignmentManager
 * @param catalogTracker
 * @return Returns true if specified region should be assigned, false if not.
 * @throws IOException
 */
public static boolean processDeadRegion(HRegionInfo hri, Result result,
    AssignmentManager assignmentManager, CatalogTracker catalogTracker)
throws IOException {
  boolean tablePresent = assignmentManager.getZKTable().isTablePresent(
      hri.getTableNameAsString());
  if (!tablePresent) {
    LOG.info("The table " + hri.getTableNameAsString()
        + " was deleted.  Hence not proceeding.");
    return false;
  }
  // If table is not disabled but the region is offlined,
  boolean disabled = assignmentManager.getZKTable().isDisabledTable(
      hri.getTableNameAsString());
  if (disabled){
    LOG.info("The table " + hri.getTableNameAsString()
        + " was disabled.  Hence not proceeding.");
    return false;
  }
  if (hri.isOffline() && hri.isSplit()) {
    LOG.debug("Offlined and split region " + hri.getRegionNameAsString() +
      "; checking daughter presence");
    if (MetaReader.getRegion(catalogTracker, hri.getRegionName()) == null) {
      return false;
    }
    fixupDaughters(result, assignmentManager, catalogTracker);
    return false;
  }
  boolean disabling = assignmentManager.getZKTable().isDisablingTable(
      hri.getTableNameAsString());
  if (disabling) {
    LOG.info("The table " + hri.getTableNameAsString()
        + " is disabled.  Hence not assigning region" + hri.getEncodedName());
    return false;
  }
  return true;
}

/**
 * Check that daughter regions are up in .META. and if not, add them.
 * @param hris All regions for this server in meta.
 * @param result The contents of the parent row in .META.
 * @return the number of daughters missing and fixed
 * @throws IOException
 */
public static int fixupDaughters(final Result result,
    final AssignmentManager assignmentManager,
    final CatalogTracker catalogTracker)
throws IOException {
  int fixedA = fixupDaughter(result, HConstants.SPLITA_QUALIFIER,
    assignmentManager, catalogTracker);
  int fixedB = fixupDaughter(result, HConstants.SPLITB_QUALIFIER,
    assignmentManager, catalogTracker);
  return fixedA + fixedB;
}

/**
 * Look for presence of the daughter OR of a split of the daughter in .META.
 * Daughter could have been split over on regionserver before a run of the
 * catalogJanitor had chance to clear reference from parent.
 * @param daughter Daughter region to search for.
 * @throws IOException 
 */
private static boolean isDaughterMissing(final CatalogTracker catalogTracker,
    final HRegionInfo daughter) throws IOException {
  FindDaughterVisitor visitor = new FindDaughterVisitor(daughter);
  // Start the scan at what should be the daughter's row in the .META.
  // We will either 1., find the daughter or some derivative split of the
  // daughter (will have same table name and start row at least but will sort
  // after because has larger regionid -- the regionid is timestamp of region
  // creation), OR, we will not find anything with same table name and start
  // row.  If the latter, then assume daughter missing and do fixup.
  byte [] startrow = daughter.getRegionName();
  MetaReader.fullScan(catalogTracker, visitor, startrow);
  return !visitor.foundDaughter();
}

public DisableTableHandler(Server server, byte [] tableName,
    CatalogTracker catalogTracker, AssignmentManager assignmentManager,
    boolean skipTableStateCheck)
throws TableNotFoundException, TableNotEnabledException, IOException {
  super(server, EventType.C_M_DISABLE_TABLE);
  this.tableName = tableName;
  this.tableNameStr = Bytes.toString(this.tableName);
  this.assignmentManager = assignmentManager;
  // Check if table exists
  // TODO: do we want to keep this in-memory as well?  i guess this is
  //       part of old master rewrite, schema to zk to check for table
  //       existence and such
  if (!MetaReader.tableExists(catalogTracker, this.tableNameStr)) {
    throw new TableNotFoundException(this.tableNameStr);
  }

  // There could be multiple client requests trying to disable or enable
  // the table at the same time. Ensure only the first request is honored
  // After that, no other requests can be accepted until the table reaches
  // DISABLED or ENABLED.
  if (!skipTableStateCheck)
  {
    try {
      if (!this.assignmentManager.getZKTable().checkEnabledAndSetDisablingTable
        (this.tableNameStr)) {
        LOG.info("Table " + tableNameStr + " isn't enabled; skipping disable");
        throw new TableNotEnabledException(this.tableNameStr);
      }
    } catch (KeeperException e) {
      throw new IOException("Unable to ensure that the table will be" +
        " disabling because of a ZooKeeper issue", e);
    }
  }
}

/**
 * Initialize all ZK based system trackers.
 * @throws IOException
 * @throws InterruptedException
 */
private void initializeZKBasedSystemTrackers() throws IOException,
    InterruptedException, KeeperException {
  this.catalogTracker = new CatalogTracker(this.zooKeeper, this.conf, this);
  this.catalogTracker.start();

  this.balancer = LoadBalancerFactory.getLoadBalancer(conf);
  this.assignmentManager = new AssignmentManager(this, serverManager,
      this.catalogTracker, this.balancer, this.executorService);
  zooKeeper.registerListenerFirst(assignmentManager);

  this.regionServerTracker = new RegionServerTracker(zooKeeper, this,
      this.serverManager);
  this.regionServerTracker.start();

  this.drainingServerTracker = new DrainingServerTracker(zooKeeper, this,
    this.serverManager);
  this.drainingServerTracker.start();

  // Set the cluster as up.  If new RSs, they'll be waiting on this before
  // going ahead with their startup.
  boolean wasUp = this.clusterStatusTracker.isClusterUp();
  if (!wasUp) this.clusterStatusTracker.setClusterUp();

  LOG.info("Server active/primary master; " + this.serverName +
      ", sessionid=0x" +
      Long.toHexString(this.zooKeeper.getRecoverableZooKeeper().getSessionId()) +
      ", cluster-up flag was=" + wasUp);

  // create the snapshot manager
  this.snapshotManager = new SnapshotManager(this, this.metrics);
}

/**
 * Bring up connection to zk ensemble and then wait until a master for this cluster and then after
 * that, wait until cluster 'up' flag has been set. This is the order in which master does things.
 * Finally put up a catalog tracker.
 * @throws IOException
 * @throws InterruptedException
 */
private void initializeZooKeeper() throws IOException, InterruptedException {
  // Open connection to zookeeper and set primary watcher
  this.zooKeeper = new ZooKeeperWatcher(conf, REGIONSERVER + ":" + this.isa.getPort(), this);

  // Create the master address manager, register with zk, and start it. Then
  // block until a master is available. No point in starting up if no master
  // running.
  this.masterAddressManager = new MasterAddressTracker(this.zooKeeper, this);
  this.masterAddressManager.start();
  blockAndCheckIfStopped(this.masterAddressManager);

  // Wait on cluster being up. Master will set this flag up in zookeeper
  // when ready.
  this.clusterStatusTracker = new ClusterStatusTracker(this.zooKeeper, this);
  this.clusterStatusTracker.start();
  blockAndCheckIfStopped(this.clusterStatusTracker);

  // Create the catalog tracker and start it;
  this.catalogTracker = new CatalogTracker(this.zooKeeper, this.conf, this);
  catalogTracker.start();

  // watch for snapshots
  try {
    this.snapshotManager = new RegionServerSnapshotManager(this);
  } catch (KeeperException e) {
    this.abort("Failed to reach zk cluster when creating snapshot handler.");
  }
}

@Override
public void postOpenDeployTasks(final HRegion r, final CatalogTracker ct, final boolean daughter)
    throws KeeperException, IOException {
  checkOpen();
  LOG.info("Post open deploy tasks for region=" + r.getRegionNameAsString() + ", daughter="
      + daughter);
  // Do checks to see if we need to compact (references or too many files)
  for (Store s : r.getStores().values()) {
    if (s.hasReferences() || s.needsCompaction()) {
      getCompactionRequester().requestCompaction(r, s, "Opening Region", null);
    }
  }
  // Update ZK, ROOT or META
  if (r.getRegionInfo().isRootRegion()) {
    RootLocationEditor.setRootLocation(getZooKeeper(), this.serverNameFromMasterPOV);
  } else if (r.getRegionInfo().isMetaRegion()) {
    MetaEditor.updateMetaLocation(ct, r.getRegionInfo(), this.serverNameFromMasterPOV);
  } else {
    if (daughter) {
      // If daughter of a split, update whole row, not just location.
      MetaEditor.addDaughter(ct, r.getRegionInfo(), this.serverNameFromMasterPOV);
    } else {
      MetaEditor.updateRegionLocation(ct, r.getRegionInfo(), this.serverNameFromMasterPOV);
    }
  }
  LOG.info("Done with post open deploy task for region=" + r.getRegionNameAsString()
      + ", daughter=" + daughter);

}

@Test
public void testMetaMigration() throws Exception {
  LOG.info("Starting testMetaWithLegacyHRI");
  final byte [] FAMILY = Bytes.toBytes("family");
  HTableDescriptor htd = new HTableDescriptor("testMetaMigration");
  HColumnDescriptor hcd = new HColumnDescriptor(FAMILY);
    htd.addFamily(hcd);
  Configuration conf = TEST_UTIL.getConfiguration();
  createMultiRegionsWithLegacyHRI(conf, htd, FAMILY,
      new byte[][]{
          HConstants.EMPTY_START_ROW,
          Bytes.toBytes("region_a"),
          Bytes.toBytes("region_b")});
  CatalogTracker ct =
    TEST_UTIL.getMiniHBaseCluster().getMaster().getCatalogTracker();
  // Erase the current version of root meta for this test.
  undoVersionInMeta();
  MetaReader.fullScanMetaAndPrint(ct);
  LOG.info("Meta Print completed.testUpdatesOnMetaWithLegacyHRI");

  Set<HTableDescriptor> htds =
    MetaMigrationRemovingHTD.updateMetaWithNewRegionInfo(
      TEST_UTIL.getHBaseCluster().getMaster());
  MetaReader.fullScanMetaAndPrint(ct);
  // Should be one entry only and it should be for the table we just added.
  assertEquals(1, htds.size());
  assertTrue(htds.contains(htd));
  // Assert that the flag in ROOT is updated to reflect the correct status
  boolean metaUpdated =
    MetaMigrationRemovingHTD.isMetaHRIUpdated(
      TEST_UTIL.getMiniHBaseCluster().getMaster());
  assertEquals(true, metaUpdated);
}

/**
 * This test assumes a master crash/failure during the meta migration process
 * and attempts to continue the meta migration process when a new master takes over.
 * When a master dies during the meta migration we will have some rows of
 * META.CatalogFamily updated with new HRI, (i.e HRI with out HTD) and some
 * still hanging with legacy HRI. (i.e HRI with HTD). When the backup master/ or
 * fresh start of master attempts the migration it will encouter some rows of META
 * already updated with new HRI and some still legacy. This test will simulate this
 * scenario and validates that the migration process can safely skip the updated
 * rows and migrate any pending rows at startup.
 * @throws Exception
 */
@Test
public void testMasterCrashDuringMetaMigration() throws Exception {
  final byte[] FAMILY = Bytes.toBytes("family");
  HTableDescriptor htd = new HTableDescriptor("testMasterCrashDuringMetaMigration");
  HColumnDescriptor hcd = new HColumnDescriptor(FAMILY);
    htd.addFamily(hcd);
  Configuration conf = TEST_UTIL.getConfiguration();
  // Create 10 New regions.
  createMultiRegionsWithNewHRI(conf, htd, FAMILY, 10);
  // Create 10 Legacy regions.
  createMultiRegionsWithLegacyHRI(conf, htd, FAMILY, 10);
  CatalogTracker ct =
    TEST_UTIL.getMiniHBaseCluster().getMaster().getCatalogTracker();
  // Erase the current version of root meta for this test.
  undoVersionInMeta();
  MetaMigrationRemovingHTD.updateRootWithMetaMigrationStatus(ct);
  //MetaReader.fullScanMetaAndPrint(ct);
  LOG.info("Meta Print completed.testUpdatesOnMetaWithLegacyHRI");

  Set<HTableDescriptor> htds =
    MetaMigrationRemovingHTD.updateMetaWithNewRegionInfo(
      TEST_UTIL.getHBaseCluster().getMaster());
  assertEquals(1, htds.size());
  assertTrue(htds.contains(htd));
  // Assert that the flag in ROOT is updated to reflect the correct status
  boolean metaUpdated = MetaMigrationRemovingHTD.
    isMetaHRIUpdated(TEST_UTIL.getMiniHBaseCluster().getMaster());
  assertEquals(true, metaUpdated);
  LOG.info("END testMetaWithLegacyHRI");
}

/**
 * @param tableNameOrRegionName Name of a table or name of a region.
 * @param ct A {@link CatalogTracker} instance (caller of this method usually has one).
 * @return a pair of HRegionInfo and ServerName if <code>tableNameOrRegionName</code> is
 *  a verified region name (we call {@link  MetaReader#getRegion( CatalogTracker, byte[])}
 *  else null.
 * Throw an exception if <code>tableNameOrRegionName</code> is null.
 * @throws IOException
 */
Pair<HRegionInfo, ServerName> getRegion(final byte[] tableNameOrRegionName,
    final CatalogTracker ct) throws IOException {
  if (tableNameOrRegionName == null) {
    throw new IllegalArgumentException("Pass a table name or region name");
  }
  Pair<HRegionInfo, ServerName> pair = MetaReader.getRegion(ct, tableNameOrRegionName);
  if (pair == null) {
    final AtomicReference<Pair<HRegionInfo, ServerName>> result =
      new AtomicReference<Pair<HRegionInfo, ServerName>>(null);
    final String encodedName = Bytes.toString(tableNameOrRegionName);
    MetaScannerVisitor visitor = new MetaScannerVisitorBase() {
      @Override
      public boolean processRow(Result data) throws IOException {
        HRegionInfo info = HRegionInfo.getHRegionInfo(data);
        if (info == null) {
          LOG.warn("No serialized HRegionInfo in " + data);
          return true;
        }
        if (!encodedName.equals(info.getEncodedName())) return true;
        ServerName sn = HRegionInfo.getServerName(data);
        result.set(new Pair<HRegionInfo, ServerName>(info, sn));
        return false; // found the region, stop
      }
    };

    MetaScanner.metaScan(conf, connection, visitor, null);
    pair = result.get();
  }
  return pair;
}

@Test
public void testUnassignWithSplitAtSameTime() throws KeeperException, IOException {
  // Region to use in test.
  final HRegionInfo hri = HRegionInfo.FIRST_META_REGIONINFO;
  // First amend the servermanager mock so that when we do send close of the
  // first meta region on SERVERNAME_A, it will return true rather than
  // default null.
  Mockito.when(this.serverManager.sendRegionClose(SERVERNAME_A, hri, -1)).thenReturn(true);
  // Need a mocked catalog tracker.
  CatalogTracker ct = Mockito.mock(CatalogTracker.class);
  LoadBalancer balancer = LoadBalancerFactory.getLoadBalancer(server
      .getConfiguration());
  // Create an AM.
  AssignmentManager am =
    new AssignmentManager(this.server, this.serverManager, ct, balancer, null);
  try {
    // First make sure my mock up basically works.  Unassign a region.
    unassign(am, SERVERNAME_A, hri);
    // This delete will fail if the previous unassign did wrong thing.
    ZKAssign.deleteClosingNode(this.watcher, hri);
    // Now put a SPLITTING region in the way.  I don't have to assert it
    // go put in place.  This method puts it in place then asserts it still
    // owns it by moving state from SPLITTING to SPLITTING.
    int version = createNodeSplitting(this.watcher, hri, SERVERNAME_A);
    // Now, retry the unassign with the SPLTTING in place.  It should just
    // complete without fail; a sort of 'silent' recognition that the
    // region to unassign has been split and no longer exists: TOOD: what if
    // the split fails and the parent region comes back to life?
    unassign(am, SERVERNAME_A, hri);
    // This transition should fail if the znode has been messed with.
    ZKAssign.transitionNode(this.watcher, hri, SERVERNAME_A,
      EventType.RS_ZK_REGION_SPLITTING, EventType.RS_ZK_REGION_SPLITTING, version);
    assertTrue(am.isRegionInTransition(hri) == null);
  } finally {
    am.shutdown();
  }
}

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());
  }
}

public AssignmentManagerWithExtrasForTesting(final Server master,
    final ServerManager serverManager, final CatalogTracker catalogTracker,
    final LoadBalancer balancer, final ExecutorService service)
throws KeeperException, IOException {
  super(master, serverManager, catalogTracker, balancer, service);
  this.es = service;
  this.ct = catalogTracker;
}

/**
 * Initialize the restore helper, based on the snapshot and table information provided.
 */
private RestoreSnapshotHelper getRestoreHelper(final Path rootDir, final Path snapshotDir,
    final String sourceTableName, final HTableDescriptor htdClone) throws IOException {
  CatalogTracker catalogTracker = Mockito.mock(CatalogTracker.class);
  HTableDescriptor tableDescriptor = Mockito.mock(HTableDescriptor.class);
  ForeignExceptionDispatcher monitor = Mockito.mock(ForeignExceptionDispatcher.class);
  MonitoredTask status = Mockito.mock(MonitoredTask.class);

  SnapshotDescription sd = SnapshotDescription.newBuilder()
    .setName("snapshot").setTable(sourceTableName).build();

  return new RestoreSnapshotHelper(conf, fs, sd, snapshotDir,
    htdClone, HTableDescriptor.getTableDir(rootDir, htdClone.getName()), monitor, status);
}

public SnapshotOfRegionAssignmentFromMeta(CatalogTracker tracker, Set<TableName> disabledTables,
    boolean excludeOfflinedSplitParents) {
  this.tracker = tracker;
  tableToRegionMap = new HashMap<TableName, List<HRegionInfo>>();
  regionToRegionServerMap = new HashMap<HRegionInfo, ServerName>();
  regionServerToRegionMap = new HashMap<ServerName, List<HRegionInfo>>();
  regionNameToRegionInfoMap = new TreeMap<String, HRegionInfo>();
  existingAssignmentPlan = new FavoredNodesPlan();
  this.disabledTables = disabledTables;
  this.excludeOfflinedSplitParents = excludeOfflinedSplitParents;
}

/**
 * Update meta table with favored nodes info
 * @param regionToFavoredNodes
 * @param catalogTracker
 * @throws IOException
 */
public static void updateMetaWithFavoredNodesInfo(
    Map<HRegionInfo, List<ServerName>> regionToFavoredNodes,
    CatalogTracker catalogTracker) throws IOException {
  List<Put> puts = new ArrayList<Put>();
  for (Map.Entry<HRegionInfo, List<ServerName>> entry : regionToFavoredNodes.entrySet()) {
    Put put = makePutFromRegionInfo(entry.getKey(), entry.getValue());
    if (put != null) {
      puts.add(put);
    }
  }
  MetaEditor.putsToMetaTable(catalogTracker, puts);
  LOG.info("Added " + puts.size() + " regions in META");
}

/**
 * Process a dead region from a dead RS. Checks if the region is disabled or
 * disabling or if the region has a partially completed split.
 * @param hri
 * @param result
 * @param assignmentManager
 * @param catalogTracker
 * @return Returns true if specified region should be assigned, false if not.
 * @throws IOException
 */
public static boolean processDeadRegion(HRegionInfo hri, Result result,
    AssignmentManager assignmentManager, CatalogTracker catalogTracker)
throws IOException {
  boolean tablePresent = assignmentManager.getZKTable().isTablePresent(hri.getTable());
  if (!tablePresent) {
    LOG.info("The table " + hri.getTable()
        + " was deleted.  Hence not proceeding.");
    return false;
  }
  // If table is not disabled but the region is offlined,
  boolean disabled = assignmentManager.getZKTable().isDisabledTable(hri.getTable());
  if (disabled){
    LOG.info("The table " + hri.getTable()
        + " was disabled.  Hence not proceeding.");
    return false;
  }
  if (hri.isOffline() && hri.isSplit()) {
    //HBASE-7721: Split parent and daughters are inserted into hbase:meta as an atomic operation.
    //If the meta scanner saw the parent split, then it should see the daughters as assigned
    //to the dead server. We don't have to do anything.
    return false;
  }
  boolean disabling = assignmentManager.getZKTable().isDisablingTable(hri.getTable());
  if (disabling) {
    LOG.info("The table " + hri.getTable()
        + " is disabled.  Hence not assigning region" + hri.getEncodedName());
    return false;
  }
  return true;
}

/**
 * @param tableName Table to check.
 * @return True if table exists already.
 * @throws IOException
 */
public boolean tableExists(final TableName tableName)
throws IOException {
  boolean b = false;
  CatalogTracker ct = getCatalogTracker();
  try {
    b = MetaReader.tableExists(ct, tableName);
  } finally {
    cleanupCatalogTracker(ct);
  }
  return b;
}

public EnableTableHandler(Server server, TableName tableName,
    CatalogTracker catalogTracker, AssignmentManager assignmentManager,
    TableLockManager tableLockManager, boolean skipTableStateCheck) {
  super(server, EventType.C_M_ENABLE_TABLE);
  this.tableName = tableName;
  this.catalogTracker = catalogTracker;
  this.assignmentManager = assignmentManager;
  this.tableLockManager = tableLockManager;
  this.skipTableStateCheck = skipTableStateCheck;
}

/**
 * @return the new RegionAssignmentSnapshot
 * @throws IOException
 */
public SnapshotOfRegionAssignmentFromMeta getRegionAssignmentSnapshot()
throws IOException {
  SnapshotOfRegionAssignmentFromMeta currentAssignmentShapshot =
    new SnapshotOfRegionAssignmentFromMeta(new CatalogTracker(this.conf));
  currentAssignmentShapshot.initialize();
  return currentAssignmentShapshot;
}

@Override
public void postOpenDeployTasks(final HRegion r, final CatalogTracker ct)
throws KeeperException, IOException {
  checkOpen();
  LOG.info("Post open deploy tasks for region=" + r.getRegionNameAsString());
  // Do checks to see if we need to compact (references or too many files)
  for (Store s : r.getStores().values()) {
    if (s.hasReferences() || s.needsCompaction()) {
     this.compactSplitThread.requestSystemCompaction(r, s, "Opening Region");
    }
  }
  long openSeqNum = r.getOpenSeqNum();
  if (openSeqNum == HConstants.NO_SEQNUM) {
    // If we opened a region, we should have read some sequence number from it.
    LOG.error("No sequence number found when opening " + r.getRegionNameAsString());
    openSeqNum = 0;
  }

  // Update flushed sequence id of a recovering region in ZK
  updateRecoveringRegionLastFlushedSequenceId(r);

  // Update ZK, or META
  if (r.getRegionInfo().isMetaRegion()) {
    MetaRegionTracker.setMetaLocation(getZooKeeper(),
        this.serverNameFromMasterPOV);
  } else {
    MetaEditor.updateRegionLocation(ct, r.getRegionInfo(),
      this.serverNameFromMasterPOV, openSeqNum);
  }
  LOG.info("Finished post open deploy task for " + r.getRegionNameAsString());

}

MockServer(final HBaseTestingUtility htu)
throws NotAllMetaRegionsOnlineException, IOException, InterruptedException {
  this.c = htu.getConfiguration();
  ClientProtos.ClientService.BlockingInterface ri =
    Mockito.mock(ClientProtos.ClientService.BlockingInterface.class);
  MutateResponse.Builder builder = MutateResponse.newBuilder();
  builder.setProcessed(true);
  try {
    Mockito.when(ri.mutate(
      (RpcController)Mockito.any(), (MutateRequest)Mockito.any())).
        thenReturn(builder.build());
  } catch (ServiceException se) {
    throw ProtobufUtil.getRemoteException(se);
  }
  // Mock an HConnection and a AdminProtocol implementation.  Have the
  // HConnection return the HRI.  Have the HRI return a few mocked up responses
  // to make our test work.
  this.connection =
    HConnectionTestingUtility.getMockedConnectionAndDecorate(this.c,
      Mockito.mock(AdminProtos.AdminService.BlockingInterface.class), ri,
        ServerName.valueOf("example.org,12345,6789"),
      HRegionInfo.FIRST_META_REGIONINFO);
  // Set hbase.rootdir into test dir.
  FileSystem fs = FileSystem.get(this.c);
  Path rootdir = FSUtils.getRootDir(this.c);
  FSUtils.setRootDir(this.c, rootdir);
  this.ct = Mockito.mock(CatalogTracker.class);
  AdminProtos.AdminService.BlockingInterface hri =
    Mockito.mock(AdminProtos.AdminService.BlockingInterface.class);
  Mockito.when(this.ct.getConnection()).thenReturn(this.connection);
  Mockito.when(ct.waitForMetaServerConnection(Mockito.anyLong())).thenReturn(hri);
}

/**
 * Compact a table or an individual region.
 * Asynchronous operation.
 *
 * @param tableNameOrRegionName table or region to compact
 * @param columnFamily column family within a table or region
 * @param major True if we are to do a major compaction.
 * @throws IOException if a remote or network exception occurs
 * @throws InterruptedException
 */
private void compact(final byte[] tableNameOrRegionName,
  final byte[] columnFamily,final boolean major)
throws IOException, InterruptedException {
  CatalogTracker ct = getCatalogTracker();
  try {
    Pair<HRegionInfo, ServerName> regionServerPair
      = getRegion(tableNameOrRegionName, ct);
    if (regionServerPair != null) {
      if (regionServerPair.getSecond() == null) {
        throw new NoServerForRegionException(Bytes.toStringBinary(tableNameOrRegionName));
      } else {
        compact(regionServerPair.getSecond(), regionServerPair.getFirst(), major, columnFamily);
      }
    } else {
      final TableName tableName =
          checkTableExists(TableName.valueOf(tableNameOrRegionName), ct);
      List<Pair<HRegionInfo, ServerName>> pairs =
        MetaReader.getTableRegionsAndLocations(ct,
            tableName);
      for (Pair<HRegionInfo, ServerName> pair: pairs) {
        if (pair.getFirst().isOffline()) continue;
        if (pair.getSecond() == null) continue;
        try {
          compact(pair.getSecond(), pair.getFirst(), major, columnFamily);
        } catch (NotServingRegionException e) {
          if (LOG.isDebugEnabled()) {
            LOG.debug("Trying to" + (major ? " major" : "") + " compact " +
              pair.getFirst() + ": " +
              StringUtils.stringifyException(e));
          }
        }
      }
    }
  } finally {
    cleanupCatalogTracker(ct);
  }
}

/**
 * Check if table exists or not
 * @param tableName Name of a table.
 * @param ct A {@link CatalogTracker} instance (caller of this method usually has one).
 * @return tableName instance
 * @throws IOException if a remote or network exception occurs.
 * @throws TableNotFoundException if table does not exist.
 */
//TODO rename this method
private TableName checkTableExists(
    final TableName tableName, CatalogTracker ct)
    throws IOException {
  if (!MetaReader.tableExists(ct, tableName)) {
    throw new TableNotFoundException(tableName);
  }
  return tableName;
}