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