View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  package org.apache.hadoop.hbase.master;
20  
21  import java.io.IOException;
22  import java.net.InetAddress;
23  import java.util.ArrayList;
24  import java.util.Collections;
25  import java.util.HashMap;
26  import java.util.HashSet;
27  import java.util.Iterator;
28  import java.util.List;
29  import java.util.Map;
30  import java.util.Map.Entry;
31  import java.util.Set;
32  import java.util.SortedMap;
33  import java.util.concurrent.ConcurrentHashMap;
34  import java.util.concurrent.ConcurrentSkipListMap;
35  import java.util.concurrent.CopyOnWriteArrayList;
36  
37  import org.apache.commons.logging.Log;
38  import org.apache.commons.logging.LogFactory;
39  import org.apache.hadoop.classification.InterfaceAudience;
40  import org.apache.hadoop.conf.Configuration;
41  import org.apache.hadoop.hbase.ClockOutOfSyncException;
42  import org.apache.hadoop.hbase.HRegionInfo;
43  import org.apache.hadoop.hbase.RegionLoad;
44  import org.apache.hadoop.hbase.Server;
45  import org.apache.hadoop.hbase.ServerLoad;
46  import org.apache.hadoop.hbase.ServerName;
47  import org.apache.hadoop.hbase.YouAreDeadException;
48  import org.apache.hadoop.hbase.ZooKeeperConnectionException;
49  import org.apache.hadoop.hbase.client.HConnection;
50  import org.apache.hadoop.hbase.client.HConnectionManager;
51  import org.apache.hadoop.hbase.client.RetriesExhaustedException;
52  import org.apache.hadoop.hbase.master.balancer.BaseLoadBalancer;
53  import org.apache.hadoop.hbase.master.handler.MetaServerShutdownHandler;
54  import org.apache.hadoop.hbase.master.handler.ServerShutdownHandler;
55  import org.apache.hadoop.hbase.monitoring.MonitoredTask;
56  import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
57  import org.apache.hadoop.hbase.protobuf.RequestConverter;
58  import org.apache.hadoop.hbase.protobuf.ResponseConverter;
59  import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.AdminService;
60  import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.OpenRegionRequest;
61  import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.OpenRegionResponse;
62  import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.ServerInfo;
63  import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos.SplitLogTask.RecoveryMode;
64  import org.apache.hadoop.hbase.regionserver.HRegionServer;
65  import org.apache.hadoop.hbase.regionserver.RegionOpeningState;
66  import org.apache.hadoop.hbase.util.Bytes;
67  import org.apache.hadoop.hbase.util.Triple;
68  import org.apache.hadoop.hbase.zookeeper.ZKUtil;
69  import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
70  import org.apache.zookeeper.KeeperException;
71  
72  import com.google.common.annotations.VisibleForTesting;
73  import com.google.protobuf.ServiceException;
74  
75  /**
76   * The ServerManager class manages info about region servers.
77   * <p>
78   * Maintains lists of online and dead servers.  Processes the startups,
79   * shutdowns, and deaths of region servers.
80   * <p>
81   * Servers are distinguished in two different ways.  A given server has a
82   * location, specified by hostname and port, and of which there can only be one
83   * online at any given time.  A server instance is specified by the location
84   * (hostname and port) as well as the startcode (timestamp from when the server
85   * was started).  This is used to differentiate a restarted instance of a given
86   * server from the original instance.
87   * <p>
88   * If a sever is known not to be running any more, it is called dead. The dead
89   * server needs to be handled by a ServerShutdownHandler.  If the handler is not
90   * enabled yet, the server can't be handled right away so it is queued up.
91   * After the handler is enabled, the server will be submitted to a handler to handle.
92   * However, the handler may be just partially enabled.  If so,
93   * the server cannot be fully processed, and be queued up for further processing.
94   * A server is fully processed only after the handler is fully enabled
95   * and has completed the handling.
96   */
97  @InterfaceAudience.Private
98  public class ServerManager {
99    public static final String WAIT_ON_REGIONSERVERS_MAXTOSTART =
100       "hbase.master.wait.on.regionservers.maxtostart";
101 
102   public static final String WAIT_ON_REGIONSERVERS_MINTOSTART =
103       "hbase.master.wait.on.regionservers.mintostart";
104 
105   public static final String WAIT_ON_REGIONSERVERS_TIMEOUT =
106       "hbase.master.wait.on.regionservers.timeout";
107 
108   public static final String WAIT_ON_REGIONSERVERS_INTERVAL =
109       "hbase.master.wait.on.regionservers.interval";
110 
111   private static final Log LOG = LogFactory.getLog(ServerManager.class);
112 
113   // Set if we are to shutdown the cluster.
114   private volatile boolean clusterShutdown = false;
115 
116   private final SortedMap<byte[], Long> flushedSequenceIdByRegion =
117     new ConcurrentSkipListMap<byte[], Long>(Bytes.BYTES_COMPARATOR);
118 
119   /** Map of registered servers to their current load */
120   private final ConcurrentHashMap<ServerName, ServerLoad> onlineServers =
121     new ConcurrentHashMap<ServerName, ServerLoad>();
122 
123   /**
124    * Map of admin interfaces per registered regionserver; these interfaces we use to control
125    * regionservers out on the cluster
126    */
127   private final Map<ServerName, AdminService.BlockingInterface> rsAdmins =
128     new HashMap<ServerName, AdminService.BlockingInterface>();
129 
130   /**
131    * List of region servers <ServerName> that should not get any more new
132    * regions.
133    */
134   private final ArrayList<ServerName> drainingServers =
135     new ArrayList<ServerName>();
136 
137   private final Server master;
138   private final MasterServices services;
139   private final HConnection connection;
140 
141   private final DeadServer deadservers = new DeadServer();
142 
143   private final long maxSkew;
144   private final long warningSkew;
145   private final boolean checkingBackupMaster;
146   private BaseLoadBalancer balancer;
147 
148   /**
149    * Set of region servers which are dead but not processed immediately. If one
150    * server died before master enables ServerShutdownHandler, the server will be
151    * added to this set and will be processed through calling
152    * {@link ServerManager#processQueuedDeadServers()} by master.
153    * <p>
154    * A dead server is a server instance known to be dead, not listed in the /hbase/rs
155    * znode any more. It may have not been submitted to ServerShutdownHandler yet
156    * because the handler is not enabled.
157    * <p>
158    * A dead server, which has been submitted to ServerShutdownHandler while the
159    * handler is not enabled, is queued up.
160    * <p>
161    * So this is a set of region servers known to be dead but not submitted to
162    * ServerShutdownHander for processing yet.
163    */
164   private Set<ServerName> queuedDeadServers = new HashSet<ServerName>();
165 
166   /**
167    * Set of region servers which are dead and submitted to ServerShutdownHandler to process but not
168    * fully processed immediately.
169    * <p>
170    * If one server died before assignment manager finished the failover cleanup, the server will be
171    * added to this set and will be processed through calling
172    * {@link ServerManager#processQueuedDeadServers()} by assignment manager.
173    * <p>
174    * The Boolean value indicates whether log split is needed inside ServerShutdownHandler
175    * <p>
176    * ServerShutdownHandler processes a dead server submitted to the handler after the handler is
177    * enabled. It may not be able to complete the processing because meta is not yet online or master
178    * is currently in startup mode. In this case, the dead server will be parked in this set
179    * temporarily.
180    */
181   private Map<ServerName, Boolean> requeuedDeadServers
182     = new ConcurrentHashMap<ServerName, Boolean>();
183 
184   /** Listeners that are called on server events. */
185   private List<ServerListener> listeners = new CopyOnWriteArrayList<ServerListener>();
186 
187   /**
188    * Constructor.
189    * @param master
190    * @param services
191    * @throws ZooKeeperConnectionException
192    */
193   public ServerManager(final Server master, final MasterServices services)
194       throws IOException {
195     this(master, services, true);
196   }
197 
198   ServerManager(final Server master, final MasterServices services,
199       final boolean connect) throws IOException {
200     this.master = master;
201     this.services = services;
202     Configuration c = master.getConfiguration();
203     maxSkew = c.getLong("hbase.master.maxclockskew", 30000);
204     warningSkew = c.getLong("hbase.master.warningclockskew", 10000);
205     this.connection = connect ? HConnectionManager.getConnection(c) : null;
206 
207     // Put this in constructor so we don't cast it every time
208     //
209     // We need to check if a newly added server is a backup master
210     // only if we are configured not to assign any region to it.
211     checkingBackupMaster = (master instanceof HMaster)
212       && ((HMaster)master).balancer instanceof BaseLoadBalancer
213       && (c.getInt(BaseLoadBalancer.BACKUP_MASTER_WEIGHT_KEY,
214         BaseLoadBalancer.DEFAULT_BACKUP_MASTER_WEIGHT) < 1);
215     if (checkingBackupMaster) {
216       balancer = (BaseLoadBalancer)((HMaster)master).balancer;
217     }
218   }
219 
220   /**
221    * Add the listener to the notification list.
222    * @param listener The ServerListener to register
223    */
224   public void registerListener(final ServerListener listener) {
225     this.listeners.add(listener);
226   }
227 
228   /**
229    * Remove the listener from the notification list.
230    * @param listener The ServerListener to unregister
231    */
232   public boolean unregisterListener(final ServerListener listener) {
233     return this.listeners.remove(listener);
234   }
235 
236   /**
237    * Let the server manager know a new regionserver has come online
238    * @param ia The remote address
239    * @param port The remote port
240    * @param serverStartcode
241    * @param serverCurrentTime The current time of the region server in ms
242    * @return The ServerName we know this server as.
243    * @throws IOException
244    */
245   ServerName regionServerStartup(final InetAddress ia, final int port,
246     final long serverStartcode, long serverCurrentTime)
247   throws IOException {
248     // Test for case where we get a region startup message from a regionserver
249     // that has been quickly restarted but whose znode expiration handler has
250     // not yet run, or from a server whose fail we are currently processing.
251     // Test its host+port combo is present in serverAddresstoServerInfo.  If it
252     // is, reject the server and trigger its expiration. The next time it comes
253     // in, it should have been removed from serverAddressToServerInfo and queued
254     // for processing by ProcessServerShutdown.
255     ServerName sn = ServerName.valueOf(ia.getHostName(), port, serverStartcode);
256     checkClockSkew(sn, serverCurrentTime);
257     checkIsDead(sn, "STARTUP");
258     if (!checkAndRecordNewServer(sn, ServerLoad.EMPTY_SERVERLOAD)) {
259       LOG.warn("THIS SHOULD NOT HAPPEN, RegionServerStartup"
260         + " could not record the server: " + sn);
261     }
262     return sn;
263   }
264 
265   /**
266    * Updates last flushed sequence Ids for the regions on server sn
267    * @param sn
268    * @param hsl
269    */
270   private void updateLastFlushedSequenceIds(ServerName sn, ServerLoad hsl) {
271     Map<byte[], RegionLoad> regionsLoad = hsl.getRegionsLoad();
272     for (Entry<byte[], RegionLoad> entry : regionsLoad.entrySet()) {
273       Long existingValue = flushedSequenceIdByRegion.get(entry.getKey());
274       long l = entry.getValue().getCompleteSequenceId();
275       if (existingValue != null) {
276         if (l != -1 && l < existingValue) {
277           if (LOG.isDebugEnabled()) {
278             LOG.debug("RegionServer " + sn +
279                 " indicates a last flushed sequence id (" + entry.getValue() +
280                 ") that is less than the previous last flushed sequence id (" +
281                 existingValue + ") for region " +
282                 Bytes.toString(entry.getKey()) + " Ignoring.");
283           }
284           continue; // Don't let smaller sequence ids override greater
285           // sequence ids.
286         }
287       }
288       flushedSequenceIdByRegion.put(entry.getKey(), l);
289     }
290   }
291 
292   void regionServerReport(ServerName sn,
293       ServerLoad sl) throws YouAreDeadException {
294     checkIsDead(sn, "REPORT");
295     if (null == this.onlineServers.replace(sn, sl)) {
296       // Already have this host+port combo and its just different start code?
297       // Just let the server in. Presume master joining a running cluster.
298       // recordNewServer is what happens at the end of reportServerStartup.
299       // The only thing we are skipping is passing back to the regionserver
300       // the ServerName to use. Here we presume a master has already done
301       // that so we'll press on with whatever it gave us for ServerName.
302       if (!checkAndRecordNewServer(sn, sl)) {
303         LOG.info("RegionServerReport ignored, could not record the server: " + sn);
304         return; // Not recorded, so no need to move on
305       }
306     }
307     updateLastFlushedSequenceIds(sn, sl);
308   }
309 
310   /**
311    * Check is a server of same host and port already exists,
312    * if not, or the existed one got a smaller start code, record it.
313    *
314    * @param sn the server to check and record
315    * @param sl the server load on the server
316    * @return true if the server is recorded, otherwise, false
317    */
318   boolean checkAndRecordNewServer(
319       final ServerName serverName, final ServerLoad sl) {
320     ServerName existingServer = null;
321     synchronized (this.onlineServers) {
322       existingServer = findServerWithSameHostnamePortWithLock(serverName);
323       if (existingServer != null && (existingServer.getStartcode() > serverName.getStartcode())) {
324         LOG.info("Server serverName=" + serverName + " rejected; we already have "
325             + existingServer.toString() + " registered with same hostname and port");
326         return false;
327       }
328       recordNewServerWithLock(serverName, sl);
329     }
330 
331     // Tell our listeners that a server was added
332     if (!this.listeners.isEmpty()) {
333       for (ServerListener listener : this.listeners) {
334         listener.serverAdded(serverName);
335       }
336     }
337 
338     // Note that we assume that same ts means same server, and don't expire in that case.
339     //  TODO: ts can theoretically collide due to clock shifts, so this is a bit hacky.
340     if (existingServer != null && (existingServer.getStartcode() < serverName.getStartcode())) {
341       LOG.info("Triggering server recovery; existingServer " +
342           existingServer + " looks stale, new server:" + serverName);
343       expireServer(existingServer);
344     }
345     return true;
346   }
347 
348   /**
349    * Checks if the clock skew between the server and the master. If the clock skew exceeds the
350    * configured max, it will throw an exception; if it exceeds the configured warning threshold,
351    * it will log a warning but start normally.
352    * @param serverName Incoming servers's name
353    * @param serverCurrentTime
354    * @throws ClockOutOfSyncException if the skew exceeds the configured max value
355    */
356   private void checkClockSkew(final ServerName serverName, final long serverCurrentTime)
357   throws ClockOutOfSyncException {
358     long skew = Math.abs(System.currentTimeMillis() - serverCurrentTime);
359     if (skew > maxSkew) {
360       String message = "Server " + serverName + " has been " +
361         "rejected; Reported time is too far out of sync with master.  " +
362         "Time difference of " + skew + "ms > max allowed of " + maxSkew + "ms";
363       LOG.warn(message);
364       throw new ClockOutOfSyncException(message);
365     } else if (skew > warningSkew){
366       String message = "Reported time for server " + serverName + " is out of sync with master " +
367         "by " + skew + "ms. (Warning threshold is " + warningSkew + "ms; " +
368         "error threshold is " + maxSkew + "ms)";
369       LOG.warn(message);
370     }
371   }
372 
373   /**
374    * If this server is on the dead list, reject it with a YouAreDeadException.
375    * If it was dead but came back with a new start code, remove the old entry
376    * from the dead list.
377    * @param serverName
378    * @param what START or REPORT
379    * @throws org.apache.hadoop.hbase.YouAreDeadException
380    */
381   private void checkIsDead(final ServerName serverName, final String what)
382       throws YouAreDeadException {
383     if (this.deadservers.isDeadServer(serverName)) {
384       // host name, port and start code all match with existing one of the
385       // dead servers. So, this server must be dead.
386       String message = "Server " + what + " rejected; currently processing " +
387           serverName + " as dead server";
388       LOG.debug(message);
389       throw new YouAreDeadException(message);
390     }
391     // remove dead server with same hostname and port of newly checking in rs after master
392     // initialization.See HBASE-5916 for more information.
393     if ((this.services == null || ((HMaster) this.services).isInitialized())
394         && this.deadservers.cleanPreviousInstance(serverName)) {
395       // This server has now become alive after we marked it as dead.
396       // We removed it's previous entry from the dead list to reflect it.
397       LOG.debug(what + ":" + " Server " + serverName + " came back up," +
398           " removed it from the dead servers list");
399     }
400   }
401 
402   /**
403    * Assumes onlineServers is locked.
404    * @return ServerName with matching hostname and port.
405    */
406   private ServerName findServerWithSameHostnamePortWithLock(
407       final ServerName serverName) {
408     for (ServerName sn: this.onlineServers.keySet()) {
409       if (ServerName.isSameHostnameAndPort(serverName, sn)) return sn;
410     }
411     return null;
412   }
413 
414   /**
415    * Adds the onlineServers list. onlineServers should be locked.
416    * @param serverName The remote servers name.
417    * @param sl
418    * @return Server load from the removed server, if any.
419    */
420   @VisibleForTesting
421   void recordNewServerWithLock(final ServerName serverName, final ServerLoad sl) {
422     LOG.info("Registering server=" + serverName);
423     if (checkingBackupMaster) {
424       ZooKeeperWatcher zooKeeper = master.getZooKeeper();
425       String backupZNode = ZKUtil.joinZNode(
426         zooKeeper.backupMasterAddressesZNode, serverName.toString());
427       try {
428         if (ZKUtil.checkExists(zooKeeper, backupZNode) != -1) {
429           balancer.excludeServer(serverName);
430         }
431       } catch (KeeperException e) {
432         master.abort("Failed to check if a new server a backup master", e);
433       }
434     }
435     this.onlineServers.put(serverName, sl);
436     this.rsAdmins.remove(serverName);
437   }
438 
439   public long getLastFlushedSequenceId(byte[] regionName) {
440     long seqId = -1;
441     if (flushedSequenceIdByRegion.containsKey(regionName)) {
442       seqId = flushedSequenceIdByRegion.get(regionName);
443     }
444     return seqId;
445   }
446 
447   /**
448    * @param serverName
449    * @return ServerLoad if serverName is known else null
450    */
451   public ServerLoad getLoad(final ServerName serverName) {
452     return this.onlineServers.get(serverName);
453   }
454 
455   /**
456    * Compute the average load across all region servers.
457    * Currently, this uses a very naive computation - just uses the number of
458    * regions being served, ignoring stats about number of requests.
459    * @return the average load
460    */
461   public double getAverageLoad() {
462     int totalLoad = 0;
463     int numServers = 0;
464     for (ServerLoad sl: this.onlineServers.values()) {
465         numServers++;
466         totalLoad += sl.getNumberOfRegions();
467     }
468     return numServers == 0 ? 0 :
469       (double)totalLoad / (double)numServers;
470   }
471 
472   /**
473    * Get the count of active regionservers that are not backup
474    * masters. This count may not be accurate depending on timing.
475    * @return the count of active regionservers
476    */
477   private int countOfRegionServers() {
478     // Presumes onlineServers is a concurrent map
479     int count = this.onlineServers.size();
480     if (balancer != null) {
481       count -= balancer.getExcludedServers().size();
482       if (count < 0) count = 0;
483     }
484     return count;
485   }
486 
487   /**
488    * @return Read-only map of servers to serverinfo
489    */
490   public Map<ServerName, ServerLoad> getOnlineServers() {
491     // Presumption is that iterating the returned Map is OK.
492     synchronized (this.onlineServers) {
493       return Collections.unmodifiableMap(this.onlineServers);
494     }
495   }
496 
497 
498   public DeadServer getDeadServers() {
499     return this.deadservers;
500   }
501 
502   /**
503    * Checks if any dead servers are currently in progress.
504    * @return true if any RS are being processed as dead, false if not
505    */
506   public boolean areDeadServersInProgress() {
507     return this.deadservers.areDeadServersInProgress();
508   }
509 
510   void letRegionServersShutdown() {
511     long previousLogTime = 0;
512     ServerName sn = master.getServerName();
513     ZooKeeperWatcher zkw = master.getZooKeeper();
514     int onlineServersCt;
515     while ((onlineServersCt = onlineServers.size()) > 0){
516 
517       if (System.currentTimeMillis() > (previousLogTime + 1000)) {
518         Set<ServerName> remainingServers = onlineServers.keySet();
519         synchronized (onlineServers) {
520           if (remainingServers.size() == 1 && remainingServers.contains(sn)) {
521             // Master will delete itself later.
522             return;
523           }
524         }
525         StringBuilder sb = new StringBuilder();
526         // It's ok here to not sync on onlineServers - merely logging
527         for (ServerName key : remainingServers) {
528           if (sb.length() > 0) {
529             sb.append(", ");
530           }
531           sb.append(key);
532         }
533         LOG.info("Waiting on regionserver(s) to go down " + sb.toString());
534         previousLogTime = System.currentTimeMillis();
535       }
536 
537       try {
538         List<String> servers = ZKUtil.listChildrenNoWatch(zkw, zkw.rsZNode);
539         if (servers == null || (servers.size() == 1
540             && servers.contains(sn.toString()))) {
541           LOG.info("ZK shows there is only the master self online, exiting now");
542           // Master could have lost some ZK events, no need to wait more.
543           break;
544         }
545       } catch (KeeperException ke) {
546         LOG.warn("Failed to list regionservers", ke);
547         // ZK is malfunctioning, don't hang here
548         break;
549       }
550       synchronized (onlineServers) {
551         try {
552           if (onlineServersCt == onlineServers.size()) onlineServers.wait(100);
553         } catch (InterruptedException ignored) {
554           // continue
555         }
556       }
557     }
558   }
559 
560   /*
561    * Expire the passed server.  Add it to list of dead servers and queue a
562    * shutdown processing.
563    */
564   public synchronized void expireServer(final ServerName serverName) {
565     if (serverName.equals(master.getServerName())) {
566       if (!(master.isAborted() || master.isStopped())) {
567         master.stop("We lost our znode?");
568       }
569       return;
570     }
571     if (!services.isServerShutdownHandlerEnabled()) {
572       LOG.info("Master doesn't enable ServerShutdownHandler during initialization, "
573           + "delay expiring server " + serverName);
574       this.queuedDeadServers.add(serverName);
575       return;
576     }
577     if (this.deadservers.isDeadServer(serverName)) {
578       // TODO: Can this happen?  It shouldn't be online in this case?
579       LOG.warn("Expiration of " + serverName +
580           " but server shutdown already in progress");
581       return;
582     }
583     synchronized (onlineServers) {
584       if (!this.onlineServers.containsKey(serverName)) {
585         LOG.warn("Expiration of " + serverName + " but server not online");
586       }
587       // Remove the server from the known servers lists and update load info BUT
588       // add to deadservers first; do this so it'll show in dead servers list if
589       // not in online servers list.
590       this.deadservers.add(serverName);
591       this.onlineServers.remove(serverName);
592       onlineServers.notifyAll();
593     }
594     this.rsAdmins.remove(serverName);
595     // If cluster is going down, yes, servers are going to be expiring; don't
596     // process as a dead server
597     if (this.clusterShutdown) {
598       LOG.info("Cluster shutdown set; " + serverName +
599         " expired; onlineServers=" + this.onlineServers.size());
600       if (this.onlineServers.isEmpty()) {
601         master.stop("Cluster shutdown set; onlineServer=0");
602       }
603       return;
604     }
605 
606     boolean carryingMeta = services.getAssignmentManager().isCarryingMeta(serverName);
607     if (carryingMeta) {
608       this.services.getExecutorService().submit(new MetaServerShutdownHandler(this.master,
609         this.services, this.deadservers, serverName));
610     } else {
611       this.services.getExecutorService().submit(new ServerShutdownHandler(this.master,
612         this.services, this.deadservers, serverName, true));
613     }
614     LOG.debug("Added=" + serverName +
615       " to dead servers, submitted shutdown handler to be executed meta=" + carryingMeta);
616 
617     // Tell our listeners that a server was removed
618     if (!this.listeners.isEmpty()) {
619       for (ServerListener listener : this.listeners) {
620         listener.serverRemoved(serverName);
621       }
622     }
623   }
624 
625   public synchronized void processDeadServer(final ServerName serverName) {
626     this.processDeadServer(serverName, false);
627   }
628 
629   public synchronized void processDeadServer(final ServerName serverName, boolean shouldSplitHlog) {
630     // When assignment manager is cleaning up the zookeeper nodes and rebuilding the
631     // in-memory region states, region servers could be down. Meta table can and
632     // should be re-assigned, log splitting can be done too. However, it is better to
633     // wait till the cleanup is done before re-assigning user regions.
634     //
635     // We should not wait in the server shutdown handler thread since it can clog
636     // the handler threads and meta table could not be re-assigned in case
637     // the corresponding server is down. So we queue them up here instead.
638     if (!services.getAssignmentManager().isFailoverCleanupDone()) {
639       requeuedDeadServers.put(serverName, shouldSplitHlog);
640       return;
641     }
642 
643     this.deadservers.add(serverName);
644     this.services.getExecutorService().submit(
645       new ServerShutdownHandler(this.master, this.services, this.deadservers, serverName,
646           shouldSplitHlog));
647   }
648 
649   /**
650    * Process the servers which died during master's initialization. It will be
651    * called after HMaster#assignMeta and AssignmentManager#joinCluster.
652    * */
653   synchronized void processQueuedDeadServers() {
654     if (!services.isServerShutdownHandlerEnabled()) {
655       LOG.info("Master hasn't enabled ServerShutdownHandler");
656     }
657     Iterator<ServerName> serverIterator = queuedDeadServers.iterator();
658     while (serverIterator.hasNext()) {
659       ServerName tmpServerName = serverIterator.next();
660       expireServer(tmpServerName);
661       serverIterator.remove();
662       requeuedDeadServers.remove(tmpServerName);
663     }
664 
665     if (!services.getAssignmentManager().isFailoverCleanupDone()) {
666       LOG.info("AssignmentManager hasn't finished failover cleanup; waiting");
667     }
668 
669     for(ServerName tmpServerName : requeuedDeadServers.keySet()){
670       processDeadServer(tmpServerName, requeuedDeadServers.get(tmpServerName));
671     }
672     requeuedDeadServers.clear();
673   }
674 
675   /*
676    * Remove the server from the drain list.
677    */
678   public boolean removeServerFromDrainList(final ServerName sn) {
679     // Warn if the server (sn) is not online.  ServerName is of the form:
680     // <hostname> , <port> , <startcode>
681 
682     if (!this.isServerOnline(sn)) {
683       LOG.warn("Server " + sn + " is not currently online. " +
684                "Removing from draining list anyway, as requested.");
685     }
686     // Remove the server from the draining servers lists.
687     return this.drainingServers.remove(sn);
688   }
689 
690   /*
691    * Add the server to the drain list.
692    */
693   public boolean addServerToDrainList(final ServerName sn) {
694     // Warn if the server (sn) is not online.  ServerName is of the form:
695     // <hostname> , <port> , <startcode>
696 
697     if (!this.isServerOnline(sn)) {
698       LOG.warn("Server " + sn + " is not currently online. " +
699                "Ignoring request to add it to draining list.");
700       return false;
701     }
702     // Add the server to the draining servers lists, if it's not already in
703     // it.
704     if (this.drainingServers.contains(sn)) {
705       LOG.warn("Server " + sn + " is already in the draining server list." +
706                "Ignoring request to add it again.");
707       return false;
708     }
709     return this.drainingServers.add(sn);
710   }
711 
712   // RPC methods to region servers
713 
714   /**
715    * Sends an OPEN RPC to the specified server to open the specified region.
716    * <p>
717    * Open should not fail but can if server just crashed.
718    * <p>
719    * @param server server to open a region
720    * @param region region to open
721    * @param versionOfOfflineNode that needs to be present in the offline node
722    * when RS tries to change the state from OFFLINE to other states.
723    * @param favoredNodes
724    */
725   public RegionOpeningState sendRegionOpen(final ServerName server,
726       HRegionInfo region, int versionOfOfflineNode, List<ServerName> favoredNodes)
727   throws IOException {
728     AdminService.BlockingInterface admin = getRsAdmin(server);
729     if (admin == null) {
730       LOG.warn("Attempting to send OPEN RPC to server " + server.toString() +
731         " failed because no RPC connection found to this server");
732       return RegionOpeningState.FAILED_OPENING;
733     }
734     OpenRegionRequest request = RequestConverter.buildOpenRegionRequest(server, 
735       region, versionOfOfflineNode, favoredNodes, 
736       (RecoveryMode.LOG_REPLAY == this.services.getMasterFileSystem().getLogRecoveryMode()));
737     try {
738       OpenRegionResponse response = admin.openRegion(null, request);
739       return ResponseConverter.getRegionOpeningState(response);
740     } catch (ServiceException se) {
741       throw ProtobufUtil.getRemoteException(se);
742     }
743   }
744 
745   /**
746    * Sends an OPEN RPC to the specified server to open the specified region.
747    * <p>
748    * Open should not fail but can if server just crashed.
749    * <p>
750    * @param server server to open a region
751    * @param regionOpenInfos info of a list of regions to open
752    * @return a list of region opening states
753    */
754   public List<RegionOpeningState> sendRegionOpen(ServerName server,
755       List<Triple<HRegionInfo, Integer, List<ServerName>>> regionOpenInfos)
756   throws IOException {
757     AdminService.BlockingInterface admin = getRsAdmin(server);
758     if (admin == null) {
759       LOG.warn("Attempting to send OPEN RPC to server " + server.toString() +
760         " failed because no RPC connection found to this server");
761       return null;
762     }
763 
764     OpenRegionRequest request = RequestConverter.buildOpenRegionRequest(regionOpenInfos, 
765       (RecoveryMode.LOG_REPLAY == this.services.getMasterFileSystem().getLogRecoveryMode()));
766     try {
767       OpenRegionResponse response = admin.openRegion(null, request);
768       return ResponseConverter.getRegionOpeningStateList(response);
769     } catch (ServiceException se) {
770       throw ProtobufUtil.getRemoteException(se);
771     }
772   }
773 
774   /**
775    * Sends an CLOSE RPC to the specified server to close the specified region.
776    * <p>
777    * A region server could reject the close request because it either does not
778    * have the specified region or the region is being split.
779    * @param server server to open a region
780    * @param region region to open
781    * @param versionOfClosingNode
782    *   the version of znode to compare when RS transitions the znode from
783    *   CLOSING state.
784    * @param dest - if the region is moved to another server, the destination server. null otherwise.
785    * @return true if server acknowledged close, false if not
786    * @throws IOException
787    */
788   public boolean sendRegionClose(ServerName server, HRegionInfo region,
789     int versionOfClosingNode, ServerName dest, boolean transitionInZK) throws IOException {
790     if (server == null) throw new NullPointerException("Passed server is null");
791     AdminService.BlockingInterface admin = getRsAdmin(server);
792     if (admin == null) {
793       throw new IOException("Attempting to send CLOSE RPC to server " +
794         server.toString() + " for region " +
795         region.getRegionNameAsString() +
796         " failed because no RPC connection found to this server");
797     }
798     return ProtobufUtil.closeRegion(admin, server, region.getRegionName(),
799       versionOfClosingNode, dest, transitionInZK);
800   }
801 
802   public boolean sendRegionClose(ServerName server,
803       HRegionInfo region, int versionOfClosingNode) throws IOException {
804     return sendRegionClose(server, region, versionOfClosingNode, null, true);
805   }
806 
807   /**
808    * Sends an MERGE REGIONS RPC to the specified server to merge the specified
809    * regions.
810    * <p>
811    * A region server could reject the close request because it either does not
812    * have the specified region.
813    * @param server server to merge regions
814    * @param region_a region to merge
815    * @param region_b region to merge
816    * @param forcible true if do a compulsory merge, otherwise we will only merge
817    *          two adjacent regions
818    * @throws IOException
819    */
820   public void sendRegionsMerge(ServerName server, HRegionInfo region_a,
821       HRegionInfo region_b, boolean forcible) throws IOException {
822     if (server == null)
823       throw new NullPointerException("Passed server is null");
824     if (region_a == null || region_b == null)
825       throw new NullPointerException("Passed region is null");
826     AdminService.BlockingInterface admin = getRsAdmin(server);
827     if (admin == null) {
828       throw new IOException("Attempting to send MERGE REGIONS RPC to server "
829           + server.toString() + " for region "
830           + region_a.getRegionNameAsString() + ","
831           + region_b.getRegionNameAsString()
832           + " failed because no RPC connection found to this server");
833     }
834     ProtobufUtil.mergeRegions(admin, region_a, region_b, forcible);
835   }
836 
837   /**
838    * Check if a region server is reachable and has the expected start code
839    */
840   public boolean isServerReachable(ServerName server) {
841     if (server == null) throw new NullPointerException("Passed server is null");
842     int maximumAttempts = Math.max(1, master.getConfiguration().getInt(
843       "hbase.master.maximum.ping.server.attempts", 10));
844     for (int i = 0; i < maximumAttempts; i++) {
845       try {
846         AdminService.BlockingInterface admin = getRsAdmin(server);
847         if (admin != null) {
848           ServerInfo info = ProtobufUtil.getServerInfo(admin);
849           return info != null && info.hasServerName()
850             && server.getStartcode() == info.getServerName().getStartCode();
851         }
852       } catch (IOException ioe) {
853         LOG.debug("Couldn't reach " + server + ", try=" + i
854           + " of " + maximumAttempts, ioe);
855       }
856     }
857     return false;
858   }
859 
860     /**
861     * @param sn
862     * @return Admin interface for the remote regionserver named <code>sn</code>
863     * @throws IOException
864     * @throws RetriesExhaustedException wrapping a ConnectException if failed
865     */
866   @SuppressWarnings("deprecation")
867   private AdminService.BlockingInterface getRsAdmin(final ServerName sn)
868   throws IOException {
869     AdminService.BlockingInterface admin = this.rsAdmins.get(sn);
870     if (admin == null) {
871       LOG.debug("New admin connection to " + sn.toString());
872       if (sn.equals(master.getServerName()) && master instanceof HRegionServer) {
873         // A master is also a region server now, see HBASE-10569 for details
874         admin = ((HRegionServer)master).getRSRpcServices();
875       } else {
876         admin = this.connection.getAdmin(sn);
877       }
878       this.rsAdmins.put(sn, admin);
879     }
880     return admin;
881   }
882 
883   /**
884    * Wait for the region servers to report in.
885    * We will wait until one of this condition is met:
886    *  - the master is stopped
887    *  - the 'hbase.master.wait.on.regionservers.maxtostart' number of
888    *    region servers is reached
889    *  - the 'hbase.master.wait.on.regionservers.mintostart' is reached AND
890    *   there have been no new region server in for
891    *      'hbase.master.wait.on.regionservers.interval' time AND
892    *   the 'hbase.master.wait.on.regionservers.timeout' is reached
893    *
894    * @throws InterruptedException
895    */
896   public void waitForRegionServers(MonitoredTask status)
897   throws InterruptedException {
898     final long interval = this.master.getConfiguration().
899       getLong(WAIT_ON_REGIONSERVERS_INTERVAL, 1500);
900     final long timeout = this.master.getConfiguration().
901       getLong(WAIT_ON_REGIONSERVERS_TIMEOUT, 4500);
902     int minToStart = this.master.getConfiguration().
903       getInt(WAIT_ON_REGIONSERVERS_MINTOSTART, 2);
904     if (minToStart < 1) {
905       LOG.warn(String.format(
906         "The value of '%s' (%d) can not be less than 1, ignoring.",
907         WAIT_ON_REGIONSERVERS_MINTOSTART, minToStart));
908       minToStart = 1;
909     }
910     int maxToStart = this.master.getConfiguration().
911       getInt(WAIT_ON_REGIONSERVERS_MAXTOSTART, Integer.MAX_VALUE);
912     if (maxToStart < minToStart) {
913         LOG.warn(String.format(
914             "The value of '%s' (%d) is set less than '%s' (%d), ignoring.",
915             WAIT_ON_REGIONSERVERS_MAXTOSTART, maxToStart,
916             WAIT_ON_REGIONSERVERS_MINTOSTART, minToStart));
917         maxToStart = Integer.MAX_VALUE;
918     }
919 
920     long now =  System.currentTimeMillis();
921     final long startTime = now;
922     long slept = 0;
923     long lastLogTime = 0;
924     long lastCountChange = startTime;
925     int count = countOfRegionServers();
926     int oldCount = 0;
927     ServerName masterSn = master.getServerName();
928     boolean selfCheckedIn = isServerOnline(masterSn);
929     while (!this.master.isStopped() && (!selfCheckedIn || (count < maxToStart
930         && (lastCountChange+interval > now || timeout > slept || count < minToStart)))) {
931       // Log some info at every interval time or if there is a change
932       if (oldCount != count || lastLogTime+interval < now){
933         lastLogTime = now;
934         String msg =
935           "Waiting for region servers count to settle; currently"+
936             " checked in " + count + ", slept for " + slept + " ms," +
937             " expecting minimum of " + minToStart + ", maximum of "+ maxToStart+
938             ", timeout of "+timeout+" ms, interval of "+interval+" ms," +
939             " selfCheckedIn " + selfCheckedIn;
940         LOG.info(msg);
941         status.setStatus(msg);
942       }
943 
944       // We sleep for some time
945       final long sleepTime = 50;
946       Thread.sleep(sleepTime);
947       now =  System.currentTimeMillis();
948       slept = now - startTime;
949 
950       selfCheckedIn = isServerOnline(masterSn);
951 
952       oldCount = count;
953       count = countOfRegionServers();
954       if (count != oldCount) {
955         lastCountChange = now;
956       }
957     }
958 
959     LOG.info("Finished waiting for region servers count to settle;" +
960       " checked in " + count + ", slept for " + slept + " ms," +
961       " expecting minimum of " + minToStart + ", maximum of "+ maxToStart+","+
962       " master is "+ (this.master.isStopped() ? "stopped.": "running," +
963       " selfCheckedIn " + selfCheckedIn)
964     );
965   }
966 
967   /**
968    * @return A copy of the internal list of online servers.
969    */
970   public List<ServerName> getOnlineServersList() {
971     // TODO: optimize the load balancer call so we don't need to make a new list
972     // TODO: FIX. THIS IS POPULAR CALL.
973     return new ArrayList<ServerName>(this.onlineServers.keySet());
974   }
975 
976   /**
977    * @return A copy of the internal list of draining servers.
978    */
979   public List<ServerName> getDrainingServersList() {
980     return new ArrayList<ServerName>(this.drainingServers);
981   }
982 
983   /**
984    * @return A copy of the internal set of deadNotExpired servers.
985    */
986   Set<ServerName> getDeadNotExpiredServers() {
987     return new HashSet<ServerName>(this.queuedDeadServers);
988   }
989 
990   /**
991    * During startup, if we figure it is not a failover, i.e. there is
992    * no more HLog files to split, we won't try to recover these dead servers.
993    * So we just remove them from the queue. Use caution in calling this.
994    */
995   void removeRequeuedDeadServers() {
996     requeuedDeadServers.clear();
997   }
998 
999   /**
1000    * @return A copy of the internal map of requeuedDeadServers servers and their corresponding
1001    *         splitlog need flag.
1002    */
1003   Map<ServerName, Boolean> getRequeuedDeadServers() {
1004     return Collections.unmodifiableMap(this.requeuedDeadServers);
1005   }
1006 
1007   public boolean isServerOnline(ServerName serverName) {
1008     return serverName != null && onlineServers.containsKey(serverName);
1009   }
1010 
1011   /**
1012    * Check if a server is known to be dead.  A server can be online,
1013    * or known to be dead, or unknown to this manager (i.e, not online,
1014    * not known to be dead either. it is simply not tracked by the
1015    * master any more, for example, a very old previous instance).
1016    */
1017   public synchronized boolean isServerDead(ServerName serverName) {
1018     return serverName == null || deadservers.isDeadServer(serverName)
1019       || queuedDeadServers.contains(serverName)
1020       || requeuedDeadServers.containsKey(serverName);
1021   }
1022 
1023   public void shutdownCluster() {
1024     this.clusterShutdown = true;
1025     this.master.stop("Cluster shutdown requested");
1026   }
1027 
1028   public boolean isClusterShutdown() {
1029     return this.clusterShutdown;
1030   }
1031 
1032   /**
1033    * Stop the ServerManager.  Currently closes the connection to the master.
1034    */
1035   public void stop() {
1036     if (connection != null) {
1037       try {
1038         connection.close();
1039       } catch (IOException e) {
1040         LOG.error("Attempt to close connection to master failed", e);
1041       }
1042     }
1043   }
1044 
1045   /**
1046    * Creates a list of possible destinations for a region. It contains the online servers, but not
1047    *  the draining or dying servers.
1048    *  @param serverToExclude can be null if there is no server to exclude
1049    */
1050   public List<ServerName> createDestinationServersList(final ServerName serverToExclude){
1051     final List<ServerName> destServers = getOnlineServersList();
1052 
1053     if (serverToExclude != null){
1054       destServers.remove(serverToExclude);
1055     }
1056 
1057     // Loop through the draining server list and remove them from the server list
1058     final List<ServerName> drainingServersCopy = getDrainingServersList();
1059     if (!drainingServersCopy.isEmpty()) {
1060       for (final ServerName server: drainingServersCopy) {
1061         destServers.remove(server);
1062       }
1063     }
1064 
1065     // Remove the deadNotExpired servers from the server list.
1066     removeDeadNotExpiredServers(destServers);
1067     return destServers;
1068   }
1069 
1070   /**
1071    * Calls {@link #createDestinationServersList} without server to exclude.
1072    */
1073   public List<ServerName> createDestinationServersList(){
1074     return createDestinationServersList(null);
1075   }
1076 
1077     /**
1078     * Loop through the deadNotExpired server list and remove them from the
1079     * servers.
1080     * This function should be used carefully outside of this class. You should use a high level
1081     *  method such as {@link #createDestinationServersList()} instead of managing you own list.
1082     */
1083   void removeDeadNotExpiredServers(List<ServerName> servers) {
1084     Set<ServerName> deadNotExpiredServersCopy = this.getDeadNotExpiredServers();
1085     if (!deadNotExpiredServersCopy.isEmpty()) {
1086       for (ServerName server : deadNotExpiredServersCopy) {
1087         LOG.debug("Removing dead but not expired server: " + server
1088           + " from eligible server pool.");
1089         servers.remove(server);
1090       }
1091     }
1092   }
1093 
1094   /**
1095    * To clear any dead server with same host name and port of any online server
1096    */
1097   void clearDeadServersWithSameHostNameAndPortOfOnlineServer() {
1098     for (ServerName serverName : getOnlineServersList()) {
1099       deadservers.cleanAllPreviousInstances(serverName);
1100     }
1101   }
1102 }