http://git-wip-us.apache.org/repos/asf/hbase/blob/a3c5a744/hbase-server/src/main/java/org/apache/hadoop/hbase/master/AssignmentManager.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/AssignmentManager.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/AssignmentManager.java new file mode 100644 index 0000000..69ebd97 --- /dev/null +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/AssignmentManager.java @@ -0,0 +1,3053 @@ +/** + * + * 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. + */ +package org.apache.hadoop.hbase.master; + +import com.google.common.annotations.VisibleForTesting; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.NavigableMap; +import java.util.Random; +import java.util.Set; +import java.util.TreeMap; +import java.util.concurrent.Callable; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hbase.CoordinatedStateException; +import org.apache.hadoop.hbase.HBaseIOException; +import org.apache.hadoop.hbase.HConstants; +import org.apache.hadoop.hbase.HRegionInfo; +import org.apache.hadoop.hbase.HRegionLocation; +import org.apache.hadoop.hbase.HTableDescriptor; +import org.apache.hadoop.hbase.MetaTableAccessor; +import org.apache.hadoop.hbase.NotServingRegionException; +import org.apache.hadoop.hbase.RegionLocations; +import org.apache.hadoop.hbase.RegionStateListener; +import org.apache.hadoop.hbase.ServerName; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.classification.InterfaceAudience; +import org.apache.hadoop.hbase.client.MasterSwitchType; +import org.apache.hadoop.hbase.client.RegionReplicaUtil; +import org.apache.hadoop.hbase.client.Result; +import org.apache.hadoop.hbase.client.TableState; +import org.apache.hadoop.hbase.executor.EventHandler; +import org.apache.hadoop.hbase.executor.EventType; +import org.apache.hadoop.hbase.executor.ExecutorService; +import org.apache.hadoop.hbase.favored.FavoredNodesManager; +import org.apache.hadoop.hbase.favored.FavoredNodesPromoter; +import org.apache.hadoop.hbase.ipc.FailedServerException; +import org.apache.hadoop.hbase.ipc.RpcClient; +import org.apache.hadoop.hbase.ipc.ServerNotRunningYetException; +import org.apache.hadoop.hbase.master.RegionState.State; +import org.apache.hadoop.hbase.master.normalizer.NormalizationPlan.PlanType; +import org.apache.hadoop.hbase.shaded.protobuf.generated.RegionServerStatusProtos.RegionStateTransition; +import org.apache.hadoop.hbase.shaded.protobuf.generated.RegionServerStatusProtos.RegionStateTransition.TransitionCode; +import org.apache.hadoop.hbase.quotas.QuotaExceededException; +import org.apache.hadoop.hbase.regionserver.RegionOpeningState; +import org.apache.hadoop.hbase.regionserver.RegionServerAbortedException; +import org.apache.hadoop.hbase.regionserver.RegionServerStoppedException; +import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; +import org.apache.hadoop.hbase.util.FSUtils; +import org.apache.hadoop.hbase.util.KeyLocker; +import org.apache.hadoop.hbase.util.Pair; +import org.apache.hadoop.hbase.util.PairOfSameType; +import org.apache.hadoop.hbase.util.RetryCounter; +import org.apache.hadoop.hbase.util.Threads; +import org.apache.hadoop.hbase.wal.AbstractFSWALProvider; +import org.apache.hadoop.hbase.zookeeper.MetaTableLocator; +import org.apache.hadoop.ipc.RemoteException; +import org.apache.hadoop.util.StringUtils; +import org.apache.zookeeper.KeeperException; + +/** + * Manages and performs region assignment. + * Related communications with regionserver are all done over RPC. + */ +@InterfaceAudience.Private +public class AssignmentManager { + private static final Log LOG = LogFactory.getLog(AssignmentManager.class); + + protected final MasterServices server; + + private ServerManager serverManager; + + private boolean shouldAssignRegionsWithFavoredNodes; + + private LoadBalancer balancer; + + private final MetricsAssignmentManager metricsAssignmentManager; + + private AtomicInteger numRegionsOpened = new AtomicInteger(0); + + final private KeyLocker<String> locker = new KeyLocker<>(); + + 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 or bad 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<>(); + + private final TableStateManager tableStateManager; + + private final ExecutorService executorService; + + private java.util.concurrent.ExecutorService threadPoolExecutorService; + private ScheduledThreadPoolExecutor scheduledThreadPoolExecutor; + + 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; + private final int bulkPerRegionOpenTimeGuesstimate; + + // 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<>(); + + // 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<>(); + + private RegionStateListener regionStateListener; + + private RetryCounter.BackoffPolicy backoffPolicy; + private RetryCounter.RetryConfig retryConfig; + /** + * 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 + * @throws IOException + */ + public AssignmentManager(MasterServices server, ServerManager serverManager, + final LoadBalancer balancer, final ExecutorService service, MetricsMaster metricsMaster, + 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(); + + 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; + // Only read favored nodes if using the favored nodes load balancer. + this.shouldAssignRegionsWithFavoredNodes = this.balancer instanceof FavoredNodesPromoter; + int maxThreads = conf.getInt("hbase.assignment.threads.max", 30); + + this.threadPoolExecutorService = Threads.getBoundedCachedThreadPool( + maxThreads, 60L, TimeUnit.SECONDS, Threads.newDaemonThreadFactory("AM.")); + + this.scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1, + Threads.newDaemonThreadFactory("AM.Scheduler")); + + 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.bulkPerRegionOpenTimeGuesstimate = + conf.getInt("hbase.bulk.assignment.perregion.open.time", 10000); + + this.metricsAssignmentManager = new MetricsAssignmentManager(); + + // Configurations for retrying opening a region on receiving a FAILED_OPEN + this.retryConfig = new RetryCounter.RetryConfig(); + this.retryConfig.setSleepInterval(conf.getLong("hbase.assignment.retry.sleep.initial", 0l)); + // Set the max time limit to the initial sleep interval so we use a constant time sleep strategy + // if the user does not set a max sleep limit + this.retryConfig.setMaxSleepTime(conf.getLong("hbase.assignment.retry.sleep.max", + retryConfig.getSleepInterval())); + this.backoffPolicy = getBackoffPolicy(); + } + + /** + * Returns the backoff policy used for Failed Region Open retries + * @return the backoff policy used for Failed Region Open retries + */ + RetryCounter.BackoffPolicy getBackoffPolicy() { + return new RetryCounter.ExponentialBackoffPolicyWithLimit(); + } + + MetricsAssignmentManager getAssignmentManagerMetrics() { + return this.metricsAssignmentManager; + } + + /** + * 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.getConnection(), 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<>(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 + * @throws CoordinatedStateException + */ + void joinCluster() + throws IOException, KeeperException, InterruptedException, CoordinatedStateException { + 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); + + 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. + * Used by master joining an cluster. If we figure this is a clean cluster + * startup, will assign all user regions. + * @param deadServers Set of servers that are offline probably legitimately that were carrying + * regions according to a scan of hbase:meta. Can be null. + * @throws IOException + * @throws InterruptedException + */ + boolean processDeadServersAndRegionsInTransition(final Set<ServerName> deadServers) + throws KeeperException, IOException, InterruptedException, CoordinatedStateException { + // TODO Needed? List<String> nodes = ZKUtil.listChildrenNoWatch(watcher, watcher.assignmentZNode); + 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()); + } + // Check if there are any regions on these servers + failover = false; + for (ServerName serverName : serverManager.getDeadServers().copyServerNames()) { + if (regionStates.getRegionAssignments().values().contains(serverName)) { + LOG.debug("Found regions on dead server: " + serverName); + failover = true; + break; + } + } + } + Set<ServerName> onlineServers = serverManager.getOnlineServers().keySet(); + if (!failover) { + // If any one region except meta is assigned, it's a failover. + for (Map.Entry<HRegionInfo, ServerName> en: + regionStates.getRegionAssignments().entrySet()) { + HRegionInfo hri = en.getKey(); + if (!hri.isMetaTable() + && onlineServers.contains(en.getValue())) { + LOG.debug("Found region " + 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. + Set<RegionState> regionsInTransition = regionStates.getRegionsInTransition(); + if (!regionsInTransition.isEmpty()) { + for (RegionState regionState: regionsInTransition) { + ServerName serverName = regionState.getServerName(); + if (!regionState.getRegion().isMetaRegion() + && serverName != null && onlineServers.contains(serverName)) { + LOG.debug("Found " + regionState + " for region " + + regionState.getRegion().getRegionNameAsString() + " for server " + + serverName + "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 WALs are not split yet. For meta WALs, they should have + // been split already, if any. We can walk through those queued dead servers, + // if they don't have any WALs, this restart should be considered as a clean one + Set<ServerName> queuedDeadServers = serverManager.getRequeuedDeadServers().keySet(); + if (!queuedDeadServers.isEmpty()) { + Configuration conf = server.getConfiguration(); + Path walRootDir = FSUtils.getWALRootDir(conf); + FileSystem walFs = FSUtils.getWALFileSystem(conf); + for (ServerName serverName: queuedDeadServers) { + // In the case of a clean exit, the shutdown handler would have presplit any WALs and + // removed empty directories. + Path walDir = new Path(walRootDir, + AbstractFSWALProvider.getWALDirectoryName(serverName.toString())); + Path splitDir = walDir.suffix(AbstractFSWALProvider.SPLITTING_EXT); + if (checkWals(walFs, walDir) || checkWals(walFs, 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()); + } + + // 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. Don't reassign user regions"); + assignAllUserRegions(allRegions); + } else { + LOG.info("Failover! Reassign user regions"); + } + // 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; + } + + private boolean checkWals(FileSystem fs, Path dir) throws IOException { + if (!fs.exists(dir)) { + LOG.debug(dir + " doesn't exist"); + return false; + } + if (!fs.getFileStatus(dir).isDirectory()) { + LOG.warn(dir + " is not a directory"); + return false; + } + FileStatus[] files = FSUtils.listStatus(fs, dir); + if (files == null || files.length == 0) { + LOG.debug(dir + " has no files"); + return false; + } + for (int i = 0; i < files.length; i++) { + if (files[i].isFile() && files[i].getLen() > 0) { + LOG.debug(dir + " has a non-empty file: " + files[i].getPath()); + return true; + } else if (files[i].isDirectory() && checkWals(fs, files[i].getPath())) { + LOG.debug(dir + " is a directory and has a non-empty file: " + files[i].getPath()); + return true; + } + } + LOG.debug("Found 0 non-empty wal files for :" + dir); + return false; + } + + /** + * 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"); + } + } + + void processFavoredNodesForDaughters(HRegionInfo parent, + HRegionInfo regionA, HRegionInfo regionB) throws IOException { + if (shouldAssignFavoredNodes(parent)) { + List<ServerName> onlineServers = this.serverManager.getOnlineServersList(); + ((FavoredNodesPromoter) this.balancer). + generateFavoredNodesForDaughter(onlineServers, parent, regionA, regionB); + } + } + + void processFavoredNodesForMerge(HRegionInfo merged, HRegionInfo regionA, HRegionInfo regionB) + throws IOException { + if (shouldAssignFavoredNodes(merged)) { + ((FavoredNodesPromoter)this.balancer). + generateFavoredNodesForMergedRegion(merged, regionA, regionB); + } + } + + /* + * Favored nodes should be applied only when FavoredNodes balancer is configured and the region + * belongs to a non-system table. + */ + private boolean shouldAssignFavoredNodes(HRegionInfo region) { + return this.shouldAssignRegionsWithFavoredNodes + && FavoredNodesManager.isFavoredNodeApplicable(region); + } + + /** + * 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<>(regionCount); + for (HRegionInfo region : regions) { + encodedNames.add(region.getEncodedName()); + } + + List<HRegionInfo> failedToOpenRegions = new ArrayList<>(); + Map<String, Lock> locks = locker.acquireLocks(encodedNames); + try { + Map<String, RegionPlan> plans = new HashMap<>(regionCount); + List<RegionState> states = new ArrayList<>(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<>(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 (shouldAssignFavoredNodes(region)) { + favoredNodes = server.getFavoredNodesManager().getFavoredNodesWithDNPort(region); + } + regionOpenInfos.add(new Pair<>(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); + // No need to retry, the region server is a goner. + return false; + } else if (e instanceof ServerNotRunningYetException) { + long now = System.currentTimeMillis(); + if (now < maxWaitTime) { + if (LOG.isDebugEnabled()) { + LOG.debug("Server is not yet up; waiting up to " + + (maxWaitTime - now) + "ms", e); + } + Thread.sleep(100); + i--; // reset the try count + continue; + } + } else if (e instanceof java.net.SocketTimeoutException + && this.serverManager.isServerOnline(destination)) { + // In case socket is timed out and the region server is still online, + // the openRegion RPC could have been accepted by the server and + // just the response didn't go through. So we will retry to + // open the region on the same server. + if (LOG.isDebugEnabled()) { + LOG.debug("Bulk assigner openRegion() to " + destination + + " has timed out, but the regions might" + + " already be opened on it.", e); + } + // wait and reset the re-try count, server might be just busy. + Thread.sleep(100); + i--; + continue; + } else if (e instanceof FailedServerException && i < maximumAttempts) { + // In case the server is in the failed server list, no point to + // retry too soon. Retry after the failed_server_expiry time + long sleepTime = 1 + conf.getInt(RpcClient.FAILED_SERVER_EXPIRY_KEY, + RpcClient.FAILED_SERVER_EXPIRY_DEFAULT); + if (LOG.isDebugEnabled()) { + LOG.debug(destination + " is on failed server list; waiting " + + sleepTime + "ms", e); + } + Thread.sleep(sleepTime); + continue; + } + throw e; + } + } + } catch (IOException e) { + // Can be a socket timeout, EOF, NoRouteToHost, etc + LOG.info("Unable to communicate with " + destination + + " in order to assign regions, ", e); + for (RegionState state: states) { + HRegionInfo region = state.getRegion(); + forceRegionStateToOffline(region, true); + } + return false; + } + } finally { + for (Lock lock : locks.values()) { + lock.unlock(); + } + } + + if (!failedToOpenRegions.isEmpty()) { + for (HRegionInfo region : failedToOpenRegions) { + if (!regionStates.isRegionOnline(region)) { + invokeAssign(region); + } + } + } + + // wait for assignment completion + ArrayList<HRegionInfo> userRegionSet = new ArrayList<>(regions.size()); + for (HRegionInfo region: regions) { + if (!region.getTable().isSystemTable()) { + userRegionSet.add(region); + } + } + if (!waitForAssignment(userRegionSet, true, userRegionSet.size(), + System.currentTimeMillis())) { + LOG.debug("some user regions are still in transition: " + userRegionSet); + } + LOG.debug("Bulk assigning done for " + destination); + return true; + } finally { + metricsAssignmentManager.updateBulkAssignTime(EnvironmentEdgeManager.currentTime() - startTime); + } + } + + /** + * Send CLOSE RPC if the server is online, otherwise, offline the region. + * + * The RPC will be sent only to the region sever found in the region state + * if it is passed in, otherwise, to the src server specified. If region + * state is not specified, we don't update region state at all, instead + * we just send the RPC call. This is useful for some cleanup without + * messing around the region states (see handleRegion, on region opened + * on an unexpected server scenario, for an example) + */ + private void unassign(final HRegionInfo region, + final ServerName server, final ServerName dest) { + for (int i = 1; i <= this.maximumAttempts; i++) { + if (this.server.isStopped() || this.server.isAborted()) { + LOG.debug("Server stopped/aborted; skipping unassign of " + region); + return; + } + if (!serverManager.isServerOnline(server)) { + LOG.debug("Offline " + region.getRegionNameAsString() + + ", no need to unassign since it's on a dead server: " + server); + regionStates.updateRegionState(region, State.OFFLINE); + return; + } + try { + // Send CLOSE RPC + if (serverManager.sendRegionClose(server, region, dest)) { + LOG.debug("Sent CLOSE to " + server + " for region " + + region.getRegionNameAsString()); + return; + } + // This never happens. Currently regionserver close always return true. + // Todo; this can now happen (0.96) if there is an exception in a coprocessor + LOG.warn("Server " + server + " region CLOSE RPC returned false for " + + region.getRegionNameAsString()); + } catch (Throwable t) { + long sleepTime = 0; + Configuration conf = this.server.getConfiguration(); + if (t instanceof RemoteException) { + t = ((RemoteException)t).unwrapRemoteException(); + } + if (t instanceof RegionServerAbortedException + || t instanceof RegionServerStoppedException + || t instanceof ServerNotRunningYetException) { + // RS is aborting, we cannot offline the region since the region may need to do WAL + // recovery. Until we see the RS expiration, we should retry. + sleepTime = 1 + conf.getInt(RpcClient.FAILED_SERVER_EXPIRY_KEY, + RpcClient.FAILED_SERVER_EXPIRY_DEFAULT); + + } else if (t instanceof NotServingRegionException) { + LOG.debug("Offline " + region.getRegionNameAsString() + + ", it's not any more on " + server, t); + regionStates.updateRegionState(region, State.OFFLINE); + return; + } else if (t instanceof FailedServerException && i < maximumAttempts) { + // In case the server is in the failed server list, no point to + // retry too soon. Retry after the failed_server_expiry time + sleepTime = 1 + conf.getInt(RpcClient.FAILED_SERVER_EXPIRY_KEY, + RpcClient.FAILED_SERVER_EXPIRY_DEFAULT); + if (LOG.isDebugEnabled()) { + LOG.debug(server + " is on failed server list; waiting " + sleepTime + "ms", t); + } + } + try { + if (sleepTime > 0) { + Thread.sleep(sleepTime); + } + } catch (InterruptedException ie) { + LOG.warn("Interrupted unassign " + region.getRegionNameAsString(), ie); + Thread.currentThread().interrupt(); + regionStates.updateRegionState(region, State.FAILED_CLOSE); + return; + } + LOG.info("Server " + server + " returned " + t + " for " + + region.getRegionNameAsString() + ", try=" + i + + " of " + this.maximumAttempts, t); + } + } + // Run out of attempts + regionStates.updateRegionState(region, State.FAILED_CLOSE); + } + + /** + * Set region to OFFLINE unless it is opening and forceNewPlan is false. + */ + private RegionState forceRegionStateToOffline( + final HRegionInfo region, final boolean forceNewPlan) { + RegionState state = regionStates.getRegionState(region); + if (state == null) { + LOG.warn("Assigning but not in region states: " + region); + state = regionStates.createRegionState(region); + } + + if (forceNewPlan && LOG.isDebugEnabled()) { + LOG.debug("Force region state offline " + state); + } + + switch (state.getState()) { + case OPEN: + case OPENING: + case PENDING_OPEN: + case CLOSING: + case PENDING_CLOSE: + if (!forceNewPlan) { + LOG.debug("Skip assigning " + + region + ", it is already " + state); + return null; + } + case FAILED_CLOSE: + case FAILED_OPEN: + regionStates.updateRegionState(region, State.PENDING_CLOSE); + unassign(region, state.getServerName(), null); + state = regionStates.getRegionState(region); + if (!state.isOffline() && !state.isClosed()) { + // If the region isn't offline, we can't re-assign + // it now. It will be assigned automatically after + // the regionserver reports it's closed. + return null; + } + case OFFLINE: + case CLOSED: + break; + default: + LOG.error("Trying to assign region " + region + + ", which is " + state); + return null; + } + return state; + } + + /** + * Caller must hold lock on the passed <code>state</code> object. + * @param state + * @param forceNewPlan + */ + private void assign(RegionState state, boolean forceNewPlan) { + long startTime = EnvironmentEdgeManager.currentTime(); + try { + Configuration conf = server.getConfiguration(); + RegionPlan plan = null; + long maxWaitTime = -1; + HRegionInfo region = state.getRegion(); + Throwable previousException = null; + for (int i = 1; i <= maximumAttempts; i++) { + if (server.isStopped() || server.isAborted()) { + LOG.info("Skip assigning " + region.getRegionNameAsString() + + ", the server is stopped/aborted"); + return; + } + + if (plan == null) { // Get a server for the region at first + try { + plan = getRegionPlan(region, forceNewPlan); + } catch (HBaseIOException e) { + LOG.warn("Failed to get region plan", e); + } + } + + if (plan == null) { + LOG.warn("Unable to determine a plan to assign " + region); + + // For meta region, we have to keep retrying until succeeding + if (region.isMetaRegion()) { + if (i == maximumAttempts) { + i = 0; // re-set attempt count to 0 for at least 1 retry + + LOG.warn("Unable to determine a plan to assign a hbase:meta region " + region + + " after maximumAttempts (" + this.maximumAttempts + + "). Reset attempts count and continue retrying."); + } + waitForRetryingMetaAssignment(); + continue; + } + + regionStates.updateRegionState(region, State.FAILED_OPEN); + return; + } + LOG.info("Assigning " + region.getRegionNameAsString() + + " to " + plan.getDestination()); + // Transition RegionState to PENDING_OPEN + regionStates.updateRegionState(region, + State.PENDING_OPEN, plan.getDestination()); + + boolean needNewPlan = false; + final String assignMsg = "Failed assignment of " + region.getRegionNameAsString() + + " to " + plan.getDestination(); + try { + List<ServerName> favoredNodes = ServerName.EMPTY_SERVER_LIST; + if (shouldAssignFavoredNodes(region)) { + favoredNodes = server.getFavoredNodesManager().getFavoredNodesWithDNPort(region); + } + serverManager.sendRegionOpen(plan.getDestination(), region, favoredNodes); + return; // we're done + } catch (Throwable t) { + if (t instanceof RemoteException) { + t = ((RemoteException) t).unwrapRemoteException(); + } + previousException = t; + + // Should we wait a little before retrying? If the server is starting it's yes. + boolean hold = (t instanceof ServerNotRunningYetException); + + // In case socket is timed out and the region server is still online, + // the openRegion RPC could have been accepted by the server and + // just the response didn't go through. So we will retry to + // open the region on the same server. + boolean retry = !hold && (t instanceof java.net.SocketTimeoutException + && this.serverManager.isServerOnline(plan.getDestination())); + + if (hold) { + LOG.warn(assignMsg + ", waiting a little before trying on the same region server " + + "try=" + i + " of " + this.maximumAttempts, t); + + if (maxWaitTime < 0) { + maxWaitTime = EnvironmentEdgeManager.currentTime() + + this.server.getConfiguration().getLong( + "hbase.regionserver.rpc.startup.waittime", 60000); + } + try { + long now = EnvironmentEdgeManager.currentTime(); + if (now < maxWaitTime) { + if (LOG.isDebugEnabled()) { + LOG.debug("Server is not yet up; waiting up to " + + (maxWaitTime - now) + "ms", t); + } + Thread.sleep(100); + i--; // reset the try count + } else { + LOG.debug("Server is not up for a while; try a new one", t); + needNewPlan = true; + } + } catch (InterruptedException ie) { + LOG.warn("Failed to assign " + + region.getRegionNameAsString() + " since interrupted", ie); + regionStates.updateRegionState(region, State.FAILED_OPEN); + Thread.currentThread().interrupt(); + return; + } + } else if (retry) { + i--; // we want to retry as many times as needed as long as the RS is not dead. + if (LOG.isDebugEnabled()) { + LOG.debug(assignMsg + ", trying to assign to the same region server due ", t); + } + } else { + needNewPlan = true; + LOG.warn(assignMsg + ", trying to assign elsewhere instead;" + + " try=" + i + " of " + this.maximumAttempts, t); + } + } + + if (i == this.maximumAttempts) { + // For meta region, we have to keep retrying until succeeding + if (region.isMetaRegion()) { + i = 0; // re-set attempt count to 0 for at least 1 retry + LOG.warn(assignMsg + + ", trying to assign a hbase:meta region reached to maximumAttempts (" + + this.maximumAttempts + "). Reset attempt counts and continue retrying."); + waitForRetryingMetaAssignment(); + } + else { + // Don't reset the region state or get a new plan any more. + // This is the last try. + continue; + } + } + + // If region opened on destination of present plan, reassigning to new + // RS may cause double assignments. In case of RegionAlreadyInTransitionException + // reassigning to same RS. + if (needNewPlan) { + // Force a new plan and reassign. Will return null if no servers. + // The new plan could be the same as the existing plan since we don't + // exclude the server of the original plan, which should not be + // excluded since it could be the only server up now. + RegionPlan newPlan = null; + try { + newPlan = getRegionPlan(region, true); + } catch (HBaseIOException e) { + LOG.warn("Failed to get region plan", e); + } + if (newPlan == null) { + regionStates.updateRegionState(region, State.FAILED_OPEN); + LOG.warn("Unable to find a viable location to assign region " + + region.getRegionNameAsString()); + return; + } + + if (plan != newPlan && !plan.getDestination().equals(newPlan.getDestination())) { + // Clean out plan we failed execute and one that doesn't look like it'll + // succeed anyways; we need a new plan! + // Transition back to OFFLINE + regionStates.updateRegionState(region, State.OFFLINE); + plan = newPlan; + } else if(plan.getDestination().equals(newPlan.getDestination()) && + previousException instanceof FailedServerException) { + try { + LOG.info("Trying to re-assign " + region.getRegionNameAsString() + + " to the same failed server."); + Thread.sleep(1 + conf.getInt(RpcClient.FAILED_SERVER_EXPIRY_KEY, + RpcClient.FAILED_SERVER_EXPIRY_DEFAULT)); + } catch (InterruptedException ie) { + LOG.warn("Failed to assign " + + region.getRegionNameAsString() + " since interrupted", ie); + regionStates.updateRegionState(region, State.FAILED_OPEN); + Thread.currentThread().interrupt(); + return; + } + } + } + } + // Run out of attempts + regionStates.updateRegionState(region, State.FAILED_OPEN); + } finally { + metricsAssignmentManager.updateAssignmentTime(EnvironmentEdgeManager.currentTime() - startTime); + } + } + + private boolean isDisabledorDisablingRegionInRIT(final HRegionInfo region) { + if (this.tableStateManager.isTableState(region.getTable(), + TableState.State.DISABLED, + TableState.State.DISABLING) || replicasToClose.contains(region)) { + LOG.info("Table " + region.getTable() + " is disabled or disabling;" + + " skipping assign of " + region.getRegionNameAsString()); + offlineDisabledRegion(region); + return true; + } + return false; + } + + /** + * @param region the region to assign + * @param forceNewPlan If true, then if an existing plan exists, a new plan + * will be generated. + * @return Plan for passed <code>region</code> (If none currently, it creates one or + * if no servers to assign, it returns null). + */ + private RegionPlan getRegionPlan(final HRegionInfo region, + final boolean forceNewPlan) throws HBaseIOException { + // Pickup existing plan or make a new one + final String encodedName = region.getEncodedName(); + final List<ServerName> destServers = + serverManager.createDestinationServersList(); + + if (destServers.isEmpty()){ + LOG.warn("Can't move " + encodedName + + ", there is no destination server available."); + return null; + } + + RegionPlan randomPlan = null; + boolean newPlan = false; + RegionPlan existingPlan; + + synchronized (this.regionPlans) { + existingPlan = this.regionPlans.get(encodedName); + + if (existingPlan != null && existingPlan.getDestination() != null) { + LOG.debug("Found an existing plan for " + region.getRegionNameAsString() + + " destination server is " + existingPlan.getDestination() + + " accepted as a dest server = " + destServers.contains(existingPlan.getDestination())); + } + + if (forceNewPlan + || existingPlan == null + || existingPlan.getDestination() == null + || !destServers.contains(existingPlan.getDestination())) { + newPlan = true; + try { + randomPlan = new RegionPlan(region, null, + balancer.randomAssignment(region, destServers)); + } catch (IOException ex) { + LOG.warn("Failed to create new plan.",ex); + return null; + } + this.regionPlans.put(encodedName, randomPlan); + } + } + + if (newPlan) { + if (randomPlan.getDestination() == null) { + LOG.warn("Can't find a destination for " + encodedName); + return null; + } + if (LOG.isDebugEnabled()) { + LOG.debug("No previous transition plan found (or ignoring " + + "an existing plan) for " + region.getRegionNameAsString() + + "; generated random plan=" + randomPlan + "; " + destServers.size() + + " (online=" + serverManager.getOnlineServers().size() + + ") available servers, forceNewPlan=" + forceNewPlan); + } + return randomPlan; + } + if (LOG.isDebugEnabled()) { + LOG.debug("Using pre-existing plan for " + + region.getRegionNameAsString() + "; plan=" + existingPlan); + } + return existingPlan; + } + + /** + * Wait for some time before retrying meta table region assignment + */ + private void waitForRetryingMetaAssignment() { + try { + Thread.sleep(this.sleepTimeBeforeRetryingMetaAssignment); + } catch (InterruptedException e) { + LOG.error("Got exception while waiting for hbase:meta assignment"); + Thread.currentThread().interrupt(); + } + } + + /** + * Unassigns the specified region. + * <p> + * Updates the RegionState and sends the CLOSE RPC unless region is being + * split by regionserver; then the unassign fails (silently) because we + * presume the region being unassigned no longer exists (its been split out + * of existence). TODO: What to do if split fails and is rolled back and + * parent is revivified? + * <p> + * If a RegionPlan is already set, it will remain. + * + * @param region server to be unassigned + */ + public void unassign(HRegionInfo region) { + unassign(region, null); + } + + + /** + * Unassigns the specified region. + * <p> + * Updates the RegionState and sends the CLOSE RPC unless region is being + * split by regionserver; then the unassign fails (silently) because we + * presume the region being unassigned no longer exists (its been split out + * of existence). TODO: What to do if split fails and is rolled back and + * parent is revivified? + * <p> + * If a RegionPlan is already set, it will remain. + * + * @param region server to be unassigned + * @param dest the destination server of the region + */ + public void unassign(HRegionInfo region, ServerName dest) { + // TODO: Method needs refactoring. Ugly buried returns throughout. Beware! + LOG.debug("Starting unassign of " + region.getRegionNameAsString() + + " (offlining), current state: " + regionStates.getRegionState(region)); + + String encodedName = region.getEncodedName(); + // Grab the state of this region and synchronize on it + // We need a lock here as we're going to do a put later and we don't want multiple states + // creation + ReentrantLock lock = locker.acquireLock(encodedName); + RegionState state = regionStates.getRegionTransitionState(encodedName); + try { + if (state == null || state.isFailedClose()) { + if (state == null) { + // Region is not in transition. + // We can unassign it only if it's not SPLIT/MERGED. + state = regionStates.getRegionState(encodedName); + if (state != null && state.isUnassignable()) { + LOG.info("Attempting to unassign " + state + ", ignored"); + // Offline region will be reassigned below + return; + } + if (state == null || state.getServerName() == null) { + // We don't know where the region is, offline it. + // No need to send CLOSE RPC + LOG.warn("Attempting to unassign a region not in RegionStates " + + region.getRegionNameAsString() + ", offlined"); + regionOffline(region); + return; + } + } + state = regionStates.updateRegionState( + region, State.PENDING_CLOSE); + } else if (state.isFailedOpen()) { + // The region is not open yet + regionOffline(region); + return; + } else { + LOG.debug("Attempting to unassign " + + region.getRegionNameAsString() + " but it is " + + "already in transition (" + state.getState()); + return; + } + + unassign(region, state.getServerName(), dest); + } finally { + lock.unlock(); + + // Region is expected to be reassigned afterwards + if (!replicasToClose.contains(region) + && regionStates.isRegionInState(region, State.OFFLINE)) { + assign(region); + } + } + } + + /** + * Used by unit tests. Return the number of regions opened so far in the life + * of the master. Increases by one every time the master opens a region + * @return the counter value of the number of regions opened so far + */ + public int getNumRegionsOpened() { + return numRegionsOpened.get(); + } + + /** + * Waits until the specified region has completed assignment. + * <p> + * If the region is already assigned, returns immediately. Otherwise, method + * blocks until the region is assigned. + * @param regionInfo region to wait on assignment for + * @return true if the region is assigned false otherwise. + * @throws InterruptedException + */ + public boolean waitForAssignment(HRegionInfo regionInfo) + throws InterruptedException { + ArrayList<HRegionInfo> regionSet = new ArrayList<>(1); + regionSet.add(regionInfo); + return waitForAssignment(regionSet, true, Long.MAX_VALUE); + } + + /** + * Waits until the specified region has completed assignment, or the deadline is reached. + */ + protected boolean waitForAssignment(final Collection<HRegionInfo> regionSet, + final boolean waitTillAllAssigned, final int reassigningRegions, + final long minEndTime) throws InterruptedException { + long deadline = minEndTime + bulkPerRegionOpenTimeGuesstimate * (reassigningRegions + 1); + if (deadline < 0) { // Overflow + deadline = Long.MAX_VALUE; // wait forever + } + return waitForAssignment(regionSet, waitTillAllAssigned, deadline); + } + + /** + * Waits until the specified region has completed assignment, or the deadline is reached. + * @param regionSet set of region to wait on. the set is modified and the assigned regions removed + * @param waitTillAllAssigned true if we should wait all the regions to be assigned + * @param deadline the timestamp after which the wait is aborted + * @return true if all the regions are assigned false otherwise. + * @throws InterruptedException + */ + protected boolean waitForAssignment(final Collection<HRegionInfo> regionSet, + final boolean waitTillAllAssigned, final long deadline) throws InterruptedException { + // We're not synchronizing on regionsInTransition now because we don't use any iterator. + while (!regionSet.isEmpty() && !server.isStopped() && deadline > System.currentTimeMillis()) { + int failedOpenCount = 0; + Iterator<HRegionInfo> regionInfoIterator = regionSet.iterator(); + while (regionInfoIterator.hasNext()) { + HRegionInfo hri = regionInfoIterator.next(); + if (regionStates.isRegionOnline(hri) || regionStates.isRegionInState(hri, + State.SPLITTING, State.SPLIT, State.MERGING, State.MERGED)) { + regionInfoIterator.remove(); + } else if (regionStates.isRegionInState(hri, State.FAILED_OPEN)) { + failedOpenCount++; + } + } + if (!waitTillAllAssigned) { + // No need to wait, let assignment going on asynchronously + break; + } + if (!regionSet.isEmpty()) { + if (failedOpenCount == regionSet.size()) { + // all the regions we are waiting had an error on open. + break; + } + regionStates.waitForUpdate(100); + } + } + return regionSet.isEmpty(); + } + + /** + * Assigns the hbase:meta region or a replica. + * <p> + * Assumes that hbase:meta is currently closed and is not being actively served by + * any RegionServer. + * @param hri TODO + */ + public void assignMeta(HRegionInfo hri) throws KeeperException { + regionStates.updateRegionState(hri, State.OFFLINE); + assign(hri); + } + + /** + * Assigns specified regions retaining assignments, if any. + * <p> + * This is a synchronous call and will return once every region has been + * assigned. If anything fails, an exception is thrown + * @throws InterruptedException + * @throws IOException + */ + public void assign(Map<HRegionInfo, ServerName> regions) + throws IOException, InterruptedException { + if (regions == null || regions.isEmpty()) { + return; + } + List<ServerName> servers = serverManager.createDestinationServersList(); + if (servers == null || servers.isEmpty()) { + throw new IOException("Found no destination server to assign region(s)"); + } + + // Reuse existing assignment info + Map<ServerName, List<HRegionInfo>> bulkPlan = + balancer.retainAssignment(regions, servers); + if (bulkPlan == null) { + throw new IOException("Unable to determine a plan to assign region(s)"); + } + + processBogusAssignments(bulkPlan); + + assign(regions.size(), servers.size(), + "retainAssignment=true", bulkPlan); + } + + /** + * Assigns specified regions round robin, if any. + * <p> + * This is a synchronous call and will return once every region has been + * assigned. If anything fails, an exception is thrown + * @throws InterruptedException + * @throws IOException + */ + public void assign(List<HRegionInfo> regions) + throws IOException, InterruptedException { + if (regions == null || regions.isEmpty()) { + return; + } + + List<ServerName> servers = serverManager.createDestinationServersList(); + if (servers == null || servers.isEmpty()) { + throw new IOException("Found no destination server to assign region(s)"); + } + + // Generate a round-robin bulk assignment plan + Map<ServerName, List<HRegionInfo>> bulkPlan = balancer.roundRobinAssignment(regions, servers); + if (bulkPlan == null) { + throw new IOException("Unable to determine a plan to assign region(s)"); + } + + processBogusAssignments(bulkPlan); + + assign(regions.size(), servers.size(), "round-robin=true", bulkPlan); + } + + private void assign(int regions, int totalServers, + String message, Map<ServerName, List<HRegionInfo>> bulkPlan) + throws InterruptedException, IOException { + + int servers = bulkPlan.size(); + if (servers == 1 || (regions < bulkAssignThresholdRegions + && servers < bulkAssignThresholdServers)) { + + // Not use bulk assignment. This could be more efficient in small + // cluster, especially mini cluster for testing, so that tests won't time out + if (LOG.isTraceEnabled()) { + LOG.trace("Not using bulk assignment since we are assigning only " + regions + + " region(s) to " + servers + " server(s)"); + } + + // invoke assignment (async) + ArrayList<HRegionInfo> userRegionSet = new ArrayList<>(regions); + for (Map.Entry<ServerName, List<HRegionInfo>> plan: bulkPlan.entrySet()) { + if (!assign(plan.getKey(), plan.getValue()) && !server.isStopped()) { + for (HRegionInfo region: plan.getValue()) { + if (!regionStates.isRegionOnline(region)) { + invokeAssign(region); + if (!region.getTable().isSystemTable()) { + userRegionSet.add(region); + } + } + } + } + } + + // wait for assignment completion + if (!waitForAssignment(userRegionSet, true, userRegionSet.size(), + System.currentTimeMillis())) { + LOG.debug("some user regions are still in transition: " + userRegionSet); + } + } else { + LOG.info("Bulk assigning " + regions + " region(s) across " + + totalServers + " server(s), " + message); + + // Use fixed count thread pool assigning. + BulkAssigner ba = new GeneralBulkAssigner( + this.server, bulkPlan, this, bulkAssignWaitTillAllAssigned); + ba.bulkAssign(); + LOG.info("Bulk assigning done"); + } + } + + /** + * 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 + */ + private void assignAllUserRegions(Map<HRegionInfo, ServerName> allRegions) + throws IOException, InterruptedException { + 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); + + Set<HRegionInfo> regionsFromMetaScan = allRegions.keySet(); + if (retainAssignment) { + assign(allRegions); + } else { + List<HRegionInfo> regions = new ArrayList<>(regionsFromMetaScan); + assign(regions); + } + + for (HRegionInfo hri : regionsFromMetaScan) { + TableName tableName = hri.getTable(); + if (!tableStateManager.isTableState(tableName, + TableState.State.ENABLED)) { + setEnabledTable(tableName); + } + } + // assign all the replicas that were not recorded in the meta + assign(replicaRegionsNotRecordedInMeta(regionsFromMetaScan, (MasterServices)server)); + } + + /** + * Get number of replicas of a table + */ + private static int getNumReplicas(MasterServices master, TableName table) { + int numReplica = 1; + try { + HTableDescriptor htd = master.getTableDescriptors().get(table); + if (htd == null) { + LOG.warn("master can not get TableDescriptor from table '" + table); + } else { + numReplica = htd.getRegionReplication(); + } + } catch (IOException e){ + LOG.warn("Couldn't get the replication attribute of the table " + table + " due to " + + e.getMessage()); + } + return numReplica; + } + + /** + * Get a list of replica regions that are: + * not recorded in meta yet. We might not have recorded the locations + * for the replicas since the replicas may not have been online yet, master restarted + * in the middle of assigning, ZK erased, etc. + * @param regionsRecordedInMeta the list of regions we know are recorded in meta + * either as a default, or, as the location of a replica + * @param master + * @return list of replica regions + * @throws IOException + */ + public static List<HRegionInfo> replicaRegionsNotRecordedInMeta( + Set<HRegionInfo> regionsRecordedInMeta, MasterServices master)throws IOException { + List<HRegionInfo> regionsNotRecordedInMeta = new ArrayList<>(); + for (HRegionInfo hri : regionsRecordedInMeta) { + TableName table = hri.getTable(); + if(master.getTableDescriptors().get(table) == null) + continue; + int desiredRegionReplication = getNumReplicas(master, table); + for (int i = 0; i < desiredRegionReplication; i++) { + HRegionInfo replica = RegionReplicaUtil.getRegionInfoForReplica(hri, i); + if (regionsRecordedInMeta.contains(replica)) continue; + regionsNotRecordedInMeta.add(replica); + } + } + return regionsNotRecordedInMeta; + } + + /** + * Rebuild the list of user regions and assignment information. + * Updates regionstates with findings as we go through list of regions. + * @return set of servers not online that hosted some regions according to a scan of hbase:meta + * @throws IOException + */ + Set<ServerName> rebuildUserRegions() throws + IOException, KeeperException { + Set<TableName> disabledOrEnablingTables = tableStateManager.getTablesInStates( + TableState.State.DISABLED, TableState.State.ENABLING); + + Set<TableName> disabledOrDisablingOrEnabling = tableStateManager.getTablesInStates( + TableState.State.DISABLED, + TableState.State.DISABLING, + TableState.State.ENABLING); + + // Region assignment from META + List<Result> results = MetaTableAccessor.fullScanRegions(server.getConnection()); + // Get any new but slow to checkin region server that joined the cluster + Set<ServerName> onlineServers = serverManager.getOnlineServers().keySet(); + // Set of offline servers to be returned + Set<ServerName> offlineServers = new HashSet<>(); + // Iterate regions in META + for (Result result : results) { + if (result == null && LOG.isDebugEnabled()){ + LOG.debug("null result from meta - ignoring but this is strange."); + continue; + } + // keep a track of replicas to close. These were the replicas of the originally + // unmerged regions. The master might have closed them before but it mightn't + // maybe because it crashed. + PairOfSameType<HRegionInfo> p = MetaTableAccessor.getMergeRegions(result); + if (p.getFirst() != null && p.getSecond() != null) { + int numReplicas = getNumReplicas(server, p.getFirst().getTable()); + for (HRegionInfo merge : p) { + for (int i = 1; i < numReplicas; i++) { + replicasToClose.add(RegionReplicaUtil.getRegionInfoForReplica(merge, i)); + } + } + } + RegionLocations rl = MetaTableAccessor.getRegionLocations(result); + if (rl == null) { + continue; + } + HRegionLocation[] locations = rl.getRegionLocations(); + if (locations == null) { + continue; + } + for (HRegionLocation hrl : locations) { + if (hrl == null) continue; + HRegionInfo regionInfo = hrl.getRegionInfo(); + if (regionInfo == null) continue; + int replicaId = regionInfo.getReplicaId(); + State state = RegionStateStore.getRegionState(result, replicaId); + // keep a track of replicas to close. These were the replicas of the split parents + // from the previous life of the master. The master should have closed them before + // but it couldn't maybe because it crashed + if (replicaId == 0 && state.equals(State.SPLIT)) { + for (HRegionLocation h : locations) { + replicasToClose.add(h.getRegionInfo()); + } + } + ServerName lastHost = hrl.getServerName(); + ServerName regionLocation = RegionStateStore.getRegionServer(result, replicaId); + regionStates.createRegionState(regionInfo, state, regionLocation, lastHost); + if (!regionStates.isRegionInState(regionInfo, State.OPEN)) { + // Region is not open (either offline or in transition), skip + continue; + } + TableName tableName = regionInfo.getTable(); + if (!onlineServers.contains(regionLocation)) { + // Region is located on a server that isn't online + offlineServers.add(regionLocation); + } else if (!disabledOrEnablingTables.contains(tableName)) { + // Region is being served and on an active server + // add only if region not in disabled or enabling table + regionStates.regionOnline(regionInfo, regionLocation); + balancer.regionOnline(regionInfo, regionLocation); + } + // need to enable the table if not disabled or disabling or enabling + // this will be used in rolling restarts + if (!disabledOrDisablingOrEnabling.contains(tableName) + && !getTableStateManager().isTableState(tableName, + TableState.State.ENABLED)) { + setEnabledTable(tableName); + } + } + } + return offlineServers; + } + + /** + * Processes list of regions in transition at startup + */ + void processRegionsInTransition(Collection<RegionState> regionsInTransition) { + // We need to send RPC call again for PENDING_OPEN/PENDING_CLOSE regions + // in case the RPC call is not sent out yet before the master was shut down + // since we update the state before we send the RPC call. We can't update + // the state after the RPC call. Otherwise, we don't know what's happened + // to the region if the master dies right after the RPC call is out. + for (RegionState regionState: regionsInTransition) { + LOG.info("Processing " + regionState); + ServerName serverName = regionState.getServerName(); + // Server could be null in case of FAILED_OPEN when master cannot find a region plan. In that + // case, try assigning it here. + if (serverName != null && !serverManager.getOnlineServers().containsKey(serverName)) { + LOG.info("Server " + serverName + " isn't online. SSH will handle this"); + continue; // SSH will handle it + } + HRegionInfo regionInfo = regionState.getRegion(); + RegionState.State state = regionState.getState(); + switch (state) { + case CLOSED: + invokeAssign(regionState.getRegion()); + break; + case PENDING_OPEN: + retrySendRegionOpen(regionState); + break; + case PENDING_CLOSE: + retrySendRegionClose(regionState); + break; + case FAILED_CLOSE: + case FAILED_OPEN: + invokeUnAssign(regionInfo); + break; + default: + // No process for other states + break; + } + } + } + + /** + * At master failover, for pending_open region, make sure + * sendRegionOpen RPC call is sent to the target regionserver + */ + private void retrySendRegionOpen(final RegionState regionState) { + this.executorService.submit( + new EventHandler(server, EventType.M_MASTER_RECOVERY) { + @Override + public void process() throws IOException { + HRegionInfo hri = regionState.getRegion(); + ServerName serverName = regionState.getServerName(); + ReentrantLock lock = locker.acquireLock(hri.getEncodedName()); + try { + for (int i = 1; i <= maximumAttempts; i++) { + if (!serverManager.isServerOnline(serverName) + || server.isStopped() || server.isAborted()) { + return; // No need any more + } + try { + if (!regionState.equals(regionStates.getRegionState(hri))) { + return; // Region is not in the expected state any more + } + List<ServerName> favoredNodes = ServerName.EMPTY_SERVER_LIST; + if (shouldAssignFavoredNodes(hri)) { + FavoredNodesManager fnm = ((MasterServices)server).getFavoredNodesManager(); + favoredNodes = fnm.getFavoredNodesWithDNPort(hri); + } + serverManager.sendRegionOpen(serverName, hri, favoredNodes); + return; // we're done + } catch (Throwable t) { + if (t instanceof RemoteException) { + t = ((RemoteException) t).unwrapRemoteException(); + } + if (t instanceof FailedServerException && i < maximumAttempts) { + // In case the server is in the failed server list, no point to + // retry too soon. Retry after the failed_server_expiry time + try { + Configuration conf = this.server.getConfiguration(); + long sleepTime = 1 + conf.getInt(RpcClient.FAILED_SERVER_EXPIRY_KEY, + RpcClient.FAILED_SERVER_EXPIRY_DEFAULT); + if (LOG.isDebugEnabled()) { + LOG.debug(serverName + " is on failed server list; waiting " + + sleepTime + "ms", t); + } + Thread.sleep(sleepTime); + continue; + } catch (InterruptedException ie) { + LOG.warn("Failed to assign " + + hri.getRegionNameAsString() + " since interrupted", ie); + regionStates.updateRegionState(hri, State.FAILED_OPEN); + Thread.currentThread().interrupt(); + return; + } + } + if (serverManager.isServerOnline(serverName) + && t instanceof java.net.SocketTimeoutException) { + i--; // reset the try count + } else { + LOG.info("Got exception in retrying sendRegionOpen for " + + regionState + "; try=" + i + " of " + maximumAttempts, t); + } + Threads.sleep(100); + } + } + // Run out of attempts + regionStates.updateRegionState(hri, State.FAILED_OPEN); + } finally { + lock.unlock(); + } + } + }); + } + + /** + * At master failover, for pending_close region, make sure + * sendRegionClose RPC call is sent to the target regionserver + */ + private void retrySendRegionClose(final RegionState regionState) { + this.executorService.submit( + new EventHandler(server, EventType.M_MASTER_RECOVERY) { + @Override + public void process() throws IOException { + HRegionInfo hri = regionState.getRegion(); + ServerName serverName = regionState.getServerName(); + ReentrantLock lock = locker.acquireLock(hri.getEncodedName()); + try { + for (int i = 1; i <= maximumAttempts; i++) { + if (!serverManager.isServerOnline(serverName) + || server.isStopped() || server.isAborted()) { + return; // No need any more + } + try { + if (!regionState.equals(regionStates.getRegionState(hri))) { + return; // Region is not in the expected state any more + } + serverManager.sendRegionClose(serverName, hri, null); + return; // Done. + } catch (Throwable t) { + if (t instanceof RemoteException) { + t = ((RemoteException) t).unwrapRemoteException(); + } + if (t instanceof FailedServerException && i < maximumAttempts) { + // In case the server is in the failed server list, no point to + // retry too soon. Retry after the failed_server_expiry time + try { + Configuration conf = this.server.getConfiguration(); + long sleepTime = 1 + conf.getInt(RpcClient.FAILED_SERVER_EXPIRY_KEY, + RpcClient.FAILED_SERVER_EXPIRY_DEFAULT); + if (LOG.isDebugEnabled()) { + LOG.debug(serverName + " is on failed server list; waiting " + + sleepTime + "ms", t); + } + Thread.sleep(sleepTime); + continue; + } catch (InterruptedException ie) { + LOG.warn("Failed to unassign " + + hri.getRegionNameAsString() + " since interrupted", ie); + regionStates.updateRegionState(hri, RegionState.State.FAILED_CLOSE); + Thread.currentThread().interrupt(); + return; + } + } + if (serverManager.isServerOnline(serverName) + && t instanceof java.net.SocketTimeoutException) { + i--; // reset the try count + } else { + LOG.info("Got exception in retrying sendRegionClose for " + + regionState + "; try=" + i + " of " + maximumAttempts, t); + } + Threads.sleep(100); + } + } + // Run out of attempts + regionStates.updateRegionState(hri, State.FAILED_CLOSE); + } finally { + lock.unlock(); + } + } + }); + } + + /** + * Set Regions in transitions metrics. + * This takes an iterator on the RegionInTransition map (CLSM), and is not synchronized. + * This iterator is not fail fast, which may lead to stale read; but that's better than + * creating a copy of the map for metrics computation, as this method will be invoked + * on a frequent interval. + */ + public void updateRegionsInTransitionMetrics() { + long currentTime = System.currentTimeMillis(); + int totalRITs = 0; + int totalRITsOverThreshold = 0; + long oldestRITTime = 0; + int ritThreshold = this.server.getConfiguration(). + getInt(HConstants.METRICS_RIT_STUCK_WARNING_THRESHOLD, 60000); + for (RegionState state: regionStates.getRegionsInTransition()) { + totalRITs++; + long ritTime = currentTime - state.getStamp(); + if (ritTime > ritThreshold) { // more than the threshold + totalRITsOverThreshold++; + } + if (oldestRITTime < ritTime) { + oldestRITTime = ritTime; + } + } + if (this.metricsAssignmentManager != null) { + this.metricsAssignmentManager.updateRITOldestAge(oldestRITTime); + this.metricsAssignmentManager.updateRITCount(totalRITs); + this.metricsAssignmentManager.updateRITCountOverThreshold(totalRITsOverThreshold); + } + } + + /** + * @param region Region whose plan we are to clear. + */ + private void clearRegionPlan(final HRegionInfo region) { + synchronized (this.regionPlans) { + this.regionPlans.remove(region.getEncodedName()); + } + } + + /** + * Wait on region to clear regions-in-transition. + * @param hri Region to wait on. + * @throws IOException + */ + public void waitOnRegionToClearRegionsInTransition(final HRegionInfo hri) + throws IOException, InterruptedException { + waitOnRegionToClearRegionsInTransition(hri, -1L); + } + + /** + * Wait on region to clear regions-in-transition or time out + * @param hri + * @param timeOut Milliseconds to wait for current region to be out of transition state. + * @return True when a region clears regions-in-transition before timeout otherwise false + * @throws InterruptedException + */ + public boolean waitOnRegionToClearRegionsInTransition(final HRegionInfo hri, long timeOut) + throws InterruptedException { + if (!regionStates.isRegionInTransition(hri)) { + return true; + } + long end = (timeOut <= 0) ? Long.MAX_VALUE : EnvironmentEdgeManager.currentTime() + + timeOut; + // There is already a timeout monitor on regions in transition so I + // should not have to have one here too? + LOG.info("Waiting for " + hri.getEncodedName() + + " to leave regions-in-transition, timeOut=" + timeOut + " ms."); + while (!this.server.isStopped() && regionStates.isRegionInTransition(hri)) { + regionStates.waitForUpdate(100); + if (EnvironmentEdgeManager.currentTime() > end) { + LOG.info("Timed out on waiting for " + hri.getEncodedName() + " to be assigned."); + return false; + } + } + if (this.server.isStopped()) { + LOG.info("Giving up wait on regions in transition because stoppable.isStopped is set"); + return false; + } + return true; + } + + void invokeAssign(HRegionInfo regionInfo) { + threadPoolExecutorService.submit(new AssignCallable(this, regionInfo)); + } + + void invokeAssignLater(HRegionInfo regionInfo, long sleepMillis) { + scheduledThreadPoolExecutor.schedule(new DelayedAssignCallable( + new AssignCallable(this, regionInfo)), sleepMillis, TimeUnit.MILLISECONDS); + } + + void invokeUnAssign(HRegionInfo regionInfo) { + threadPoolExecutorService.submit(new UnAssignCallable(this, regionInfo)); + } + + public boolean isCarryingMeta(ServerName serverName) { + return isCarryingRegion(serverName, HRegionInfo.FIRST_META_REGIONINFO); + } + + public boolean isCarryingMetaReplica(ServerName serverName, int replicaId) { + return isCarryingRegion(serverName, + RegionReplicaUtil.getRegionInfoForReplica(HRegionInfo.FIRST_META_REGIONINFO, replicaId)); + } + + public boolean isCarryingMetaReplica(ServerName serverName, HRegionInfo metaHri) { + return isCarryingRegion(serverName, metaHri); + } + + /** + * Check if the shutdown server carries the specific region. + * @return whether the serverName currently hosts the region + */ + private boolean isCarryingRegion(ServerName serverName, HRegionInfo hri) { + RegionState regionState = regionStates.getRegionTransitionState(hri); + ServerName transitionAddr = regionState != null? regionState.getServerName(): null; + if (transitionAddr != null) { + boolean matchTransitionAddr = transitionAddr.equals(serverName); + LOG.debug("Checking region=" + hri.getRegionNameAsString() + + ", transitioning on server=" + matchTransitionAddr + + " server being checked: " + serverName + + ", matches=" + matchTransitionAddr); + return matchTransitionAddr; + } + + ServerName assignedAddr = regionStates.getRegionServerOfRegion(hri); + boolean matchAssignedAddr = serverName.equals(assignedAddr); + LOG.debug("based on AM, current region=" + hri.getRegionNameAsString() + + " is on server=" + assignedAddr + ", server being checked: " + + serverName); + return matchAssignedAddr; + } + + /** + * Clean out crashed server removing any assignments. + * @param sn Server that went down. + * @return list of regions in transition on this server + */ + public List<HRegionInfo> cleanOutCrashedServerReferences(final ServerName sn) { + // Clean out any existing assignment plans for this server + synchronized (this.regionPlans) { + for (Iterator <Map.Entry<String, RegionPlan>> i = this.regionPlans.entrySet().iterator(); + i.hasNext();) { + Map.Entry<String, RegionPlan> e = i.next(); + ServerName otherSn = e.getValue().getDestination(); + // The name will be null if the region is planned for a random assign. + if (otherSn != null && otherSn.equals(sn)) { + // Use iterator's remove else we'll get CME + i.remove(); + } + } + } + List<HRegionInfo> rits = regionStates.serverOffline(sn); + for (Iterator<HRegionInfo> it = rits.iterator(); it.hasNext(); ) { + HRegionInfo hri = it.next(); + String encodedName = hri.getEncodedName(); + + // We need a lock on the region as we could update it + Lock lock = locker.acquireLock(encodedName); + try { + RegionState regionState = regionStates.getRegionTransitionState(encodedName); + if (regionState == null + || (regionState.getServerName() != null && !regionState.isOnServer(sn)) + || !RegionStates.isOneOfStates(regionState, State.PENDING_OPEN, + State.OPENING, State.FAILED_OPEN, State.FAILED_CLOSE, State.OFFLINE)) { + LOG.info("Skip " + regionState + " since it is not opening/failed_close" + + " on the dead server any more: " + sn); + it.remove(); + } else { + if (tab
<TRUNCATED>
