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