*

* Licensed to the Apache Software Foundation (ASF) under one

* or more contributor license agreements.  See the NOTICE file

* distributed with this work for additional information

* regarding copyright ownership.  The ASF licenses this file

* to you under the Apache License, Version 2.0 (the

* "License"); you may not use this file except in compliance

* with the License.  You may obtain a copy of the License at

*

*     http://www.apache.org/licenses/LICENSE-2.0

*

* Unless required by applicable law or agreed to in writing, software

* distributed under the License is distributed on an "AS IS" BASIS,

* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

* See the License for the specific language governing permissions and

* limitations under the License.

*/

* Manages and performs region assignment.

* Related communications with regionserver are all done over RPC.

*/

  private static final Log LOG = LogFactory.getLog(AssignmentManager.class);

  protected final Server server;

  private ServerManager serverManager;

  private boolean shouldAssignRegionsWithFavoredNodes;

  private LoadBalancer balancer;

  private final MetricsAssignmentManager metricsAssignmentManager;

  private final TableLockManager tableLockManager;

  private AtomicInteger numRegionsOpened = new AtomicInteger(0);

  final private KeyLocker<String> locker = new KeyLocker<String>();

  Set<HRegionInfo> replicasToClose = Collections.synchronizedSet(new HashSet<HRegionInfo>());

  /**

   * Map of regions to reopen after the schema of a table is changed. Key -

   * encoded region name, value - HRegionInfo

   */

  private final Map <String, HRegionInfo> regionsToReopen;

  /*

   * Maximum times we recurse an assignment/unassignment.

   * See below in {@link #assign()} and {@link #unassign()}.

   */

  private final int maximumAttempts;

  /**

   * The sleep time for which the assignment will wait before retrying in case of hbase:meta assignment

   * failure due to lack of availability of region plan

   */

  private final long sleepTimeBeforeRetryingMetaAssignment;

  /** Plans for region movement. Key is the encoded version of a region name*/

  // TODO: When do plans get cleaned out?  Ever? In server open and in server

  // shutdown processing -- St.Ack

  // All access to this Map must be synchronized.

  final NavigableMap<String, RegionPlan> regionPlans =

    new TreeMap<String, RegionPlan>();

  private final TableStateManager tableStateManager;

  private final ExecutorService executorService;

  // Thread pool executor service. TODO, consolidate with executorService?

  private java.util.concurrent.ExecutorService threadPoolExecutorService;

  private final RegionStates regionStates;

  // The threshold to use bulk assigning. Using bulk assignment

  // only if assigning at least this many regions to at least this

  // many servers. If assigning fewer regions to fewer servers,

  // bulk assigning may be not as efficient.

  private final int bulkAssignThresholdRegions;

  private final int bulkAssignThresholdServers;

  // Should bulk assignment wait till all regions are assigned,

  // or it is timed out?  This is useful to measure bulk assignment

  // performance, but not needed in most use cases.

  private final boolean bulkAssignWaitTillAllAssigned;

  /**

   * Indicator that AssignmentManager has recovered the region states so

   * that ServerShutdownHandler can be fully enabled and re-assign regions

   * of dead servers. So that when re-assignment happens, AssignmentManager

   * has proper region states.

   *

   * Protected to ease testing.

   */

  protected final AtomicBoolean failoverCleanupDone = new AtomicBoolean(false);

  /**

   * A map to track the count a region fails to open in a row.

   * So that we don't try to open a region forever if the failure is

   * unrecoverable.  We don't put this information in region states

   * because we don't expect this to happen frequently; we don't

   * want to copy this information over during each state transition either.

   */

  private final ConcurrentHashMap<String, AtomicInteger>

    failedOpenTracker = new ConcurrentHashMap<String, AtomicInteger>();

  // In case not using ZK for region assignment, region states

  // are persisted in meta with a state store

  private final RegionStateStore regionStateStore;

  /**

   * For testing only!  Set to true to skip handling of split.

   */

  @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="MS_SHOULD_BE_FINAL")

  public static boolean TEST_SKIP_SPLIT_HANDLING = false;

  /** Listeners that are called on assignment events. */

  private List<AssignmentListener> listeners = new CopyOnWriteArrayList<AssignmentListener>();

  /**

   * Constructs a new assignment manager.

   *

   * @param server instance of HMaster this AM running inside

   * @param serverManager serverManager for associated HMaster

   * @param balancer implementation of {@link LoadBalancer}

   * @param service Executor service

   * @param metricsMaster metrics manager

   * @param tableLockManager TableLock manager

   * @throws IOException

   */

  public AssignmentManager(Server server, ServerManager serverManager,

      final LoadBalancer balancer,

      final ExecutorService service, MetricsMaster metricsMaster,

      final TableLockManager tableLockManager,

      final TableStateManager tableStateManager)

          throws IOException {

    this.server = server;

    this.serverManager = serverManager;

    this.executorService = service;

    this.regionStateStore = new RegionStateStore(server);

    this.regionsToReopen = Collections.synchronizedMap

                           (new HashMap<String, HRegionInfo> ());

    Configuration conf = server.getConfiguration();

    // Only read favored nodes if using the favored nodes load balancer.

    this.shouldAssignRegionsWithFavoredNodes = conf.getClass(

           HConstants.HBASE_MASTER_LOADBALANCER_CLASS, Object.class).equals(

           FavoredNodeLoadBalancer.class);

    this.tableStateManager = tableStateManager;

    // This is the max attempts, not retries, so it should be at least 1.

    this.maximumAttempts = Math.max(1,

      this.server.getConfiguration().getInt("hbase.assignment.maximum.attempts", 10));

    this.sleepTimeBeforeRetryingMetaAssignment = this.server.getConfiguration().getLong(

        "hbase.meta.assignment.retry.sleeptime", 1000l);

    this.balancer = balancer;

    int maxThreads = conf.getInt("hbase.assignment.threads.max", 30);

    this.threadPoolExecutorService = Threads.getBoundedCachedThreadPool(

      maxThreads, 60L, TimeUnit.SECONDS, Threads.newDaemonThreadFactory("AM."));

    this.regionStates = new RegionStates(

      server, tableStateManager, serverManager, regionStateStore);

    this.bulkAssignWaitTillAllAssigned =

      conf.getBoolean("hbase.bulk.assignment.waittillallassigned", false);

    this.bulkAssignThresholdRegions = conf.getInt("hbase.bulk.assignment.threshold.regions", 7);

    this.bulkAssignThresholdServers = conf.getInt("hbase.bulk.assignment.threshold.servers", 3);

    this.metricsAssignmentManager = new MetricsAssignmentManager();

    this.tableLockManager = tableLockManager;

  }

  /**

   * Add the listener to the notification list.

   * @param listener The AssignmentListener to register

   */

  public void registerListener(final AssignmentListener listener) {

    this.listeners.add(listener);

  }

  /**

   * Remove the listener from the notification list.

   * @param listener The AssignmentListener to unregister

   */

  public boolean unregisterListener(final AssignmentListener listener) {

    return this.listeners.remove(listener);

  }

  /**

   * @return Instance of ZKTableStateManager.

   */

  public TableStateManager getTableStateManager() {

    // These are 'expensive' to make involving trip to zk ensemble so allow

    // sharing.

    return this.tableStateManager;

  }

  /**

   * This SHOULD not be public. It is public now

   * because of some unit tests.

   *

   * TODO: make it package private and keep RegionStates in the master package

   */

  public RegionStates getRegionStates() {

    return regionStates;

  }

  /**

   * Used in some tests to mock up region state in meta

   */

  @VisibleForTesting

  RegionStateStore getRegionStateStore() {

    return regionStateStore;

  }

  public RegionPlan getRegionReopenPlan(HRegionInfo hri) {

    return new RegionPlan(hri, null, regionStates.getRegionServerOfRegion(hri));

  }

  /**

   * Add a regionPlan for the specified region.

   * @param encodedName

   * @param plan

   */

  public void addPlan(String encodedName, RegionPlan plan) {

    synchronized (regionPlans) {

      regionPlans.put(encodedName, plan);

    }

  }

  /**

   * Add a map of region plans.

   */

  public void addPlans(Map<String, RegionPlan> plans) {

    synchronized (regionPlans) {

      regionPlans.putAll(plans);

    }

  }

  /**

   * Set the list of regions that will be reopened

   * because of an update in table schema

   *

   * @param regions

   *          list of regions that should be tracked for reopen

   */

  public void setRegionsToReopen(List <HRegionInfo> regions) {

    for(HRegionInfo hri : regions) {

      regionsToReopen.put(hri.getEncodedName(), hri);

    }

  }

  /**

   * Used by the client to identify if all regions have the schema updates

   *

   * @param tableName

   * @return Pair indicating the status of the alter command

   * @throws IOException

   */

  public Pair<Integer, Integer> getReopenStatus(TableName tableName)

      throws IOException {

    List<HRegionInfo> hris;

    if (TableName.META_TABLE_NAME.equals(tableName)) {

      hris = new MetaTableLocator().getMetaRegions(server.getZooKeeper());

    } else {

      hris = MetaTableAccessor.getTableRegions(server.getShortCircuitConnection(), tableName, true);

    }

    Integer pending = 0;

    for (HRegionInfo hri : hris) {

      String name = hri.getEncodedName();

      // no lock concurrent access ok: sequential consistency respected.

      if (regionsToReopen.containsKey(name)

          || regionStates.isRegionInTransition(name)) {

        pending++;

      }

    }

    return new Pair<Integer, Integer>(pending, hris.size());

  }

  /**

   * Used by ServerShutdownHandler to make sure AssignmentManager has completed

   * the failover cleanup before re-assigning regions of dead servers. So that

   * when re-assignment happens, AssignmentManager has proper region states.

   */

  public boolean isFailoverCleanupDone() {

    return failoverCleanupDone.get();

  }

  /**

   * To avoid racing with AM, external entities may need to lock a region,

   * for example, when SSH checks what regions to skip re-assigning.

   */

  public Lock acquireRegionLock(final String encodedName) {

    return locker.acquireLock(encodedName);

  }

  /**

   * Now, failover cleanup is completed. Notify server manager to

   * process queued up dead servers processing, if any.

   */

  void failoverCleanupDone() {

    failoverCleanupDone.set(true);

    serverManager.processQueuedDeadServers();

  }

  /**

   * Called on startup.

   * Figures whether a fresh cluster start of we are joining extant running cluster.

   * @throws IOException

   * @throws KeeperException

   * @throws InterruptedException

   */

  void joinCluster() throws IOException,

          KeeperException, InterruptedException {

    long startTime = System.currentTimeMillis();

    // Concurrency note: In the below the accesses on regionsInTransition are

    // outside of a synchronization block where usually all accesses to RIT are

    // synchronized.  The presumption is that in this case it is safe since this

    // method is being played by a single thread on startup.

    // TODO: Regions that have a null location and are not in regionsInTransitions

    // need to be handled.

    // Scan hbase:meta to build list of existing regions, servers, and assignment

    // Returns servers who have not checked in (assumed dead) that some regions

    // were assigned to (according to the meta)

    Set<ServerName> deadServers = rebuildUserRegions();

    // This method will assign all user regions if a clean server startup or

    // it will reconstruct master state and cleanup any leftovers from

    // previous master process.

    boolean failover = processDeadServersAndRegionsInTransition(deadServers);

    recoverTableInDisablingState();

    recoverTableInEnablingState();

    LOG.info("Joined the cluster in " + (System.currentTimeMillis()

      - startTime) + "ms, failover=" + failover);

  }

  /**

   * Process all regions that are in transition in zookeeper and also

   * processes the list of dead servers by scanning the META.

   * Used by master joining an cluster.  If we figure this is a clean cluster

   * startup, will assign all user regions.

   * @param deadServers

   *          Map of dead servers and their regions. Can be null.

   * @throws IOException

   * @throws InterruptedException

   */

  boolean processDeadServersAndRegionsInTransition(final Set<ServerName> deadServers)

          throws IOException, InterruptedException {

    boolean failover = !serverManager.getDeadServers().isEmpty();

    if (failover) {

      // This may not be a failover actually, especially if meta is on this master.

      if (LOG.isDebugEnabled()) {

        LOG.debug("Found dead servers out on cluster " + serverManager.getDeadServers());

      }

    } else {

      // If any one region except meta is assigned, it's a failover.

      Set<ServerName> onlineServers = serverManager.getOnlineServers().keySet();

      for (Map.Entry<HRegionInfo, ServerName> en:

          regionStates.getRegionAssignments().entrySet()) {

        HRegionInfo hri = en.getKey();

        if (!hri.isMetaTable()

            && onlineServers.contains(en.getValue())) {

          LOG.debug("Found " + hri + " out on cluster");

          failover = true;

          break;

        }

      }

      if (!failover) {

        // If any region except meta is in transition on a live server, it's a failover.

        Map<String, RegionState> regionsInTransition = regionStates.getRegionsInTransition();

        if (!regionsInTransition.isEmpty()) {

          for (RegionState regionState: regionsInTransition.values()) {

            if (!regionState.getRegion().isMetaRegion()

                && onlineServers.contains(regionState.getServerName())) {

              LOG.debug("Found " + regionState + " in RITs");

              failover = true;

              break;

            }

          }

        }

      }

    }

    if (!failover) {

      // If we get here, we have a full cluster restart. It is a failover only

      // if there are some HLogs are not split yet. For meta HLogs, they should have

      // been split already, if any. We can walk through those queued dead servers,

      // if they don't have any HLogs, this restart should be considered as a clean one

      Set<ServerName> queuedDeadServers = serverManager.getRequeuedDeadServers().keySet();

      if (!queuedDeadServers.isEmpty()) {

        Configuration conf = server.getConfiguration();

        Path rootdir = FSUtils.getRootDir(conf);

        FileSystem fs = rootdir.getFileSystem(conf);

        for (ServerName serverName: queuedDeadServers) {

          Path logDir = new Path(rootdir, HLogUtil.getHLogDirectoryName(serverName.toString()));

          Path splitDir = logDir.suffix(HLog.SPLITTING_EXT);

          if (fs.exists(logDir) || fs.exists(splitDir)) {

            LOG.debug("Found queued dead server " + serverName);

            failover = true;

            break;

          }

        }

        if (!failover) {

          // We figured that it's not a failover, so no need to

          // work on these re-queued dead servers any more.

          LOG.info("AM figured that it's not a failover and cleaned up "

            + queuedDeadServers.size() + " queued dead servers");

          serverManager.removeRequeuedDeadServers();

        }

      }

    }

    Set<TableName> disabledOrDisablingOrEnabling = null;

    Map<HRegionInfo, ServerName> allRegions = null;

    if (!failover) {

      disabledOrDisablingOrEnabling = tableStateManager.getTablesInStates(

        TableState.State.DISABLED, TableState.State.DISABLING,

        TableState.State.ENABLING);

      // Clean re/start, mark all user regions closed before reassignment

      allRegions = regionStates.closeAllUserRegions(

        disabledOrDisablingOrEnabling);

    }

    // Now region states are restored

    regionStateStore.start();

    if (failover) {

      if (deadServers != null && !deadServers.isEmpty()) {

        for (ServerName serverName: deadServers) {

          if (!serverManager.isServerDead(serverName)) {

            serverManager.expireServer(serverName); // Let SSH do region re-assign

          }

        }

      }

      processRegionsInTransition(regionStates.getRegionsInTransition().values());

    }

    // Now we can safely claim failover cleanup completed and enable

    // ServerShutdownHandler for further processing. The nodes (below)

    // in transition, if any, are for regions not related to those

    // dead servers at all, and can be done in parallel to SSH.

    failoverCleanupDone();

    if (!failover) {

      // Fresh cluster startup.

      LOG.info("Clean cluster startup. Assigning user regions");

      assignAllUserRegions(allRegions);

    }

    // unassign replicas of the split parents and the merged regions

    // the daughter replicas are opened in assignAllUserRegions if it was

    // not already opened.

    for (HRegionInfo h : replicasToClose) {

      unassign(h);

    }

    replicasToClose.clear();

    return failover;

  }

  /**

   * When a region is closed, it should be removed from the regionsToReopen

   * @param hri HRegionInfo of the region which was closed

   */

  public void removeClosedRegion(HRegionInfo hri) {

    if (regionsToReopen.remove(hri.getEncodedName()) != null) {

      LOG.debug("Removed region from reopening regions because it was closed");

    }

  }

  // TODO: processFavoredNodes might throw an exception, for e.g., if the

  // meta could not be contacted/updated. We need to see how seriously to treat

  // this problem as. Should we fail the current assignment. We should be able

  // to recover from this problem eventually (if the meta couldn't be updated

  // things should work normally and eventually get fixed up).

  void processFavoredNodes(List<HRegionInfo> regions) throws IOException {

    if (!shouldAssignRegionsWithFavoredNodes) return;

    // The AM gets the favored nodes info for each region and updates the meta

    // table with that info

    Map<HRegionInfo, List<ServerName>> regionToFavoredNodes =

        new HashMap<HRegionInfo, List<ServerName>>();

    for (HRegionInfo region : regions) {

      regionToFavoredNodes.put(region,

          ((FavoredNodeLoadBalancer)this.balancer).getFavoredNodes(region));

    }

    FavoredNodeAssignmentHelper.updateMetaWithFavoredNodesInfo(regionToFavoredNodes,

      this.server.getShortCircuitConnection());

  }

  /**

   * Marks the region as online.  Removes it from regions in transition and

   * updates the in-memory assignment information.

   * <p>

   * Used when a region has been successfully opened on a region server.

   * @param regionInfo

   * @param sn

   */

  void regionOnline(HRegionInfo regionInfo, ServerName sn) {

    regionOnline(regionInfo, sn, HConstants.NO_SEQNUM);

  }

  void regionOnline(HRegionInfo regionInfo, ServerName sn, long openSeqNum) {

    numRegionsOpened.incrementAndGet();

    regionStates.regionOnline(regionInfo, sn, openSeqNum);

    // Remove plan if one.

    clearRegionPlan(regionInfo);

    balancer.regionOnline(regionInfo, sn);

    // Tell our listeners that a region was opened

    sendRegionOpenedNotification(regionInfo, sn);

  }

  /**

   * Marks the region as offline.  Removes it from regions in transition and

   * removes in-memory assignment information.

   * <p>

   * Used when a region has been closed and should remain closed.

   * @param regionInfo

   */

  public void regionOffline(final HRegionInfo regionInfo) {

    regionOffline(regionInfo, null);

  }

  public void offlineDisabledRegion(HRegionInfo regionInfo) {

    replicasToClose.remove(regionInfo);

    regionOffline(regionInfo);

  }

  // Assignment methods

  /**

   * Assigns the specified region.

   * <p>

   * If a RegionPlan is available with a valid destination then it will be used

   * to determine what server region is assigned to.  If no RegionPlan is

   * available, region will be assigned to a random available server.

   * <p>

   * Updates the RegionState and sends the OPEN RPC.

   * <p>

   * This will only succeed if the region is in transition and in a CLOSED or

   * OFFLINE state or not in transition, and of course, the

   * chosen server is up and running (It may have just crashed!).

   *

   * @param region server to be assigned

   */

  public void assign(HRegionInfo region) {

    assign(region, false);

  }

  /**

   * Use care with forceNewPlan. It could cause double assignment.

   */

  public void assign(HRegionInfo region, boolean forceNewPlan) {

    if (isDisabledorDisablingRegionInRIT(region)) {

      return;

    }

    String encodedName = region.getEncodedName();

    Lock lock = locker.acquireLock(encodedName);

    try {

      RegionState state = forceRegionStateToOffline(region, forceNewPlan);

      if (state != null) {

        if (regionStates.wasRegionOnDeadServer(encodedName)) {

          LOG.info("Skip assigning " + region.getRegionNameAsString()

            + ", it's host " + regionStates.getLastRegionServerOfRegion(encodedName)

            + " is dead but not processed yet");

          return;

        }

        assign(state, forceNewPlan);

      }

    } finally {

      lock.unlock();

    }

  }

  /**

   * Bulk assign regions to <code>destination</code>.

   * @param destination

   * @param regions Regions to assign.

   * @return true if successful

   */

  boolean assign(final ServerName destination, final List<HRegionInfo> regions)

    throws InterruptedException {

    long startTime = EnvironmentEdgeManager.currentTime();

    try {

      int regionCount = regions.size();

      if (regionCount == 0) {

        return true;

      }

      LOG.info("Assigning " + regionCount + " region(s) to " + destination.toString());

      Set<String> encodedNames = new HashSet<String>(regionCount);

      for (HRegionInfo region : regions) {

        encodedNames.add(region.getEncodedName());

      }

      List<HRegionInfo> failedToOpenRegions = new ArrayList<HRegionInfo>();

      Map<String, Lock> locks = locker.acquireLocks(encodedNames);

      try {

        Map<String, RegionPlan> plans = new HashMap<String, RegionPlan>(regionCount);

        List<RegionState> states = new ArrayList<RegionState>(regionCount);

        for (HRegionInfo region : regions) {

          String encodedName = region.getEncodedName();

          if (!isDisabledorDisablingRegionInRIT(region)) {

            RegionState state = forceRegionStateToOffline(region, false);

            boolean onDeadServer = false;

            if (state != null) {

              if (regionStates.wasRegionOnDeadServer(encodedName)) {

                LOG.info("Skip assigning " + region.getRegionNameAsString()

                  + ", it's host " + regionStates.getLastRegionServerOfRegion(encodedName)

                  + " is dead but not processed yet");

                onDeadServer = true;

              } else {

                RegionPlan plan = new RegionPlan(region, state.getServerName(), destination);

                plans.put(encodedName, plan);

                states.add(state);

                continue;

              }

            }

            // Reassign if the region wasn't on a dead server

            if (!onDeadServer) {

              LOG.info("failed to force region state to offline, "

                + "will reassign later: " + region);

              failedToOpenRegions.add(region); // assign individually later

            }

          }

          // Release the lock, this region is excluded from bulk assign because

          // we can't update its state, or set its znode to offline.

          Lock lock = locks.remove(encodedName);

          lock.unlock();

        }

        if (server.isStopped()) {

          return false;

        }

        // Add region plans, so we can updateTimers when one region is opened so

        // that unnecessary timeout on RIT is reduced.

        this.addPlans(plans);

        List<Pair<HRegionInfo, List<ServerName>>> regionOpenInfos =

          new ArrayList<Pair<HRegionInfo, List<ServerName>>>(states.size());

        for (RegionState state: states) {

          HRegionInfo region = state.getRegion();

          regionStates.updateRegionState(

            region, State.PENDING_OPEN, destination);

          List<ServerName> favoredNodes = ServerName.EMPTY_SERVER_LIST;

          if (this.shouldAssignRegionsWithFavoredNodes) {

            favoredNodes = ((FavoredNodeLoadBalancer)this.balancer).getFavoredNodes(region);

          }

          regionOpenInfos.add(new Pair<HRegionInfo, List<ServerName>>(

            region, favoredNodes));

        }

        // Move on to open regions.

        try {

          // Send OPEN RPC. If it fails on a IOE or RemoteException,

          // regions will be assigned individually.

          Configuration conf = server.getConfiguration();

          long maxWaitTime = System.currentTimeMillis() +

            conf.getLong("hbase.regionserver.rpc.startup.waittime", 60000);

          for (int i = 1; i <= maximumAttempts && !server.isStopped(); i++) {

            try {

              List<RegionOpeningState> regionOpeningStateList = serverManager

                .sendRegionOpen(destination, regionOpenInfos);

              for (int k = 0, n = regionOpeningStateList.size(); k < n; k++) {

                RegionOpeningState openingState = regionOpeningStateList.get(k);

                if (openingState != RegionOpeningState.OPENED) {

                  HRegionInfo region = regionOpenInfos.get(k).getFirst();

                  LOG.info("Got opening state " + openingState

                    + ", will reassign later: " + region);

                  // Failed opening this region, reassign it later

                  forceRegionStateToOffline(region, true);

                  failedToOpenRegions.add(region);

                }

              }

              break;

            } catch (IOException e) {

              if (e instanceof RemoteException) {

                e = ((RemoteException)e).unwrapRemoteException();

              }

              if (e instanceof RegionServerStoppedException) {

                LOG.warn("The region server was shut down, ", e);