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