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