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