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.regionserver;
20  
21  import java.io.IOException;
22  import java.io.InterruptedIOException;
23  import java.lang.Thread.UncaughtExceptionHandler;
24  import java.lang.management.ManagementFactory;
25  import java.lang.management.MemoryUsage;
26  import java.lang.reflect.Constructor;
27  import java.net.BindException;
28  import java.net.InetAddress;
29  import java.net.InetSocketAddress;
30  import java.util.ArrayList;
31  import java.util.Collection;
32  import java.util.Collections;
33  import java.util.Comparator;
34  import java.util.HashMap;
35  import java.util.HashSet;
36  import java.util.Iterator;
37  import java.util.List;
38  import java.util.Map;
39  import java.util.Map.Entry;
40  import java.util.Set;
41  import java.util.SortedMap;
42  import java.util.TreeMap;
43  import java.util.TreeSet;
44  import java.util.concurrent.ConcurrentHashMap;
45  import java.util.concurrent.ConcurrentMap;
46  import java.util.concurrent.ConcurrentSkipListMap;
47  import java.util.concurrent.atomic.AtomicBoolean;
48  import java.util.concurrent.atomic.AtomicReference;
49  import java.util.concurrent.locks.ReentrantReadWriteLock;
50  
51  import javax.management.MalformedObjectNameException;
52  import javax.management.ObjectName;
53  import javax.servlet.http.HttpServlet;
54  
55  import org.apache.commons.lang.math.RandomUtils;
56  import org.apache.commons.logging.Log;
57  import org.apache.commons.logging.LogFactory;
58  import org.apache.hadoop.conf.Configuration;
59  import org.apache.hadoop.fs.FileSystem;
60  import org.apache.hadoop.fs.Path;
61  import org.apache.hadoop.hbase.ChoreService;
62  import org.apache.hadoop.hbase.ClockOutOfSyncException;
63  import org.apache.hadoop.hbase.CoordinatedStateManager;
64  import org.apache.hadoop.hbase.CoordinatedStateManagerFactory;
65  import org.apache.hadoop.hbase.HBaseConfiguration;
66  import org.apache.hadoop.hbase.HBaseInterfaceAudience;
67  import org.apache.hadoop.hbase.HConstants;
68  import org.apache.hadoop.hbase.HRegionInfo;
69  import org.apache.hadoop.hbase.HealthCheckChore;
70  import org.apache.hadoop.hbase.MetaTableAccessor;
71  import org.apache.hadoop.hbase.NotServingRegionException;
72  import org.apache.hadoop.hbase.ScheduledChore;
73  import org.apache.hadoop.hbase.ServerName;
74  import org.apache.hadoop.hbase.Stoppable;
75  import org.apache.hadoop.hbase.TableDescriptors;
76  import org.apache.hadoop.hbase.TableName;
77  import org.apache.hadoop.hbase.YouAreDeadException;
78  import org.apache.hadoop.hbase.ZNodeClearer;
79  import org.apache.hadoop.hbase.classification.InterfaceAudience;
80  import org.apache.hadoop.hbase.client.ClusterConnection;
81  import org.apache.hadoop.hbase.client.ConnectionFactory;
82  import org.apache.hadoop.hbase.client.ConnectionUtils;
83  import org.apache.hadoop.hbase.client.RpcRetryingCallerFactory;
84  import org.apache.hadoop.hbase.conf.ConfigurationManager;
85  import org.apache.hadoop.hbase.coordination.BaseCoordinatedStateManager;
86  import org.apache.hadoop.hbase.coordination.SplitLogWorkerCoordination;
87  import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
88  import org.apache.hadoop.hbase.exceptions.RegionMovedException;
89  import org.apache.hadoop.hbase.exceptions.RegionOpeningException;
90  import org.apache.hadoop.hbase.exceptions.UnknownProtocolException;
91  import org.apache.hadoop.hbase.executor.ExecutorService;
92  import org.apache.hadoop.hbase.executor.ExecutorType;
93  import org.apache.hadoop.hbase.fs.HFileSystem;
94  import org.apache.hadoop.hbase.http.InfoServer;
95  import org.apache.hadoop.hbase.io.hfile.CacheConfig;
96  import org.apache.hadoop.hbase.io.hfile.HFile;
97  import org.apache.hadoop.hbase.ipc.RpcClient;
98  import org.apache.hadoop.hbase.ipc.RpcClientFactory;
99  import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
100 import org.apache.hadoop.hbase.ipc.RpcServerInterface;
101 import org.apache.hadoop.hbase.ipc.ServerNotRunningYetException;
102 import org.apache.hadoop.hbase.ipc.ServerRpcController;
103 import org.apache.hadoop.hbase.master.HMaster;
104 import org.apache.hadoop.hbase.master.RegionState.State;
105 import org.apache.hadoop.hbase.master.TableLockManager;
106 import org.apache.hadoop.hbase.mob.MobCacheConfig;
107 import org.apache.hadoop.hbase.procedure.RegionServerProcedureManagerHost;
108 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
109 import org.apache.hadoop.hbase.protobuf.RequestConverter;
110 import org.apache.hadoop.hbase.protobuf.ResponseConverter;
111 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
112 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.CoprocessorServiceCall;
113 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.CoprocessorServiceRequest;
114 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.CoprocessorServiceResponse;
115 import org.apache.hadoop.hbase.protobuf.generated.ClusterStatusProtos;
116 import org.apache.hadoop.hbase.protobuf.generated.ClusterStatusProtos.RegionLoad;
117 import org.apache.hadoop.hbase.protobuf.generated.ClusterStatusProtos.RegionStoreSequenceIds;
118 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.Coprocessor;
119 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.Coprocessor.Builder;
120 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.NameStringPair;
121 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.RegionServerInfo;
122 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.RegionSpecifier;
123 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.RegionSpecifier.RegionSpecifierType;
124 import org.apache.hadoop.hbase.protobuf.generated.RegionServerStatusProtos.GetLastFlushedSequenceIdRequest;
125 import org.apache.hadoop.hbase.protobuf.generated.RegionServerStatusProtos.GetLastFlushedSequenceIdResponse;
126 import org.apache.hadoop.hbase.protobuf.generated.RegionServerStatusProtos.RegionServerReportRequest;
127 import org.apache.hadoop.hbase.protobuf.generated.RegionServerStatusProtos.RegionServerStartupRequest;
128 import org.apache.hadoop.hbase.protobuf.generated.RegionServerStatusProtos.RegionServerStartupResponse;
129 import org.apache.hadoop.hbase.protobuf.generated.RegionServerStatusProtos.RegionServerStatusService;
130 import org.apache.hadoop.hbase.protobuf.generated.RegionServerStatusProtos.RegionStateTransition;
131 import org.apache.hadoop.hbase.protobuf.generated.RegionServerStatusProtos.RegionStateTransition.TransitionCode;
132 import org.apache.hadoop.hbase.protobuf.generated.RegionServerStatusProtos.ReportRSFatalErrorRequest;
133 import org.apache.hadoop.hbase.protobuf.generated.RegionServerStatusProtos.ReportRegionStateTransitionRequest;
134 import org.apache.hadoop.hbase.protobuf.generated.RegionServerStatusProtos.ReportRegionStateTransitionResponse;
135 import org.apache.hadoop.hbase.quotas.RegionServerQuotaManager;
136 import org.apache.hadoop.hbase.regionserver.compactions.CompactionProgress;
137 import org.apache.hadoop.hbase.regionserver.handler.CloseMetaHandler;
138 import org.apache.hadoop.hbase.regionserver.handler.CloseRegionHandler;
139 import org.apache.hadoop.hbase.regionserver.handler.RegionReplicaFlushHandler;
140 import org.apache.hadoop.hbase.regionserver.wal.MetricsWAL;
141 import org.apache.hadoop.hbase.regionserver.wal.WALActionsListener;
142 import org.apache.hadoop.hbase.replication.regionserver.ReplicationLoad;
143 import org.apache.hadoop.hbase.security.Superusers;
144 import org.apache.hadoop.hbase.security.UserProvider;
145 import org.apache.hadoop.hbase.trace.SpanReceiverHost;
146 import org.apache.hadoop.hbase.util.Addressing;
147 import org.apache.hadoop.hbase.util.ByteStringer;
148 import org.apache.hadoop.hbase.util.Bytes;
149 import org.apache.hadoop.hbase.util.CompressionTest;
150 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
151 import org.apache.hadoop.hbase.util.FSTableDescriptors;
152 import org.apache.hadoop.hbase.util.FSUtils;
153 import org.apache.hadoop.hbase.util.HasThread;
154 import org.apache.hadoop.hbase.util.JSONBean;
155 import org.apache.hadoop.hbase.util.JvmPauseMonitor;
156 import org.apache.hadoop.hbase.util.ServerRegionReplicaUtil;
157 import org.apache.hadoop.hbase.util.Sleeper;
158 import org.apache.hadoop.hbase.util.Threads;
159 import org.apache.hadoop.hbase.util.VersionInfo;
160 import org.apache.hadoop.hbase.wal.DefaultWALProvider;
161 import org.apache.hadoop.hbase.wal.WAL;
162 import org.apache.hadoop.hbase.wal.WALFactory;
163 import org.apache.hadoop.hbase.zookeeper.ClusterStatusTracker;
164 import org.apache.hadoop.hbase.zookeeper.MasterAddressTracker;
165 import org.apache.hadoop.hbase.zookeeper.MetaTableLocator;
166 import org.apache.hadoop.hbase.zookeeper.RecoveringRegionWatcher;
167 import org.apache.hadoop.hbase.zookeeper.ZKClusterId;
168 import org.apache.hadoop.hbase.zookeeper.ZKSplitLog;
169 import org.apache.hadoop.hbase.zookeeper.ZKUtil;
170 import org.apache.hadoop.hbase.zookeeper.ZooKeeperNodeTracker;
171 import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
172 import org.apache.hadoop.ipc.RemoteException;
173 import org.apache.hadoop.metrics.util.MBeanUtil;
174 import org.apache.hadoop.util.ReflectionUtils;
175 import org.apache.hadoop.util.StringUtils;
176 import org.apache.zookeeper.KeeperException;
177 import org.apache.zookeeper.KeeperException.NoNodeException;
178 import org.apache.zookeeper.data.Stat;
179 
180 import com.google.common.annotations.VisibleForTesting;
181 import com.google.common.base.Preconditions;
182 import com.google.common.collect.Maps;
183 import com.google.protobuf.BlockingRpcChannel;
184 import com.google.protobuf.Descriptors;
185 import com.google.protobuf.Message;
186 import com.google.protobuf.RpcCallback;
187 import com.google.protobuf.RpcController;
188 import com.google.protobuf.Service;
189 import com.google.protobuf.ServiceException;
190 
191 /**
192  * HRegionServer makes a set of HRegions available to clients. It checks in with
193  * the HMaster. There are many HRegionServers in a single HBase deployment.
194  */
195 @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.TOOLS)
196 @SuppressWarnings("deprecation")
197 public class HRegionServer extends HasThread implements
198     RegionServerServices, LastSequenceId {
199 
200   private static final Log LOG = LogFactory.getLog(HRegionServer.class);
201 
202   /**
203    * For testing only!  Set to true to skip notifying region assignment to master .
204    */
205   @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="MS_SHOULD_BE_FINAL")
206   public static boolean TEST_SKIP_REPORTING_TRANSITION = false;
207 
208   /*
209    * Strings to be used in forming the exception message for
210    * RegionsAlreadyInTransitionException.
211    */
212   protected static final String OPEN = "OPEN";
213   protected static final String CLOSE = "CLOSE";
214 
215   //RegionName vs current action in progress
216   //true - if open region action in progress
217   //false - if close region action in progress
218   protected final ConcurrentMap<byte[], Boolean> regionsInTransitionInRS =
219     new ConcurrentSkipListMap<byte[], Boolean>(Bytes.BYTES_COMPARATOR);
220 
221   // Cache flushing
222   protected MemStoreFlusher cacheFlusher;
223 
224   protected HeapMemoryManager hMemManager;
225 
226   /**
227    * Cluster connection to be shared by services.
228    * Initialized at server startup and closed when server shuts down.
229    * Clients must never close it explicitly.
230    */
231   protected ClusterConnection clusterConnection;
232 
233   /*
234    * Long-living meta table locator, which is created when the server is started and stopped
235    * when server shuts down. References to this locator shall be used to perform according
236    * operations in EventHandlers. Primary reason for this decision is to make it mockable
237    * for tests.
238    */
239   protected MetaTableLocator metaTableLocator;
240 
241   // Watch if a region is out of recovering state from ZooKeeper
242   @SuppressWarnings("unused")
243   private RecoveringRegionWatcher recoveringRegionWatcher;
244 
245   /**
246    * Go here to get table descriptors.
247    */
248   protected TableDescriptors tableDescriptors;
249 
250   // Replication services. If no replication, this handler will be null.
251   protected ReplicationSourceService replicationSourceHandler;
252   protected ReplicationSinkService replicationSinkHandler;
253 
254   // Compactions
255   public CompactSplitThread compactSplitThread;
256 
257   /**
258    * Map of regions currently being served by this region server. Key is the
259    * encoded region name.  All access should be synchronized.
260    */
261   protected final Map<String, Region> onlineRegions = new ConcurrentHashMap<String, Region>();
262 
263   /**
264    * Map of encoded region names to the DataNode locations they should be hosted on
265    * We store the value as InetSocketAddress since this is used only in HDFS
266    * API (create() that takes favored nodes as hints for placing file blocks).
267    * We could have used ServerName here as the value class, but we'd need to
268    * convert it to InetSocketAddress at some point before the HDFS API call, and
269    * it seems a bit weird to store ServerName since ServerName refers to RegionServers
270    * and here we really mean DataNode locations.
271    */
272   protected final Map<String, InetSocketAddress[]> regionFavoredNodesMap =
273       new ConcurrentHashMap<String, InetSocketAddress[]>();
274 
275   /**
276    * Set of regions currently being in recovering state which means it can accept writes(edits from
277    * previous failed region server) but not reads. A recovering region is also an online region.
278    */
279   protected final Map<String, Region> recoveringRegions = Collections
280       .synchronizedMap(new HashMap<String, Region>());
281 
282   // Leases
283   protected Leases leases;
284 
285   // Instance of the hbase executor service.
286   protected ExecutorService service;
287 
288   // If false, the file system has become unavailable
289   protected volatile boolean fsOk;
290   protected HFileSystem fs;
291 
292   // Set when a report to the master comes back with a message asking us to
293   // shutdown. Also set by call to stop when debugging or running unit tests
294   // of HRegionServer in isolation.
295   private volatile boolean stopped = false;
296 
297   // Go down hard. Used if file system becomes unavailable and also in
298   // debugging and unit tests.
299   private volatile boolean abortRequested;
300 
301   ConcurrentMap<String, Integer> rowlocks = new ConcurrentHashMap<String, Integer>();
302 
303   // A state before we go into stopped state.  At this stage we're closing user
304   // space regions.
305   private boolean stopping = false;
306 
307   private volatile boolean killed = false;
308 
309   protected final Configuration conf;
310 
311   private Path rootDir;
312 
313   protected final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
314 
315   final int numRetries;
316   protected final int threadWakeFrequency;
317   protected final int msgInterval;
318 
319   protected final int numRegionsToReport;
320 
321   // Stub to do region server status calls against the master.
322   private volatile RegionServerStatusService.BlockingInterface rssStub;
323   // RPC client. Used to make the stub above that does region server status checking.
324   RpcClient rpcClient;
325 
326   private RpcRetryingCallerFactory rpcRetryingCallerFactory;
327   private RpcControllerFactory rpcControllerFactory;
328 
329   private UncaughtExceptionHandler uncaughtExceptionHandler;
330 
331   // Info server. Default access so can be used by unit tests. REGIONSERVER
332   // is name of the webapp and the attribute name used stuffing this instance
333   // into web context.
334   protected InfoServer infoServer;
335   private JvmPauseMonitor pauseMonitor;
336 
337   /** region server process name */
338   public static final String REGIONSERVER = "regionserver";
339 
340   MetricsRegionServer metricsRegionServer;
341   private SpanReceiverHost spanReceiverHost;
342 
343   /**
344    * ChoreService used to schedule tasks that we want to run periodically
345    */
346   private final ChoreService choreService;
347 
348   /*
349    * Check for compactions requests.
350    */
351   ScheduledChore compactionChecker;
352 
353   /*
354    * Check for flushes
355    */
356   ScheduledChore periodicFlusher;
357 
358   protected volatile WALFactory walFactory;
359 
360   // WAL roller. log is protected rather than private to avoid
361   // eclipse warning when accessed by inner classes
362   final LogRoller walRoller;
363   // Lazily initialized if this RegionServer hosts a meta table.
364   final AtomicReference<LogRoller> metawalRoller = new AtomicReference<LogRoller>();
365 
366   // flag set after we're done setting up server threads
367   final AtomicBoolean online = new AtomicBoolean(false);
368 
369   // zookeeper connection and watcher
370   protected ZooKeeperWatcher zooKeeper;
371 
372   // master address tracker
373   private MasterAddressTracker masterAddressTracker;
374 
375   // Cluster Status Tracker
376   protected ClusterStatusTracker clusterStatusTracker;
377 
378   // Log Splitting Worker
379   private SplitLogWorker splitLogWorker;
380 
381   // A sleeper that sleeps for msgInterval.
382   protected final Sleeper sleeper;
383 
384   private final int operationTimeout;
385   private final int shortOperationTimeout;
386 
387   private final RegionServerAccounting regionServerAccounting;
388 
389   // Cache configuration and block cache reference
390   protected CacheConfig cacheConfig;
391   // Cache configuration for mob
392   final MobCacheConfig mobCacheConfig;
393 
394   /** The health check chore. */
395   private HealthCheckChore healthCheckChore;
396 
397   /** The nonce manager chore. */
398   private ScheduledChore nonceManagerChore;
399 
400   private Map<String, Service> coprocessorServiceHandlers = Maps.newHashMap();
401 
402   /**
403    * The server name the Master sees us as.  Its made from the hostname the
404    * master passes us, port, and server startcode. Gets set after registration
405    * against  Master.
406    */
407   protected ServerName serverName;
408 
409   /*
410    * hostname specified by hostname config
411    */
412   protected String useThisHostnameInstead;
413 
414   // key to the config parameter of server hostname
415   // the specification of server hostname is optional. The hostname should be resolvable from
416   // both master and region server
417   final static String RS_HOSTNAME_KEY = "hbase.regionserver.hostname";
418 
419   final static String MASTER_HOSTNAME_KEY = "hbase.master.hostname";
420 
421   /**
422    * This servers startcode.
423    */
424   protected final long startcode;
425 
426   /**
427    * Unique identifier for the cluster we are a part of.
428    */
429   private String clusterId;
430 
431   /**
432    * MX Bean for RegionServerInfo
433    */
434   private ObjectName mxBean = null;
435 
436   /**
437    * Chore to clean periodically the moved region list
438    */
439   private MovedRegionsCleaner movedRegionsCleaner;
440 
441   // chore for refreshing store files for secondary regions
442   private StorefileRefresherChore storefileRefresher;
443 
444   private RegionServerCoprocessorHost rsHost;
445 
446   private RegionServerProcedureManagerHost rspmHost;
447 
448   private RegionServerQuotaManager rsQuotaManager;
449 
450   // Table level lock manager for locking for region operations
451   protected TableLockManager tableLockManager;
452 
453   /**
454    * Nonce manager. Nonces are used to make operations like increment and append idempotent
455    * in the case where client doesn't receive the response from a successful operation and
456    * retries. We track the successful ops for some time via a nonce sent by client and handle
457    * duplicate operations (currently, by failing them; in future we might use MVCC to return
458    * result). Nonces are also recovered from WAL during, recovery; however, the caveats (from
459    * HBASE-3787) are:
460    * - WAL recovery is optimized, and under high load we won't read nearly nonce-timeout worth
461    *   of past records. If we don't read the records, we don't read and recover the nonces.
462    *   Some WALs within nonce-timeout at recovery may not even be present due to rolling/cleanup.
463    * - There's no WAL recovery during normal region move, so nonces will not be transfered.
464    * We can have separate additional "Nonce WAL". It will just contain bunch of numbers and
465    * won't be flushed on main path - because WAL itself also contains nonces, if we only flush
466    * it before memstore flush, for a given nonce we will either see it in the WAL (if it was
467    * never flushed to disk, it will be part of recovery), or we'll see it as part of the nonce
468    * log (or both occasionally, which doesn't matter). Nonce log file can be deleted after the
469    * latest nonce in it expired. It can also be recovered during move.
470    */
471   final ServerNonceManager nonceManager;
472 
473   private UserProvider userProvider;
474 
475   protected final RSRpcServices rpcServices;
476 
477   protected BaseCoordinatedStateManager csm;
478 
479   /**
480    * Configuration manager is used to register/deregister and notify the configuration observers
481    * when the regionserver is notified that there was a change in the on disk configs.
482    */
483   protected final ConfigurationManager configurationManager;
484 
485   /**
486    * Starts a HRegionServer at the default location.
487    * @param conf
488    * @throws IOException
489    * @throws InterruptedException
490    */
491   public HRegionServer(Configuration conf) throws IOException, InterruptedException {
492     this(conf, CoordinatedStateManagerFactory.getCoordinatedStateManager(conf));
493   }
494 
495   /**
496    * Starts a HRegionServer at the default location
497    * @param conf
498    * @param csm implementation of CoordinatedStateManager to be used
499    * @throws IOException
500    */
501   public HRegionServer(Configuration conf, CoordinatedStateManager csm)
502       throws IOException {
503     this.fsOk = true;
504     this.conf = conf;
505     HFile.checkHFileVersion(this.conf);
506     checkCodecs(this.conf);
507     this.userProvider = UserProvider.instantiate(conf);
508     Superusers.initialize(conf);
509     FSUtils.setupShortCircuitRead(this.conf);
510     // Disable usage of meta replicas in the regionserver
511     this.conf.setBoolean(HConstants.USE_META_REPLICAS, false);
512 
513     // Config'ed params
514     this.numRetries = this.conf.getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,
515         HConstants.DEFAULT_HBASE_CLIENT_RETRIES_NUMBER);
516     this.threadWakeFrequency = conf.getInt(HConstants.THREAD_WAKE_FREQUENCY, 10 * 1000);
517     this.msgInterval = conf.getInt("hbase.regionserver.msginterval", 3 * 1000);
518 
519     this.sleeper = new Sleeper(this.msgInterval, this);
520 
521     boolean isNoncesEnabled = conf.getBoolean(HConstants.HBASE_RS_NONCES_ENABLED, true);
522     this.nonceManager = isNoncesEnabled ? new ServerNonceManager(this.conf) : null;
523 
524     this.numRegionsToReport = conf.getInt(
525       "hbase.regionserver.numregionstoreport", 10);
526 
527     this.operationTimeout = conf.getInt(
528       HConstants.HBASE_CLIENT_OPERATION_TIMEOUT,
529       HConstants.DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT);
530 
531     this.shortOperationTimeout = conf.getInt(
532       HConstants.HBASE_RPC_SHORTOPERATION_TIMEOUT_KEY,
533       HConstants.DEFAULT_HBASE_RPC_SHORTOPERATION_TIMEOUT);
534 
535     this.abortRequested = false;
536     this.stopped = false;
537 
538     rpcServices = createRpcServices();
539     this.startcode = System.currentTimeMillis();
540     if (this instanceof HMaster) {
541       useThisHostnameInstead = conf.get(MASTER_HOSTNAME_KEY);
542     } else {
543       useThisHostnameInstead = conf.get(RS_HOSTNAME_KEY);
544     }
545     String hostName = shouldUseThisHostnameInstead() ? useThisHostnameInstead :
546       rpcServices.isa.getHostName();
547     serverName = ServerName.valueOf(hostName, rpcServices.isa.getPort(), startcode);
548 
549     rpcControllerFactory = RpcControllerFactory.instantiate(this.conf);
550     rpcRetryingCallerFactory = RpcRetryingCallerFactory.instantiate(this.conf);
551 
552     // login the zookeeper client principal (if using security)
553     ZKUtil.loginClient(this.conf, "hbase.zookeeper.client.keytab.file",
554       "hbase.zookeeper.client.kerberos.principal", hostName);
555     // login the server principal (if using secure Hadoop)
556     login(userProvider, hostName);
557 
558     regionServerAccounting = new RegionServerAccounting();
559     cacheConfig = new CacheConfig(conf);
560     mobCacheConfig = new MobCacheConfig(conf);
561     uncaughtExceptionHandler = new UncaughtExceptionHandler() {
562       @Override
563       public void uncaughtException(Thread t, Throwable e) {
564         abort("Uncaught exception in service thread " + t.getName(), e);
565       }
566     };
567 
568     // Set 'fs.defaultFS' to match the filesystem on hbase.rootdir else
569     // underlying hadoop hdfs accessors will be going against wrong filesystem
570     // (unless all is set to defaults).
571     FSUtils.setFsDefault(this.conf, FSUtils.getRootDir(this.conf));
572     // Get fs instance used by this RS.  Do we use checksum verification in the hbase? If hbase
573     // checksum verification enabled, then automatically switch off hdfs checksum verification.
574     boolean useHBaseChecksum = conf.getBoolean(HConstants.HBASE_CHECKSUM_VERIFICATION, true);
575     this.fs = new HFileSystem(this.conf, useHBaseChecksum);
576     this.rootDir = FSUtils.getRootDir(this.conf);
577     this.tableDescriptors = getFsTableDescriptors();
578 
579     service = new ExecutorService(getServerName().toShortString());
580     spanReceiverHost = SpanReceiverHost.getInstance(getConfiguration());
581 
582     // Some unit tests don't need a cluster, so no zookeeper at all
583     if (!conf.getBoolean("hbase.testing.nocluster", false)) {
584       // Open connection to zookeeper and set primary watcher
585       zooKeeper = new ZooKeeperWatcher(conf, getProcessName() + ":" +
586         rpcServices.isa.getPort(), this, canCreateBaseZNode());
587 
588       this.csm = (BaseCoordinatedStateManager) csm;
589       this.csm.initialize(this);
590       this.csm.start();
591 
592       tableLockManager = TableLockManager.createTableLockManager(
593         conf, zooKeeper, serverName);
594 
595       masterAddressTracker = new MasterAddressTracker(getZooKeeper(), this);
596       masterAddressTracker.start();
597 
598       clusterStatusTracker = new ClusterStatusTracker(zooKeeper, this);
599       clusterStatusTracker.start();
600     }
601     this.configurationManager = new ConfigurationManager();
602 
603     rpcServices.start();
604     putUpWebUI();
605     this.walRoller = new LogRoller(this, this);
606     this.choreService = new ChoreService(getServerName().toString());
607   }
608 
609   protected TableDescriptors getFsTableDescriptors() throws IOException {
610     return new FSTableDescriptors(this.conf,
611       this.fs, this.rootDir, !canUpdateTableDescriptor(), false);
612   }
613 
614   /*
615    * Returns true if configured hostname should be used
616    */
617   protected boolean shouldUseThisHostnameInstead() {
618     return useThisHostnameInstead != null && !useThisHostnameInstead.isEmpty();
619   }
620 
621   protected void login(UserProvider user, String host) throws IOException {
622     user.login("hbase.regionserver.keytab.file",
623       "hbase.regionserver.kerberos.principal", host);
624   }
625 
626   protected void waitForMasterActive(){
627   }
628 
629   protected String getProcessName() {
630     return REGIONSERVER;
631   }
632 
633   protected boolean canCreateBaseZNode() {
634     return false;
635   }
636 
637   protected boolean canUpdateTableDescriptor() {
638     return false;
639   }
640 
641   protected RSRpcServices createRpcServices() throws IOException {
642     return new RSRpcServices(this);
643   }
644 
645   protected void configureInfoServer() {
646     infoServer.addServlet("rs-status", "/rs-status", RSStatusServlet.class);
647     infoServer.setAttribute(REGIONSERVER, this);
648   }
649 
650   protected Class<? extends HttpServlet> getDumpServlet() {
651     return RSDumpServlet.class;
652   }
653 
654   protected void doMetrics() {
655   }
656 
657   @Override
658   public boolean registerService(Service instance) {
659     /*
660      * No stacking of instances is allowed for a single service name
661      */
662     Descriptors.ServiceDescriptor serviceDesc = instance.getDescriptorForType();
663     if (coprocessorServiceHandlers.containsKey(serviceDesc.getFullName())) {
664       LOG.error("Coprocessor service " + serviceDesc.getFullName()
665           + " already registered, rejecting request from " + instance);
666       return false;
667     }
668 
669     coprocessorServiceHandlers.put(serviceDesc.getFullName(), instance);
670     if (LOG.isDebugEnabled()) {
671       LOG.debug("Registered regionserver coprocessor service: service=" + serviceDesc.getFullName());
672     }
673     return true;
674   }
675 
676   /**
677    * Create a 'smarter' HConnection, one that is capable of by-passing RPC if the request is to
678    * the local server.  Safe to use going to local or remote server.
679    * Create this instance in a method can be intercepted and mocked in tests.
680    * @throws IOException
681    */
682   @VisibleForTesting
683   protected ClusterConnection createClusterConnection() throws IOException {
684     // Create a cluster connection that when appropriate, can short-circuit and go directly to the
685     // local server if the request is to the local server bypassing RPC. Can be used for both local
686     // and remote invocations.
687     return ConnectionUtils.createShortCircuitHConnection(
688       ConnectionFactory.createConnection(conf), serverName, rpcServices, rpcServices);
689   }
690 
691   /**
692    * Run test on configured codecs to make sure supporting libs are in place.
693    * @param c
694    * @throws IOException
695    */
696   private static void checkCodecs(final Configuration c) throws IOException {
697     // check to see if the codec list is available:
698     String [] codecs = c.getStrings("hbase.regionserver.codecs", (String[])null);
699     if (codecs == null) return;
700     for (String codec : codecs) {
701       if (!CompressionTest.testCompression(codec)) {
702         throw new IOException("Compression codec " + codec +
703           " not supported, aborting RS construction");
704       }
705     }
706   }
707 
708   public String getClusterId() {
709     return this.clusterId;
710   }
711 
712   /**
713    * Setup our cluster connection if not already initialized.
714    * @throws IOException
715    */
716   protected synchronized void setupClusterConnection() throws IOException {
717     if (clusterConnection == null) {
718       clusterConnection = createClusterConnection();
719       metaTableLocator = new MetaTableLocator();
720     }
721   }
722 
723   /**
724    * All initialization needed before we go register with Master.
725    *
726    * @throws IOException
727    * @throws InterruptedException
728    */
729   private void preRegistrationInitialization(){
730     try {
731       setupClusterConnection();
732 
733       // Health checker thread.
734       if (isHealthCheckerConfigured()) {
735         int sleepTime = this.conf.getInt(HConstants.HEALTH_CHORE_WAKE_FREQ,
736           HConstants.DEFAULT_THREAD_WAKE_FREQUENCY);
737         healthCheckChore = new HealthCheckChore(sleepTime, this, getConfiguration());
738       }
739       this.pauseMonitor = new JvmPauseMonitor(conf);
740       pauseMonitor.start();
741 
742       initializeZooKeeper();
743       if (!isStopped() && !isAborted()) {
744         initializeThreads();
745       }
746     } catch (Throwable t) {
747       // Call stop if error or process will stick around for ever since server
748       // puts up non-daemon threads.
749       this.rpcServices.stop();
750       abort("Initialization of RS failed.  Hence aborting RS.", t);
751     }
752   }
753 
754   /**
755    * Bring up connection to zk ensemble and then wait until a master for this
756    * cluster and then after that, wait until cluster 'up' flag has been set.
757    * This is the order in which master does things.
758    * Finally open long-living server short-circuit connection.
759    * @throws IOException
760    * @throws InterruptedException
761    */
762   private void initializeZooKeeper() throws IOException, InterruptedException {
763     // Create the master address tracker, register with zk, and start it.  Then
764     // block until a master is available.  No point in starting up if no master
765     // running.
766     blockAndCheckIfStopped(this.masterAddressTracker);
767 
768     // Wait on cluster being up.  Master will set this flag up in zookeeper
769     // when ready.
770     blockAndCheckIfStopped(this.clusterStatusTracker);
771 
772     // Retrieve clusterId
773     // Since cluster status is now up
774     // ID should have already been set by HMaster
775     try {
776       clusterId = ZKClusterId.readClusterIdZNode(this.zooKeeper);
777       if (clusterId == null) {
778         this.abort("Cluster ID has not been set");
779       }
780       LOG.info("ClusterId : "+clusterId);
781     } catch (KeeperException e) {
782       this.abort("Failed to retrieve Cluster ID",e);
783     }
784 
785     // In case colocated master, wait here till it's active.
786     // So backup masters won't start as regionservers.
787     // This is to avoid showing backup masters as regionservers
788     // in master web UI, or assigning any region to them.
789     waitForMasterActive();
790     if (isStopped() || isAborted()) {
791       return; // No need for further initialization
792     }
793 
794     // watch for snapshots and other procedures
795     try {
796       rspmHost = new RegionServerProcedureManagerHost();
797       rspmHost.loadProcedures(conf);
798       rspmHost.initialize(this);
799     } catch (KeeperException e) {
800       this.abort("Failed to reach zk cluster when creating procedure handler.", e);
801     }
802     // register watcher for recovering regions
803     this.recoveringRegionWatcher = new RecoveringRegionWatcher(this.zooKeeper, this);
804   }
805 
806   /**
807    * Utilty method to wait indefinitely on a znode availability while checking
808    * if the region server is shut down
809    * @param tracker znode tracker to use
810    * @throws IOException any IO exception, plus if the RS is stopped
811    * @throws InterruptedException
812    */
813   private void blockAndCheckIfStopped(ZooKeeperNodeTracker tracker)
814       throws IOException, InterruptedException {
815     while (tracker.blockUntilAvailable(this.msgInterval, false) == null) {
816       if (this.stopped) {
817         throw new IOException("Received the shutdown message while waiting.");
818       }
819     }
820   }
821 
822   /**
823    * @return False if cluster shutdown in progress
824    */
825   private boolean isClusterUp() {
826     return clusterStatusTracker != null && clusterStatusTracker.isClusterUp();
827   }
828 
829   private void initializeThreads() throws IOException {
830     // Cache flushing thread.
831     this.cacheFlusher = new MemStoreFlusher(conf, this);
832 
833     // Compaction thread
834     this.compactSplitThread = new CompactSplitThread(this);
835 
836     // Background thread to check for compactions; needed if region has not gotten updates
837     // in a while. It will take care of not checking too frequently on store-by-store basis.
838     this.compactionChecker = new CompactionChecker(this, this.threadWakeFrequency, this);
839     this.periodicFlusher = new PeriodicMemstoreFlusher(this.threadWakeFrequency, this);
840     this.leases = new Leases(this.threadWakeFrequency);
841 
842     // Create the thread to clean the moved regions list
843     movedRegionsCleaner = MovedRegionsCleaner.create(this);
844 
845     if (this.nonceManager != null) {
846       // Create the scheduled chore that cleans up nonces.
847       nonceManagerChore = this.nonceManager.createCleanupScheduledChore(this);
848     }
849 
850     // Setup the Quota Manager
851     rsQuotaManager = new RegionServerQuotaManager(this);
852 
853     // Setup RPC client for master communication
854     rpcClient = RpcClientFactory.createClient(conf, clusterId, new InetSocketAddress(
855         rpcServices.isa.getAddress(), 0));
856 
857     boolean onlyMetaRefresh = false;
858     int storefileRefreshPeriod = conf.getInt(
859         StorefileRefresherChore.REGIONSERVER_STOREFILE_REFRESH_PERIOD
860       , StorefileRefresherChore.DEFAULT_REGIONSERVER_STOREFILE_REFRESH_PERIOD);
861     if (storefileRefreshPeriod == 0) {
862       storefileRefreshPeriod = conf.getInt(
863           StorefileRefresherChore.REGIONSERVER_META_STOREFILE_REFRESH_PERIOD,
864           StorefileRefresherChore.DEFAULT_REGIONSERVER_STOREFILE_REFRESH_PERIOD);
865       onlyMetaRefresh = true;
866     }
867     if (storefileRefreshPeriod > 0) {
868       this.storefileRefresher = new StorefileRefresherChore(storefileRefreshPeriod,
869           onlyMetaRefresh, this, this);
870     }
871     registerConfigurationObservers();
872   }
873 
874   private void registerConfigurationObservers() {
875     // Registering the compactSplitThread object with the ConfigurationManager.
876     configurationManager.registerObserver(this.compactSplitThread);
877   }
878 
879   /**
880    * The HRegionServer sticks in this loop until closed.
881    */
882   @Override
883   public void run() {
884     try {
885       // Do pre-registration initializations; zookeeper, lease threads, etc.
886       preRegistrationInitialization();
887     } catch (Throwable e) {
888       abort("Fatal exception during initialization", e);
889     }
890 
891     try {
892       if (!isStopped() && !isAborted()) {
893         ShutdownHook.install(conf, fs, this, Thread.currentThread());
894         // Set our ephemeral znode up in zookeeper now we have a name.
895         createMyEphemeralNode();
896         // Initialize the RegionServerCoprocessorHost now that our ephemeral
897         // node was created, in case any coprocessors want to use ZooKeeper
898         this.rsHost = new RegionServerCoprocessorHost(this, this.conf);
899       }
900 
901       // Try and register with the Master; tell it we are here.  Break if
902       // server is stopped or the clusterup flag is down or hdfs went wacky.
903       while (keepLooping()) {
904         RegionServerStartupResponse w = reportForDuty();
905         if (w == null) {
906           LOG.warn("reportForDuty failed; sleeping and then retrying.");
907           this.sleeper.sleep();
908         } else {
909           handleReportForDutyResponse(w);
910           break;
911         }
912       }
913 
914       if (!isStopped() && isHealthy()){
915         // start the snapshot handler and other procedure handlers,
916         // since the server is ready to run
917         rspmHost.start();
918 
919         // Start the Quota Manager
920         rsQuotaManager.start(getRpcServer().getScheduler());
921       }
922 
923       // We registered with the Master.  Go into run mode.
924       long lastMsg = System.currentTimeMillis();
925       long oldRequestCount = -1;
926       // The main run loop.
927       while (!isStopped() && isHealthy()) {
928         if (!isClusterUp()) {
929           if (isOnlineRegionsEmpty()) {
930             stop("Exiting; cluster shutdown set and not carrying any regions");
931           } else if (!this.stopping) {
932             this.stopping = true;
933             LOG.info("Closing user regions");
934             closeUserRegions(this.abortRequested);
935           } else if (this.stopping) {
936             boolean allUserRegionsOffline = areAllUserRegionsOffline();
937             if (allUserRegionsOffline) {
938               // Set stopped if no more write requests tp meta tables
939               // since last time we went around the loop.  Any open
940               // meta regions will be closed on our way out.
941               if (oldRequestCount == getWriteRequestCount()) {
942                 stop("Stopped; only catalog regions remaining online");
943                 break;
944               }
945               oldRequestCount = getWriteRequestCount();
946             } else {
947               // Make sure all regions have been closed -- some regions may
948               // have not got it because we were splitting at the time of
949               // the call to closeUserRegions.
950               closeUserRegions(this.abortRequested);
951             }
952             LOG.debug("Waiting on " + getOnlineRegionsAsPrintableString());
953           }
954         }
955         long now = System.currentTimeMillis();
956         if ((now - lastMsg) >= msgInterval) {
957           tryRegionServerReport(lastMsg, now);
958           lastMsg = System.currentTimeMillis();
959           doMetrics();
960         }
961         if (!isStopped() && !isAborted()) {
962           this.sleeper.sleep();
963         }
964       } // for
965     } catch (Throwable t) {
966       if (!rpcServices.checkOOME(t)) {
967         String prefix = t instanceof YouAreDeadException? "": "Unhandled: ";
968         abort(prefix + t.getMessage(), t);
969       }
970     }
971     // Run shutdown.
972     if (mxBean != null) {
973       MBeanUtil.unregisterMBean(mxBean);
974       mxBean = null;
975     }
976     if (this.leases != null) this.leases.closeAfterLeasesExpire();
977     if (this.splitLogWorker != null) {
978       splitLogWorker.stop();
979     }
980     if (this.infoServer != null) {
981       LOG.info("Stopping infoServer");
982       try {
983         this.infoServer.stop();
984       } catch (Exception e) {
985         LOG.error("Failed to stop infoServer", e);
986       }
987     }
988     // Send cache a shutdown.
989     if (cacheConfig != null && cacheConfig.isBlockCacheEnabled()) {
990       cacheConfig.getBlockCache().shutdown();
991     }
992     mobCacheConfig.getMobFileCache().shutdown();
993 
994     if (movedRegionsCleaner != null) {
995       movedRegionsCleaner.stop("Region Server stopping");
996     }
997 
998     // Send interrupts to wake up threads if sleeping so they notice shutdown.
999     // TODO: Should we check they are alive? If OOME could have exited already
1000     if (this.hMemManager != null) this.hMemManager.stop();
1001     if (this.cacheFlusher != null) this.cacheFlusher.interruptIfNecessary();
1002     if (this.compactSplitThread != null) this.compactSplitThread.interruptIfNecessary();
1003     if (this.compactionChecker != null) this.compactionChecker.cancel(true);
1004     if (this.healthCheckChore != null) this.healthCheckChore.cancel(true);
1005     if (this.nonceManagerChore != null) this.nonceManagerChore.cancel(true);
1006     if (this.storefileRefresher != null) this.storefileRefresher.cancel(true);
1007     sendShutdownInterrupt();
1008 
1009     // Stop the quota manager
1010     if (rsQuotaManager != null) {
1011       rsQuotaManager.stop();
1012     }
1013 
1014     // Stop the snapshot and other procedure handlers, forcefully killing all running tasks
1015     if (rspmHost != null) {
1016       rspmHost.stop(this.abortRequested || this.killed);
1017     }
1018 
1019     if (this.killed) {
1020       // Just skip out w/o closing regions.  Used when testing.
1021     } else if (abortRequested) {
1022       if (this.fsOk) {
1023         closeUserRegions(abortRequested); // Don't leave any open file handles
1024       }
1025       LOG.info("aborting server " + this.serverName);
1026     } else {
1027       closeUserRegions(abortRequested);
1028       LOG.info("stopping server " + this.serverName);
1029     }
1030 
1031     // so callers waiting for meta without timeout can stop
1032     if (this.metaTableLocator != null) this.metaTableLocator.stop();
1033     if (this.clusterConnection != null && !clusterConnection.isClosed()) {
1034       try {
1035         this.clusterConnection.close();
1036       } catch (IOException e) {
1037         // Although the {@link Closeable} interface throws an {@link
1038         // IOException}, in reality, the implementation would never do that.
1039         LOG.warn("Attempt to close server's short circuit HConnection failed.", e);
1040       }
1041     }
1042 
1043     // Closing the compactSplit thread before closing meta regions
1044     if (!this.killed && containsMetaTableRegions()) {
1045       if (!abortRequested || this.fsOk) {
1046         if (this.compactSplitThread != null) {
1047           this.compactSplitThread.join();
1048           this.compactSplitThread = null;
1049         }
1050         closeMetaTableRegions(abortRequested);
1051       }
1052     }
1053 
1054     if (!this.killed && this.fsOk) {
1055       waitOnAllRegionsToClose(abortRequested);
1056       LOG.info("stopping server " + this.serverName +
1057         "; all regions closed.");
1058     }
1059 
1060     //fsOk flag may be changed when closing regions throws exception.
1061     if (this.fsOk) {
1062       shutdownWAL(!abortRequested);
1063     }
1064 
1065     // Make sure the proxy is down.
1066     if (this.rssStub != null) {
1067       this.rssStub = null;
1068     }
1069     if (this.rpcClient != null) {
1070       this.rpcClient.close();
1071     }
1072     if (this.leases != null) {
1073       this.leases.close();
1074     }
1075     if (this.pauseMonitor != null) {
1076       this.pauseMonitor.stop();
1077     }
1078 
1079     if (!killed) {
1080       stopServiceThreads();
1081     }
1082 
1083     if (this.rpcServices != null) {
1084       this.rpcServices.stop();
1085     }
1086 
1087     try {
1088       deleteMyEphemeralNode();
1089     } catch (KeeperException.NoNodeException nn) {
1090     } catch (KeeperException e) {
1091       LOG.warn("Failed deleting my ephemeral node", e);
1092     }
1093     // We may have failed to delete the znode at the previous step, but
1094     //  we delete the file anyway: a second attempt to delete the znode is likely to fail again.
1095     ZNodeClearer.deleteMyEphemeralNodeOnDisk();
1096 
1097     if (this.zooKeeper != null) {
1098       this.zooKeeper.close();
1099     }
1100     LOG.info("stopping server " + this.serverName +
1101       "; zookeeper connection closed.");
1102 
1103     LOG.info(Thread.currentThread().getName() + " exiting");
1104   }
1105 
1106   private boolean containsMetaTableRegions() {
1107     return onlineRegions.containsKey(HRegionInfo.FIRST_META_REGIONINFO.getEncodedName());
1108   }
1109 
1110   private boolean areAllUserRegionsOffline() {
1111     if (getNumberOfOnlineRegions() > 2) return false;
1112     boolean allUserRegionsOffline = true;
1113     for (Map.Entry<String, Region> e: this.onlineRegions.entrySet()) {
1114       if (!e.getValue().getRegionInfo().isMetaTable()) {
1115         allUserRegionsOffline = false;
1116         break;
1117       }
1118     }
1119     return allUserRegionsOffline;
1120   }
1121 
1122   /**
1123    * @return Current write count for all online regions.
1124    */
1125   private long getWriteRequestCount() {
1126     long writeCount = 0;
1127     for (Map.Entry<String, Region> e: this.onlineRegions.entrySet()) {
1128       writeCount += e.getValue().getWriteRequestsCount();
1129     }
1130     return writeCount;
1131   }
1132 
1133   @VisibleForTesting
1134   protected void tryRegionServerReport(long reportStartTime, long reportEndTime)
1135   throws IOException {
1136     RegionServerStatusService.BlockingInterface rss = rssStub;
1137     if (rss == null) {
1138       // the current server could be stopping.
1139       return;
1140     }
1141     ClusterStatusProtos.ServerLoad sl = buildServerLoad(reportStartTime, reportEndTime);
1142     try {
1143       RegionServerReportRequest.Builder request = RegionServerReportRequest.newBuilder();
1144       ServerName sn = ServerName.parseVersionedServerName(
1145         this.serverName.getVersionedBytes());
1146       request.setServer(ProtobufUtil.toServerName(sn));
1147       request.setLoad(sl);
1148       rss.regionServerReport(null, request.build());
1149     } catch (ServiceException se) {
1150       IOException ioe = ProtobufUtil.getRemoteException(se);
1151       if (ioe instanceof YouAreDeadException) {
1152         // This will be caught and handled as a fatal error in run()
1153         throw ioe;
1154       }
1155       if (rssStub == rss) {
1156         rssStub = null;
1157       }
1158       // Couldn't connect to the master, get location from zk and reconnect
1159       // Method blocks until new master is found or we are stopped
1160       createRegionServerStatusStub();
1161     }
1162   }
1163 
1164   ClusterStatusProtos.ServerLoad buildServerLoad(long reportStartTime, long reportEndTime)
1165       throws IOException {
1166     // We're getting the MetricsRegionServerWrapper here because the wrapper computes requests
1167     // per second, and other metrics  As long as metrics are part of ServerLoad it's best to use
1168     // the wrapper to compute those numbers in one place.
1169     // In the long term most of these should be moved off of ServerLoad and the heart beat.
1170     // Instead they should be stored in an HBase table so that external visibility into HBase is
1171     // improved; Additionally the load balancer will be able to take advantage of a more complete
1172     // history.
1173     MetricsRegionServerWrapper regionServerWrapper = metricsRegionServer.getRegionServerWrapper();
1174     Collection<Region> regions = getOnlineRegionsLocalContext();
1175     MemoryUsage memory = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage();
1176 
1177     ClusterStatusProtos.ServerLoad.Builder serverLoad =
1178       ClusterStatusProtos.ServerLoad.newBuilder();
1179     serverLoad.setNumberOfRequests((int) regionServerWrapper.getRequestsPerSecond());
1180     serverLoad.setTotalNumberOfRequests((int) regionServerWrapper.getTotalRequestCount());
1181     serverLoad.setUsedHeapMB((int)(memory.getUsed() / 1024 / 1024));
1182     serverLoad.setMaxHeapMB((int) (memory.getMax() / 1024 / 1024));
1183     Set<String> coprocessors = getWAL(null).getCoprocessorHost().getCoprocessors();
1184     Builder coprocessorBuilder = Coprocessor.newBuilder();
1185     for (String coprocessor : coprocessors) {
1186       serverLoad.addCoprocessors(coprocessorBuilder.setName(coprocessor).build());
1187     }
1188     RegionLoad.Builder regionLoadBldr = RegionLoad.newBuilder();
1189     RegionSpecifier.Builder regionSpecifier = RegionSpecifier.newBuilder();
1190     for (Region region : regions) {
1191       if (region.getCoprocessorHost() != null) {
1192         Set<String> regionCoprocessors = region.getCoprocessorHost().getCoprocessors();
1193         Iterator<String> iterator = regionCoprocessors.iterator();
1194         while (iterator.hasNext()) {
1195           serverLoad.addCoprocessors(coprocessorBuilder.setName(iterator.next()).build());
1196         }
1197       }
1198       serverLoad.addRegionLoads(createRegionLoad(region, regionLoadBldr, regionSpecifier));
1199       for (String coprocessor : getWAL(region.getRegionInfo()).getCoprocessorHost()
1200           .getCoprocessors()) {
1201         serverLoad.addCoprocessors(coprocessorBuilder.setName(coprocessor).build());
1202       }
1203     }
1204     serverLoad.setReportStartTime(reportStartTime);
1205     serverLoad.setReportEndTime(reportEndTime);
1206     if (this.infoServer != null) {
1207       serverLoad.setInfoServerPort(this.infoServer.getPort());
1208     } else {
1209       serverLoad.setInfoServerPort(-1);
1210     }
1211 
1212     // for the replicationLoad purpose. Only need to get from one service
1213     // either source or sink will get the same info
1214     ReplicationSourceService rsources = getReplicationSourceService();
1215 
1216     if (rsources != null) {
1217       // always refresh first to get the latest value
1218       ReplicationLoad rLoad = rsources.refreshAndGetReplicationLoad();
1219       if (rLoad != null) {
1220         serverLoad.setReplLoadSink(rLoad.getReplicationLoadSink());
1221         for (ClusterStatusProtos.ReplicationLoadSource rLS : rLoad.getReplicationLoadSourceList()) {
1222           serverLoad.addReplLoadSource(rLS);
1223         }
1224       }
1225     }
1226 
1227     return serverLoad.build();
1228   }
1229 
1230   String getOnlineRegionsAsPrintableString() {
1231     StringBuilder sb = new StringBuilder();
1232     for (Region r: this.onlineRegions.values()) {
1233       if (sb.length() > 0) sb.append(", ");
1234       sb.append(r.getRegionInfo().getEncodedName());
1235     }
1236     return sb.toString();
1237   }
1238 
1239   /**
1240    * Wait on regions close.
1241    */
1242   private void waitOnAllRegionsToClose(final boolean abort) {
1243     // Wait till all regions are closed before going out.
1244     int lastCount = -1;
1245     long previousLogTime = 0;
1246     Set<String> closedRegions = new HashSet<String>();
1247     boolean interrupted = false;
1248     try {
1249       while (!isOnlineRegionsEmpty()) {
1250         int count = getNumberOfOnlineRegions();
1251         // Only print a message if the count of regions has changed.
1252         if (count != lastCount) {
1253           // Log every second at most
1254           if (System.currentTimeMillis() > (previousLogTime + 1000)) {
1255             previousLogTime = System.currentTimeMillis();
1256             lastCount = count;
1257             LOG.info("Waiting on " + count + " regions to close");
1258             // Only print out regions still closing if a small number else will
1259             // swamp the log.
1260             if (count < 10 && LOG.isDebugEnabled()) {
1261               LOG.debug(this.onlineRegions);
1262             }
1263           }
1264         }
1265         // Ensure all user regions have been sent a close. Use this to
1266         // protect against the case where an open comes in after we start the
1267         // iterator of onlineRegions to close all user regions.
1268         for (Map.Entry<String, Region> e : this.onlineRegions.entrySet()) {
1269           HRegionInfo hri = e.getValue().getRegionInfo();
1270           if (!this.regionsInTransitionInRS.containsKey(hri.getEncodedNameAsBytes())
1271               && !closedRegions.contains(hri.getEncodedName())) {
1272             closedRegions.add(hri.getEncodedName());
1273             // Don't update zk with this close transition; pass false.
1274             closeRegionIgnoreErrors(hri, abort);
1275               }
1276         }
1277         // No regions in RIT, we could stop waiting now.
1278         if (this.regionsInTransitionInRS.isEmpty()) {
1279           if (!isOnlineRegionsEmpty()) {
1280             LOG.info("We were exiting though online regions are not empty," +
1281                 " because some regions failed closing");
1282           }
1283           break;
1284         }
1285         if (sleep(200)) {
1286           interrupted = true;
1287         }
1288       }
1289     } finally {
1290       if (interrupted) {
1291         Thread.currentThread().interrupt();
1292       }
1293     }
1294   }
1295 
1296   private boolean sleep(long millis) {
1297     boolean interrupted = false;
1298     try {
1299       Thread.sleep(millis);
1300     } catch (InterruptedException e) {
1301       LOG.warn("Interrupted while sleeping");
1302       interrupted = true;
1303     }
1304     return interrupted;
1305   }
1306 
1307   private void shutdownWAL(final boolean close) {
1308     if (this.walFactory != null) {
1309       try {
1310         if (close) {
1311           walFactory.close();
1312         } else {
1313           walFactory.shutdown();
1314         }
1315       } catch (Throwable e) {
1316         e = e instanceof RemoteException ? ((RemoteException) e).unwrapRemoteException() : e;
1317         LOG.error("Shutdown / close of WAL failed: " + e);
1318         LOG.debug("Shutdown / close exception details:", e);
1319       }
1320     }
1321   }
1322 
1323   /*
1324    * Run init. Sets up wal and starts up all server threads.
1325    *
1326    * @param c Extra configuration.
1327    */
1328   protected void handleReportForDutyResponse(final RegionServerStartupResponse c)
1329   throws IOException {
1330     try {
1331       for (NameStringPair e : c.getMapEntriesList()) {
1332         String key = e.getName();
1333         // The hostname the master sees us as.
1334         if (key.equals(HConstants.KEY_FOR_HOSTNAME_SEEN_BY_MASTER)) {
1335           String hostnameFromMasterPOV = e.getValue();
1336           this.serverName = ServerName.valueOf(hostnameFromMasterPOV,
1337             rpcServices.isa.getPort(), this.startcode);
1338           if (shouldUseThisHostnameInstead() &&
1339               !hostnameFromMasterPOV.equals(useThisHostnameInstead)) {
1340             String msg = "Master passed us a different hostname to use; was=" +
1341                 this.useThisHostnameInstead + ", but now=" + hostnameFromMasterPOV;
1342             LOG.error(msg);
1343             throw new IOException(msg);
1344           }
1345           if (!shouldUseThisHostnameInstead() &&
1346               !hostnameFromMasterPOV.equals(rpcServices.isa.getHostName())) {
1347             String msg = "Master passed us a different hostname to use; was=" +
1348                 rpcServices.isa.getHostName() + ", but now=" + hostnameFromMasterPOV;
1349             LOG.error(msg);
1350           }
1351           continue;
1352         }
1353         String value = e.getValue();
1354         if (LOG.isDebugEnabled()) {
1355           LOG.info("Config from master: " + key + "=" + value);
1356         }
1357         this.conf.set(key, value);
1358       }
1359 
1360       // hack! Maps DFSClient => RegionServer for logs.  HDFS made this
1361       // config param for task trackers, but we can piggyback off of it.
1362       if (this.conf.get("mapreduce.task.attempt.id") == null) {
1363         this.conf.set("mapreduce.task.attempt.id", "hb_rs_" +
1364           this.serverName.toString());
1365       }
1366 
1367       // Save it in a file, this will allow to see if we crash
1368       ZNodeClearer.writeMyEphemeralNodeOnDisk(getMyEphemeralNodePath());
1369 
1370       this.cacheConfig = new CacheConfig(conf);
1371       this.walFactory = setupWALAndReplication();
1372       // Init in here rather than in constructor after thread name has been set
1373       this.metricsRegionServer = new MetricsRegionServer(new MetricsRegionServerWrapperImpl(this));
1374 
1375       startServiceThreads();
1376       startHeapMemoryManager();
1377       LOG.info("Serving as " + this.serverName +
1378         ", RpcServer on " + rpcServices.isa +
1379         ", sessionid=0x" +
1380         Long.toHexString(this.zooKeeper.getRecoverableZooKeeper().getSessionId()));
1381 
1382       // Wake up anyone waiting for this server to online
1383       synchronized (online) {
1384         online.set(true);
1385         online.notifyAll();
1386       }
1387     } catch (Throwable e) {
1388       stop("Failed initialization");
1389       throw convertThrowableToIOE(cleanup(e, "Failed init"),
1390           "Region server startup failed");
1391     } finally {
1392       sleeper.skipSleepCycle();
1393     }
1394   }
1395 
1396   private void startHeapMemoryManager() {
1397     this.hMemManager = HeapMemoryManager.create(this.conf, this.cacheFlusher,
1398         this, this.regionServerAccounting);
1399     if (this.hMemManager != null) {
1400       this.hMemManager.start(getChoreService());
1401     }
1402   }
1403 
1404   private void createMyEphemeralNode() throws KeeperException, IOException {
1405     RegionServerInfo.Builder rsInfo = RegionServerInfo.newBuilder();
1406     rsInfo.setInfoPort(infoServer != null ? infoServer.getPort() : -1);
1407     byte[] data = ProtobufUtil.prependPBMagic(rsInfo.build().toByteArray());
1408     ZKUtil.createEphemeralNodeAndWatch(this.zooKeeper,
1409       getMyEphemeralNodePath(), data);
1410   }
1411 
1412   private void deleteMyEphemeralNode() throws KeeperException {
1413     ZKUtil.deleteNode(this.zooKeeper, getMyEphemeralNodePath());
1414   }
1415 
1416   @Override
1417   public RegionServerAccounting getRegionServerAccounting() {
1418     return regionServerAccounting;
1419   }
1420 
1421   @Override
1422   public TableLockManager getTableLockManager() {
1423     return tableLockManager;
1424   }
1425 
1426   /*
1427    * @param r Region to get RegionLoad for.
1428    * @param regionLoadBldr the RegionLoad.Builder, can be null
1429    * @param regionSpecifier the RegionSpecifier.Builder, can be null
1430    * @return RegionLoad instance.
1431    *
1432    * @throws IOException
1433    */
1434   private RegionLoad createRegionLoad(final Region r, RegionLoad.Builder regionLoadBldr,
1435       RegionSpecifier.Builder regionSpecifier) throws IOException {
1436     byte[] name = r.getRegionInfo().getRegionName();
1437     int stores = 0;
1438     int storefiles = 0;
1439     int storeUncompressedSizeMB = 0;
1440     int storefileSizeMB = 0;
1441     int memstoreSizeMB = (int) (r.getMemstoreSize() / 1024 / 1024);
1442     int storefileIndexSizeMB = 0;
1443     int rootIndexSizeKB = 0;
1444     int totalStaticIndexSizeKB = 0;
1445     int totalStaticBloomSizeKB = 0;
1446     long totalCompactingKVs = 0;
1447     long currentCompactedKVs = 0;
1448     List<Store> storeList = r.getStores();
1449     stores += storeList.size();
1450     for (Store store : storeList) {
1451       storefiles += store.getStorefilesCount();
1452       storeUncompressedSizeMB += (int) (store.getStoreSizeUncompressed() / 1024 / 1024);
1453       storefileSizeMB += (int) (store.getStorefilesSize() / 1024 / 1024);
1454       storefileIndexSizeMB += (int) (store.getStorefilesIndexSize() / 1024 / 1024);
1455       CompactionProgress progress = store.getCompactionProgress();
1456       if (progress != null) {
1457         totalCompactingKVs += progress.totalCompactingKVs;
1458         currentCompactedKVs += progress.currentCompactedKVs;
1459       }
1460       rootIndexSizeKB += (int) (store.getStorefilesIndexSize() / 1024);
1461       totalStaticIndexSizeKB += (int) (store.getTotalStaticIndexSize() / 1024);
1462       totalStaticBloomSizeKB += (int) (store.getTotalStaticBloomSize() / 1024);
1463     }
1464 
1465     float dataLocality =
1466         r.getHDFSBlocksDistribution().getBlockLocalityIndex(serverName.getHostname());
1467     if (regionLoadBldr == null) {
1468       regionLoadBldr = RegionLoad.newBuilder();
1469     }
1470     if (regionSpecifier == null) {
1471       regionSpecifier = RegionSpecifier.newBuilder();
1472     }
1473     regionSpecifier.setType(RegionSpecifierType.REGION_NAME);
1474     regionSpecifier.setValue(ByteStringer.wrap(name));
1475     regionLoadBldr.setRegionSpecifier(regionSpecifier.build())
1476       .setStores(stores)
1477       .setStorefiles(storefiles)
1478       .setStoreUncompressedSizeMB(storeUncompressedSizeMB)
1479       .setStorefileSizeMB(storefileSizeMB)
1480       .setMemstoreSizeMB(memstoreSizeMB)
1481       .setStorefileIndexSizeMB(storefileIndexSizeMB)
1482       .setRootIndexSizeKB(rootIndexSizeKB)
1483       .setTotalStaticIndexSizeKB(totalStaticIndexSizeKB)
1484       .setTotalStaticBloomSizeKB(totalStaticBloomSizeKB)
1485       .setReadRequestsCount(r.getReadRequestsCount())
1486       .setWriteRequestsCount(r.getWriteRequestsCount())
1487       .setTotalCompactingKVs(totalCompactingKVs)
1488       .setCurrentCompactedKVs(currentCompactedKVs)
1489       .setDataLocality(dataLocality)
1490       .setLastMajorCompactionTs(r.getOldestHfileTs(true));
1491     ((HRegion)r).setCompleteSequenceId(regionLoadBldr);
1492 
1493     return regionLoadBldr.build();
1494   }
1495 
1496   /**
1497    * @param encodedRegionName
1498    * @return An instance of RegionLoad.
1499    */
1500   public RegionLoad createRegionLoad(final String encodedRegionName) throws IOException {
1501     Region r = onlineRegions.get(encodedRegionName);
1502     return r != null ? createRegionLoad(r, null, null) : null;
1503   }
1504 
1505   /*
1506    * Inner class that runs on a long period checking if regions need compaction.
1507    */
1508   private static class CompactionChecker extends ScheduledChore {
1509     private final HRegionServer instance;
1510     private final int majorCompactPriority;
1511     private final static int DEFAULT_PRIORITY = Integer.MAX_VALUE;
1512     private long iteration = 0;
1513 
1514     CompactionChecker(final HRegionServer h, final int sleepTime,
1515         final Stoppable stopper) {
1516       super("CompactionChecker", stopper, sleepTime);
1517       this.instance = h;
1518       LOG.info(this.getName() + " runs every " + StringUtils.formatTime(sleepTime));
1519 
1520       /* MajorCompactPriority is configurable.
1521        * If not set, the compaction will use default priority.
1522        */
1523       this.majorCompactPriority = this.instance.conf.
1524         getInt("hbase.regionserver.compactionChecker.majorCompactPriority",
1525         DEFAULT_PRIORITY);
1526     }
1527 
1528     @Override
1529     protected void chore() {
1530       for (Region r : this.instance.onlineRegions.values()) {
1531         if (r == null)
1532           continue;
1533         for (Store s : r.getStores()) {
1534           try {
1535             long multiplier = s.getCompactionCheckMultiplier();
1536             assert multiplier > 0;
1537             if (iteration % multiplier != 0) continue;
1538             if (s.needsCompaction()) {
1539               // Queue a compaction. Will recognize if major is needed.
1540               this.instance.compactSplitThread.requestSystemCompaction(r, s, getName()
1541                   + " requests compaction");
1542             } else if (s.isMajorCompaction()) {
1543               if (majorCompactPriority == DEFAULT_PRIORITY
1544                   || majorCompactPriority > ((HRegion)r).getCompactPriority()) {
1545                 this.instance.compactSplitThread.requestCompaction(r, s, getName()
1546                     + " requests major compaction; use default priority", null);
1547               } else {
1548                 this.instance.compactSplitThread.requestCompaction(r, s, getName()
1549                     + " requests major compaction; use configured priority",
1550                   this.majorCompactPriority, null);
1551               }
1552             }
1553           } catch (IOException e) {
1554             LOG.warn("Failed major compaction check on " + r, e);
1555           }
1556         }
1557       }
1558       iteration = (iteration == Long.MAX_VALUE) ? 0 : (iteration + 1);
1559     }
1560   }
1561 
1562   static class PeriodicMemstoreFlusher extends ScheduledChore {
1563     final HRegionServer server;
1564     final static int RANGE_OF_DELAY = 20000; //millisec
1565     final static int MIN_DELAY_TIME = 3000; //millisec
1566     public PeriodicMemstoreFlusher(int cacheFlushInterval, final HRegionServer server) {
1567       super(server.getServerName() + "-MemstoreFlusherChore", server, cacheFlushInterval);
1568       this.server = server;
1569     }
1570 
1571     @Override
1572     protected void chore() {
1573       final StringBuffer whyFlush = new StringBuffer();
1574       for (Region r : this.server.onlineRegions.values()) {
1575         if (r == null) continue;
1576         if (((HRegion)r).shouldFlush(whyFlush)) {
1577           FlushRequester requester = server.getFlushRequester();
1578           if (requester != null) {
1579             long randomDelay = RandomUtils.nextInt(RANGE_OF_DELAY) + MIN_DELAY_TIME;
1580             LOG.info(getName() + " requesting flush of " +
1581               r.getRegionInfo().getRegionNameAsString() + " because " +
1582               whyFlush.toString() +
1583               " after random delay " + randomDelay + "ms");
1584             //Throttle the flushes by putting a delay. If we don't throttle, and there
1585             //is a balanced write-load on the regions in a table, we might end up
1586             //overwhelming the filesystem with too many flushes at once.
1587             requester.requestDelayedFlush(r, randomDelay, false);
1588           }
1589         }
1590       }
1591     }
1592   }
1593 
1594   /**
1595    * Report the status of the server. A server is online once all the startup is
1596    * completed (setting up filesystem, starting service threads, etc.). This
1597    * method is designed mostly to be useful in tests.
1598    *
1599    * @return true if online, false if not.
1600    */
1601   public boolean isOnline() {
1602     return online.get();
1603   }
1604 
1605   /**
1606    * Setup WAL log and replication if enabled.
1607    * Replication setup is done in here because it wants to be hooked up to WAL.
1608    * @return A WAL instance.
1609    * @throws IOException
1610    */
1611   private WALFactory setupWALAndReplication() throws IOException {
1612     // TODO Replication make assumptions here based on the default filesystem impl
1613     final Path oldLogDir = new Path(rootDir, HConstants.HREGION_OLDLOGDIR_NAME);
1614     final String logName = DefaultWALProvider.getWALDirectoryName(this.serverName.toString());
1615 
1616     Path logdir = new Path(rootDir, logName);
1617     if (LOG.isDebugEnabled()) LOG.debug("logdir=" + logdir);
1618     if (this.fs.exists(logdir)) {
1619       throw new RegionServerRunningException("Region server has already " +
1620         "created directory at " + this.serverName.toString());
1621     }
1622 
1623     // Instantiate replication manager if replication enabled.  Pass it the
1624     // log directories.
1625     createNewReplicationInstance(conf, this, this.fs, logdir, oldLogDir);
1626 
1627     // listeners the wal factory will add to wals it creates.
1628     final List<WALActionsListener> listeners = new ArrayList<WALActionsListener>();
1629     listeners.add(new MetricsWAL());
1630     if (this.replicationSourceHandler != null &&
1631         this.replicationSourceHandler.getWALActionsListener() != null) {
1632       // Replication handler is an implementation of WALActionsListener.
1633       listeners.add(this.replicationSourceHandler.getWALActionsListener());
1634     }
1635 
1636     return new WALFactory(conf, listeners, serverName.toString());
1637   }
1638 
1639   /**
1640    * We initialize the roller for the wal that handles meta lazily
1641    * since we don't know if this regionserver will handle it. All calls to
1642    * this method return a reference to the that same roller. As newly referenced
1643    * meta regions are brought online, they will be offered to the roller for maintenance.
1644    * As a part of that registration process, the roller will add itself as a
1645    * listener on the wal.
1646    */
1647   protected LogRoller ensureMetaWALRoller() {
1648     // Using a tmp log roller to ensure metaLogRoller is alive once it is not
1649     // null
1650     LogRoller roller = metawalRoller.get();
1651     if (null == roller) {
1652       LogRoller tmpLogRoller = new LogRoller(this, this);
1653       String n = Thread.currentThread().getName();
1654       Threads.setDaemonThreadRunning(tmpLogRoller.getThread(),
1655           n + "-MetaLogRoller", uncaughtExceptionHandler);
1656       if (metawalRoller.compareAndSet(null, tmpLogRoller)) {
1657         roller = tmpLogRoller;
1658       } else {
1659         // Another thread won starting the roller
1660         Threads.shutdown(tmpLogRoller.getThread());
1661         roller = metawalRoller.get();
1662       }
1663     }
1664     return roller;
1665   }
1666 
1667   public MetricsRegionServer getRegionServerMetrics() {
1668     return this.metricsRegionServer;
1669   }
1670 
1671   /**
1672    * @return Master address tracker instance.
1673    */
1674   public MasterAddressTracker getMasterAddressTracker() {
1675     return this.masterAddressTracker;
1676   }
1677 
1678   /*
1679    * Start maintenance Threads, Server, Worker and lease checker threads.
1680    * Install an UncaughtExceptionHandler that calls abort of RegionServer if we
1681    * get an unhandled exception. We cannot set the handler on all threads.
1682    * Server's internal Listener thread is off limits. For Server, if an OOME, it
1683    * waits a while then retries. Meantime, a flush or a compaction that tries to
1684    * run should trigger same critical condition and the shutdown will run. On
1685    * its way out, this server will shut down Server. Leases are sort of
1686    * inbetween. It has an internal thread that while it inherits from Chore, it
1687    * keeps its own internal stop mechanism so needs to be stopped by this
1688    * hosting server. Worker logs the exception and exits.
1689    */
1690   private void startServiceThreads() throws IOException {
1691     // Start executor services
1692     this.service.startExecutorService(ExecutorType.RS_OPEN_REGION,
1693       conf.getInt("hbase.regionserver.executor.openregion.threads", 3));
1694     this.service.startExecutorService(ExecutorType.RS_OPEN_META,
1695       conf.getInt("hbase.regionserver.executor.openmeta.threads", 1));
1696     this.service.startExecutorService(ExecutorType.RS_CLOSE_REGION,
1697       conf.getInt("hbase.regionserver.executor.closeregion.threads", 3));
1698     this.service.startExecutorService(ExecutorType.RS_CLOSE_META,
1699       conf.getInt("hbase.regionserver.executor.closemeta.threads", 1));
1700     if (conf.getBoolean(StoreScanner.STORESCANNER_PARALLEL_SEEK_ENABLE, false)) {
1701       this.service.startExecutorService(ExecutorType.RS_PARALLEL_SEEK,
1702         conf.getInt("hbase.storescanner.parallel.seek.threads", 10));
1703     }
1704     this.service.startExecutorService(ExecutorType.RS_LOG_REPLAY_OPS, conf.getInt(
1705        "hbase.regionserver.wal.max.splitters", SplitLogWorkerCoordination.DEFAULT_MAX_SPLITTERS));
1706 
1707     if (ServerRegionReplicaUtil.isRegionReplicaWaitForPrimaryFlushEnabled(conf)) {
1708       this.service.startExecutorService(ExecutorType.RS_REGION_REPLICA_FLUSH_OPS,
1709         conf.getInt("hbase.regionserver.region.replica.flusher.threads",
1710           conf.getInt("hbase.regionserver.executor.openregion.threads", 3)));
1711     }
1712 
1713     Threads.setDaemonThreadRunning(this.walRoller.getThread(), getName() + ".logRoller",
1714         uncaughtExceptionHandler);
1715     this.cacheFlusher.start(uncaughtExceptionHandler);
1716 
1717     if (this.compactionChecker != null) choreService.scheduleChore(compactionChecker);
1718     if (this.periodicFlusher != null) choreService.scheduleChore(periodicFlusher);
1719     if (this.healthCheckChore != null) choreService.scheduleChore(healthCheckChore);
1720     if (this.nonceManagerChore != null) choreService.scheduleChore(nonceManagerChore);
1721     if (this.storefileRefresher != null) choreService.scheduleChore(storefileRefresher);
1722     if (this.movedRegionsCleaner != null) choreService.scheduleChore(movedRegionsCleaner);
1723 
1724     // Leases is not a Thread. Internally it runs a daemon thread. If it gets
1725     // an unhandled exception, it will just exit.
1726     Threads.setDaemonThreadRunning(this.leases.getThread(), getName() + ".leaseChecker",
1727       uncaughtExceptionHandler);
1728 
1729     if (this.replicationSourceHandler == this.replicationSinkHandler &&
1730         this.replicationSourceHandler != null) {
1731       this.replicationSourceHandler.startReplicationService();
1732     } else {
1733       if (this.replicationSourceHandler != null) {
1734         this.replicationSourceHandler.startReplicationService();
1735       }
1736       if (this.replicationSinkHandler != null) {
1737         this.replicationSinkHandler.startReplicationService();
1738       }
1739     }
1740 
1741     // Create the log splitting worker and start it
1742     // set a smaller retries to fast fail otherwise splitlogworker could be blocked for
1743     // quite a while inside HConnection layer. The worker won't be available for other
1744     // tasks even after current task is preempted after a split task times out.
1745     Configuration sinkConf = HBaseConfiguration.create(conf);
1746     sinkConf.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,
1747       conf.getInt("hbase.log.replay.retries.number", 8)); // 8 retries take about 23 seconds
1748     sinkConf.setInt(HConstants.HBASE_RPC_TIMEOUT_KEY,
1749       conf.getInt("hbase.log.replay.rpc.timeout", 30000)); // default 30 seconds
1750     sinkConf.setInt("hbase.client.serverside.retries.multiplier", 1);
1751     this.splitLogWorker = new SplitLogWorker(this, sinkConf, this, this, walFactory);
1752     splitLogWorker.start();
1753   }
1754 
1755   /**
1756    * Puts up the webui.
1757    * @return Returns final port -- maybe different from what we started with.
1758    * @throws IOException
1759    */
1760   private int putUpWebUI() throws IOException {
1761     int port = this.conf.getInt(HConstants.REGIONSERVER_INFO_PORT,
1762       HConstants.DEFAULT_REGIONSERVER_INFOPORT);
1763     String addr = this.conf.get("hbase.regionserver.info.bindAddress", "0.0.0.0");
1764 
1765     if(this instanceof HMaster) {
1766       port = conf.getInt(HConstants.MASTER_INFO_PORT,
1767           HConstants.DEFAULT_MASTER_INFOPORT);
1768       addr = this.conf.get("hbase.master.info.bindAddress", "0.0.0.0");
1769     }
1770     // -1 is for disabling info server
1771     if (port < 0) return port;
1772 
1773     if (!Addressing.isLocalAddress(InetAddress.getByName(addr))) {
1774       String msg =
1775           "Failed to start http info server. Address " + addr
1776               + " does not belong to this host. Correct configuration parameter: "
1777               + "hbase.regionserver.info.bindAddress";
1778       LOG.error(msg);
1779       throw new IOException(msg);
1780     }
1781     // check if auto port bind enabled
1782     boolean auto = this.conf.getBoolean(HConstants.REGIONSERVER_INFO_PORT_AUTO,
1783         false);
1784     while (true) {
1785       try {
1786         this.infoServer = new InfoServer(getProcessName(), addr, port, false, this.conf);
1787         infoServer.addServlet("dump", "/dump", getDumpServlet());
1788         configureInfoServer();
1789         this.infoServer.start();
1790         break;
1791       } catch (BindException e) {
1792         if (!auto) {
1793           // auto bind disabled throw BindException
1794           LOG.error("Failed binding http info server to port: " + port);
1795           throw e;
1796         }
1797         // auto bind enabled, try to use another port
1798         LOG.info("Failed binding http info server to port: " + port);
1799         port++;
1800       }
1801     }
1802     port = this.infoServer.getPort();
1803     conf.setInt(HConstants.REGIONSERVER_INFO_PORT, port);
1804     int masterInfoPort = conf.getInt(HConstants.MASTER_INFO_PORT,
1805       HConstants.DEFAULT_MASTER_INFOPORT);
1806     conf.setInt("hbase.master.info.port.orig", masterInfoPort);
1807     conf.setInt(HConstants.MASTER_INFO_PORT, port);
1808     return port;
1809   }
1810 
1811   /*
1812    * Verify that server is healthy
1813    */
1814   private boolean isHealthy() {
1815     if (!fsOk) {
1816       // File system problem
1817       return false;
1818     }
1819     // Verify that all threads are alive
1820     if (!(leases.isAlive()
1821         && cacheFlusher.isAlive() && walRoller.isAlive()
1822         && this.compactionChecker.isScheduled()
1823         && this.periodicFlusher.isScheduled())) {
1824       stop("One or more threads are no longer alive -- stop");
1825       return false;
1826     }
1827     final LogRoller metawalRoller = this.metawalRoller.get();
1828     if (metawalRoller != null && !metawalRoller.isAlive()) {
1829       stop("Meta WAL roller thread is no longer alive -- stop");
1830       return false;
1831     }
1832     return true;
1833   }
1834 
1835   private static final byte[] UNSPECIFIED_REGION = new byte[]{};
1836 
1837   @Override
1838   public WAL getWAL(HRegionInfo regionInfo) throws IOException {
1839     WAL wal;
1840     LogRoller roller = walRoller;
1841     //_ROOT_ and hbase:meta regions have separate WAL.
1842     if (regionInfo != null && regionInfo.isMetaTable() &&
1843         regionInfo.getReplicaId() == HRegionInfo.DEFAULT_REPLICA_ID) {
1844       roller = ensureMetaWALRoller();
1845       wal = walFactory.getMetaWAL(regionInfo.getEncodedNameAsBytes());
1846     } else if (regionInfo == null) {
1847       wal = walFactory.getWAL(UNSPECIFIED_REGION);
1848     } else {
1849       wal = walFactory.getWAL(regionInfo.getEncodedNameAsBytes());
1850     }
1851     roller.addWAL(wal);
1852     return wal;
1853   }
1854 
1855   @Override
1856   public ClusterConnection getConnection() {
1857     return this.clusterConnection;
1858   }
1859 
1860   @Override
1861   public MetaTableLocator getMetaTableLocator() {
1862     return this.metaTableLocator;
1863   }
1864 
1865   @Override
1866   public void stop(final String msg) {
1867     if (!this.stopped) {
1868       try {
1869         if (this.rsHost != null) {
1870           this.rsHost.preStop(msg);
1871         }
1872         this.stopped = true;
1873         LOG.info("STOPPED: " + msg);
1874         // Wakes run() if it is sleeping
1875         sleeper.skipSleepCycle();
1876       } catch (IOException exp) {
1877         LOG.warn("The region server did not stop", exp);
1878       }
1879     }
1880   }
1881 
1882   public void waitForServerOnline(){
1883     while (!isStopped() && !isOnline()) {
1884       synchronized (online) {
1885         try {
1886           online.wait(msgInterval);
1887         } catch (InterruptedException ie) {
1888           Thread.currentThread().interrupt();
1889           break;
1890         }
1891       }
1892     }
1893   }
1894 
1895   @Override
1896   public void postOpenDeployTasks(final Region r) throws KeeperException, IOException {
1897     postOpenDeployTasks(new PostOpenDeployContext(r, -1));
1898   }
1899 
1900   @Override
1901   public void postOpenDeployTasks(final PostOpenDeployContext context)
1902       throws KeeperException, IOException {
1903     Region r = context.getRegion();
1904     long masterSystemTime = context.getMasterSystemTime();
1905     Preconditions.checkArgument(r instanceof HRegion, "r must be an HRegion");
1906     rpcServices.checkOpen();
1907     LOG.info("Post open deploy tasks for " + r.getRegionInfo().getRegionNameAsString());
1908     // Do checks to see if we need to compact (references or too many files)
1909     for (Store s : r.getStores()) {
1910       if (s.hasReferences() || s.needsCompaction()) {
1911        this.compactSplitThread.requestSystemCompaction(r, s, "Opening Region");
1912       }
1913     }
1914     long openSeqNum = r.getOpenSeqNum();
1915     if (openSeqNum == HConstants.NO_SEQNUM) {
1916       // If we opened a region, we should have read some sequence number from it.
1917       LOG.error("No sequence number found when opening " +
1918         r.getRegionInfo().getRegionNameAsString());
1919       openSeqNum = 0;
1920     }
1921 
1922     // Update flushed sequence id of a recovering region in ZK
1923     updateRecoveringRegionLastFlushedSequenceId(r);
1924 
1925     // Notify master
1926     if (!reportRegionStateTransition(new RegionStateTransitionContext(
1927         TransitionCode.OPENED, openSeqNum, masterSystemTime, r.getRegionInfo()))) {
1928       throw new IOException("Failed to report opened region to master: "
1929         + r.getRegionInfo().getRegionNameAsString());
1930     }
1931 
1932     triggerFlushInPrimaryRegion((HRegion)r);
1933 
1934     LOG.debug("Finished post open deploy task for " + r.getRegionInfo().getRegionNameAsString());
1935   }
1936 
1937   @Override
1938   public boolean reportRegionStateTransition(TransitionCode code, HRegionInfo... hris) {
1939     return reportRegionStateTransition(code, HConstants.NO_SEQNUM, hris);
1940   }
1941 
1942   @Override
1943   public boolean reportRegionStateTransition(
1944       TransitionCode code, long openSeqNum, HRegionInfo... hris) {
1945     return reportRegionStateTransition(
1946       new RegionStateTransitionContext(code, HConstants.NO_SEQNUM, -1, hris));
1947   }
1948 
1949   @Override
1950   public boolean reportRegionStateTransition(final RegionStateTransitionContext context) {
1951     TransitionCode code = context.getCode();
1952     long openSeqNum = context.getOpenSeqNum();
1953     long masterSystemTime = context.getMasterSystemTime();
1954     HRegionInfo[] hris = context.getHris();
1955 
1956     if (TEST_SKIP_REPORTING_TRANSITION) {
1957       // This is for testing only in case there is no master
1958       // to handle the region transition report at all.
1959       if (code == TransitionCode.OPENED) {
1960         Preconditions.checkArgument(hris != null && hris.length == 1);
1961         if (hris[0].isMetaRegion()) {
1962           try {
1963             MetaTableLocator.setMetaLocation(getZooKeeper(), serverName,
1964                 hris[0].getReplicaId(),State.OPEN);
1965           } catch (KeeperException e) {
1966             LOG.info("Failed to update meta location", e);
1967             return false;
1968           }
1969         } else {
1970           try {
1971             MetaTableAccessor.updateRegionLocation(clusterConnection,
1972               hris[0], serverName, openSeqNum, masterSystemTime);
1973           } catch (IOException e) {
1974             LOG.info("Failed to update meta", e);
1975             return false;
1976           }
1977         }
1978       }
1979       return true;
1980     }
1981 
1982     ReportRegionStateTransitionRequest.Builder builder =
1983       ReportRegionStateTransitionRequest.newBuilder();
1984     builder.setServer(ProtobufUtil.toServerName(serverName));
1985     RegionStateTransition.Builder transition = builder.addTransitionBuilder();
1986     transition.setTransitionCode(code);
1987     if (code == TransitionCode.OPENED && openSeqNum >= 0) {
1988       transition.setOpenSeqNum(openSeqNum);
1989     }
1990     for (HRegionInfo hri: hris) {
1991       transition.addRegionInfo(HRegionInfo.convert(hri));
1992     }
1993     ReportRegionStateTransitionRequest request = builder.build();
1994     while (keepLooping()) {
1995       RegionServerStatusService.BlockingInterface rss = rssStub;
1996       try {
1997         if (rss == null) {
1998           createRegionServerStatusStub();
1999           continue;
2000         }
2001         ReportRegionStateTransitionResponse response =
2002           rss.reportRegionStateTransition(null, request);
2003         if (response.hasErrorMessage()) {
2004           LOG.info("Failed to transition " + hris[0]
2005             + " to " + code + ": " + response.getErrorMessage());
2006           return false;
2007         }
2008         return true;
2009       } catch (ServiceException se) {
2010         IOException ioe = ProtobufUtil.getRemoteException(se);
2011         LOG.info("Failed to report region transition, will retry", ioe);
2012         if (rssStub == rss) {
2013           rssStub = null;
2014         }
2015       }
2016     }
2017     return false;
2018   }
2019 
2020   /**
2021    * Trigger a flush in the primary region replica if this region is a secondary replica. Does not
2022    * block this thread. See RegionReplicaFlushHandler for details.
2023    */
2024   void triggerFlushInPrimaryRegion(final HRegion region) {
2025     if (ServerRegionReplicaUtil.isDefaultReplica(region.getRegionInfo())) {
2026       return;
2027     }
2028     if (!ServerRegionReplicaUtil.isRegionReplicaReplicationEnabled(region.conf) ||
2029         !ServerRegionReplicaUtil.isRegionReplicaWaitForPrimaryFlushEnabled(
2030           region.conf)) {
2031       region.setReadsEnabled(true);
2032       return;
2033     }
2034 
2035     region.setReadsEnabled(false); // disable reads before marking the region as opened.
2036     // RegionReplicaFlushHandler might reset this.
2037 
2038     // submit it to be handled by one of the handlers so that we do not block OpenRegionHandler
2039     this.service.submit(
2040       new RegionReplicaFlushHandler(this, clusterConnection,
2041         rpcRetryingCallerFactory, rpcControllerFactory, operationTimeout, region));
2042   }
2043 
2044   @Override
2045   public RpcServerInterface getRpcServer() {
2046     return rpcServices.rpcServer;
2047   }
2048 
2049   @VisibleForTesting
2050   public RSRpcServices getRSRpcServices() {
2051     return rpcServices;
2052   }
2053 
2054   /**
2055    * Cause the server to exit without closing the regions it is serving, the log
2056    * it is using and without notifying the master. Used unit testing and on
2057    * catastrophic events such as HDFS is yanked out from under hbase or we OOME.
2058    *
2059    * @param reason
2060    *          the reason we are aborting
2061    * @param cause
2062    *          the exception that caused the abort, or null
2063    */
2064   @Override
2065   public void abort(String reason, Throwable cause) {
2066     String msg = "ABORTING region server " + this + ": " + reason;
2067     if (cause != null) {
2068       LOG.fatal(msg, cause);
2069     } else {
2070       LOG.fatal(msg);
2071     }
2072     this.abortRequested = true;
2073     // HBASE-4014: show list of coprocessors that were loaded to help debug
2074     // regionserver crashes.Note that we're implicitly using
2075     // java.util.HashSet's toString() method to print the coprocessor names.
2076     LOG.fatal("RegionServer abort: loaded coprocessors are: " +
2077         CoprocessorHost.getLoadedCoprocessors());
2078     // Try and dump metrics if abort -- might give clue as to how fatal came about....
2079     try {
2080       LOG.info("Dump of metrics as JSON on abort: " + JSONBean.dumpRegionServerMetrics());
2081     } catch (MalformedObjectNameException | IOException e) {
2082       LOG.warn("Failed dumping metrics", e);
2083     }
2084 
2085     // Do our best to report our abort to the master, but this may not work
2086     try {
2087       if (cause != null) {
2088         msg += "\nCause:\n" + StringUtils.stringifyException(cause);
2089       }
2090       // Report to the master but only if we have already registered with the master.
2091       if (rssStub != null && this.serverName != null) {
2092         ReportRSFatalErrorRequest.Builder builder =
2093           ReportRSFatalErrorRequest.newBuilder();
2094         ServerName sn =
2095           ServerName.parseVersionedServerName(this.serverName.getVersionedBytes());
2096         builder.setServer(ProtobufUtil.toServerName(sn));
2097         builder.setErrorMessage(msg);
2098         rssStub.reportRSFatalError(null, builder.build());
2099       }
2100     } catch (Throwable t) {
2101       LOG.warn("Unable to report fatal error to master", t);
2102     }
2103     stop(reason);
2104   }
2105 
2106   /**
2107    * @see HRegionServer#abort(String, Throwable)
2108    */
2109   public void abort(String reason) {
2110     abort(reason, null);
2111   }
2112 
2113   @Override
2114   public boolean isAborted() {
2115     return this.abortRequested;
2116   }
2117 
2118   /*
2119    * Simulate a kill -9 of this server. Exits w/o closing regions or cleaninup
2120    * logs but it does close socket in case want to bring up server on old
2121    * hostname+port immediately.
2122    */
2123   protected void kill() {
2124     this.killed = true;
2125     abort("Simulated kill");
2126   }
2127 
2128   /**
2129    * Called on stop/abort before closing the cluster connection and meta locator.
2130    */
2131   protected void sendShutdownInterrupt() {
2132   }
2133 
2134   /**
2135    * Wait on all threads to finish. Presumption is that all closes and stops
2136    * have already been called.
2137    */
2138   protected void stopServiceThreads() {
2139     // clean up the scheduled chores
2140     if (this.choreService != null) choreService.shutdown();
2141     if (this.nonceManagerChore != null) nonceManagerChore.cancel(true);
2142     if (this.compactionChecker != null) compactionChecker.cancel(true);
2143     if (this.periodicFlusher != null) periodicFlusher.cancel(true);
2144     if (this.healthCheckChore != null) healthCheckChore.cancel(true);
2145     if (this.storefileRefresher != null) storefileRefresher.cancel(true);
2146     if (this.movedRegionsCleaner != null) movedRegionsCleaner.cancel(true);
2147 
2148     if (this.cacheFlusher != null) {
2149       this.cacheFlusher.join();
2150     }
2151 
2152     if (this.spanReceiverHost != null) {
2153       this.spanReceiverHost.closeReceivers();
2154     }
2155     if (this.walRoller != null) {
2156       Threads.shutdown(this.walRoller.getThread());
2157     }
2158     final LogRoller metawalRoller = this.metawalRoller.get();
2159     if (metawalRoller != null) {
2160       Threads.shutdown(metawalRoller.getThread());
2161     }
2162     if (this.compactSplitThread != null) {
2163       this.compactSplitThread.join();
2164     }
2165     if (this.service != null) this.service.shutdown();
2166     if (this.replicationSourceHandler != null &&
2167         this.replicationSourceHandler == this.replicationSinkHandler) {
2168       this.replicationSourceHandler.stopReplicationService();
2169     } else {
2170       if (this.replicationSourceHandler != null) {
2171         this.replicationSourceHandler.stopReplicationService();
2172       }
2173       if (this.replicationSinkHandler != null) {
2174         this.replicationSinkHandler.stopReplicationService();
2175       }
2176     }
2177   }
2178 
2179   /**
2180    * @return Return the object that implements the replication
2181    * source service.
2182    */
2183   ReplicationSourceService getReplicationSourceService() {
2184     return replicationSourceHandler;
2185   }
2186 
2187   /**
2188    * @return Return the object that implements the replication
2189    * sink service.
2190    */
2191   ReplicationSinkService getReplicationSinkService() {
2192     return replicationSinkHandler;
2193   }
2194 
2195   /**
2196    * Get the current master from ZooKeeper and open the RPC connection to it.
2197    * To get a fresh connection, the current rssStub must be null.
2198    * Method will block until a master is available. You can break from this
2199    * block by requesting the server stop.
2200    *
2201    * @return master + port, or null if server has been stopped
2202    */
2203   @VisibleForTesting
2204   protected synchronized ServerName createRegionServerStatusStub() {
2205     if (rssStub != null) {
2206       return masterAddressTracker.getMasterAddress();
2207     }
2208     ServerName sn = null;
2209     long previousLogTime = 0;
2210     boolean refresh = false; // for the first time, use cached data
2211     RegionServerStatusService.BlockingInterface intf = null;
2212     boolean interrupted = false;
2213     try {
2214       while (keepLooping()) {
2215         sn = this.masterAddressTracker.getMasterAddress(refresh);
2216         if (sn == null) {
2217           if (!keepLooping()) {
2218             // give up with no connection.
2219             LOG.debug("No master found and cluster is stopped; bailing out");
2220             return null;
2221           }
2222           if (System.currentTimeMillis() > (previousLogTime + 1000)) {
2223             LOG.debug("No master found; retry");
2224             previousLogTime = System.currentTimeMillis();
2225           }
2226           refresh = true; // let's try pull it from ZK directly
2227           if (sleep(200)) {
2228             interrupted = true;
2229           }
2230           continue;
2231         }
2232 
2233         // If we are on the active master, use the shortcut
2234         if (this instanceof HMaster && sn.equals(getServerName())) {
2235           intf = ((HMaster)this).getMasterRpcServices();
2236           break;
2237         }
2238         try {
2239           BlockingRpcChannel channel =
2240             this.rpcClient.createBlockingRpcChannel(sn, userProvider.getCurrent(),
2241               shortOperationTimeout);
2242           intf = RegionServerStatusService.newBlockingStub(channel);
2243           break;
2244         } catch (IOException e) {
2245           if (System.currentTimeMillis() > (previousLogTime + 1000)) {
2246             e = e instanceof RemoteException ?
2247               ((RemoteException)e).unwrapRemoteException() : e;
2248             if (e instanceof ServerNotRunningYetException) {
2249               LOG.info("Master isn't available yet, retrying");
2250             } else {
2251               LOG.warn("Unable to connect to master. Retrying. Error was:", e);
2252             }
2253             previousLogTime = System.currentTimeMillis();
2254           }
2255           if (sleep(200)) {
2256             interrupted = true;
2257           }
2258         }
2259       }
2260     } finally {
2261       if (interrupted) {
2262         Thread.currentThread().interrupt();
2263       }
2264     }
2265     rssStub = intf;
2266     return sn;
2267   }
2268 
2269   /**
2270    * @return True if we should break loop because cluster is going down or
2271    * this server has been stopped or hdfs has gone bad.
2272    */
2273   private boolean keepLooping() {
2274     return !this.stopped && isClusterUp();
2275   }
2276 
2277   /*
2278    * Let the master know we're here Run initialization using parameters passed
2279    * us by the master.
2280    * @return A Map of key/value configurations we got from the Master else
2281    * null if we failed to register.
2282    * @throws IOException
2283    */
2284   private RegionServerStartupResponse reportForDuty() throws IOException {
2285     ServerName masterServerName = createRegionServerStatusStub();
2286     if (masterServerName == null) return null;
2287     RegionServerStartupResponse result = null;
2288     try {
2289       rpcServices.requestCount.set(0);
2290       LOG.info("reportForDuty to master=" + masterServerName + " with port="
2291         + rpcServices.isa.getPort() + ", startcode=" + this.startcode);
2292       long now = EnvironmentEdgeManager.currentTime();
2293       int port = rpcServices.isa.getPort();
2294       RegionServerStartupRequest.Builder request = RegionServerStartupRequest.newBuilder();
2295       if (shouldUseThisHostnameInstead()) {
2296         request.setUseThisHostnameInstead(useThisHostnameInstead);
2297       }
2298       request.setPort(port);
2299       request.setServerStartCode(this.startcode);
2300       request.setServerCurrentTime(now);
2301       result = this.rssStub.regionServerStartup(null, request.build());
2302     } catch (ServiceException se) {
2303       IOException ioe = ProtobufUtil.getRemoteException(se);
2304       if (ioe instanceof ClockOutOfSyncException) {
2305         LOG.fatal("Master rejected startup because clock is out of sync", ioe);
2306         // Re-throw IOE will cause RS to abort
2307         throw ioe;
2308       } else if (ioe instanceof ServerNotRunningYetException) {
2309         LOG.debug("Master is not running yet");
2310       } else {
2311         LOG.warn("error telling master we are up", se);
2312       }
2313       rssStub = null;
2314     }
2315     return result;
2316   }
2317 
2318   @Override
2319   public RegionStoreSequenceIds getLastSequenceId(byte[] encodedRegionName) {
2320     try {
2321       GetLastFlushedSequenceIdRequest req =
2322           RequestConverter.buildGetLastFlushedSequenceIdRequest(encodedRegionName);
2323       RegionServerStatusService.BlockingInterface rss = rssStub;
2324       if (rss == null) { // Try to connect one more time
2325         createRegionServerStatusStub();
2326         rss = rssStub;
2327         if (rss == null) {
2328           // Still no luck, we tried
2329           LOG.warn("Unable to connect to the master to check " + "the last flushed sequence id");
2330           return RegionStoreSequenceIds.newBuilder().setLastFlushedSequenceId(HConstants.NO_SEQNUM)
2331               .build();
2332         }
2333       }
2334       GetLastFlushedSequenceIdResponse resp = rss.getLastFlushedSequenceId(null, req);
2335       return RegionStoreSequenceIds.newBuilder()
2336           .setLastFlushedSequenceId(resp.getLastFlushedSequenceId())
2337           .addAllStoreSequenceId(resp.getStoreLastFlushedSequenceIdList()).build();
2338     } catch (ServiceException e) {
2339       LOG.warn("Unable to connect to the master to check the last flushed sequence id", e);
2340       return RegionStoreSequenceIds.newBuilder().setLastFlushedSequenceId(HConstants.NO_SEQNUM)
2341           .build();
2342     }
2343   }
2344 
2345   /**
2346    * Closes all regions.  Called on our way out.
2347    * Assumes that its not possible for new regions to be added to onlineRegions
2348    * while this method runs.
2349    */
2350   protected void closeAllRegions(final boolean abort) {
2351     closeUserRegions(abort);
2352     closeMetaTableRegions(abort);
2353   }
2354 
2355   /**
2356    * Close meta region if we carry it
2357    * @param abort Whether we're running an abort.
2358    */
2359   void closeMetaTableRegions(final boolean abort) {
2360     Region meta = null;
2361     this.lock.writeLock().lock();
2362     try {
2363       for (Map.Entry<String, Region> e: onlineRegions.entrySet()) {
2364         HRegionInfo hri = e.getValue().getRegionInfo();
2365         if (hri.isMetaRegion()) {
2366           meta = e.getValue();
2367         }
2368         if (meta != null) break;
2369       }
2370     } finally {
2371       this.lock.writeLock().unlock();
2372     }
2373     if (meta != null) closeRegionIgnoreErrors(meta.getRegionInfo(), abort);
2374   }
2375 
2376   /**
2377    * Schedule closes on all user regions.
2378    * Should be safe calling multiple times because it wont' close regions
2379    * that are already closed or that are closing.
2380    * @param abort Whether we're running an abort.
2381    */
2382   void closeUserRegions(final boolean abort) {
2383     this.lock.writeLock().lock();
2384     try {
2385       for (Map.Entry<String, Region> e: this.onlineRegions.entrySet()) {
2386         Region r = e.getValue();
2387         if (!r.getRegionInfo().isMetaTable() && r.isAvailable()) {
2388           // Don't update zk with this close transition; pass false.
2389           closeRegionIgnoreErrors(r.getRegionInfo(), abort);
2390         }
2391       }
2392     } finally {
2393       this.lock.writeLock().unlock();
2394     }
2395   }
2396 
2397   /** @return the info server */
2398   public InfoServer getInfoServer() {
2399     return infoServer;
2400   }
2401 
2402   /**
2403    * @return true if a stop has been requested.
2404    */
2405   @Override
2406   public boolean isStopped() {
2407     return this.stopped;
2408   }
2409 
2410   @Override
2411   public boolean isStopping() {
2412     return this.stopping;
2413   }
2414 
2415   @Override
2416   public Map<String, Region> getRecoveringRegions() {
2417     return this.recoveringRegions;
2418   }
2419 
2420   /**
2421    *
2422    * @return the configuration
2423    */
2424   @Override
2425   public Configuration getConfiguration() {
2426     return conf;
2427   }
2428 
2429   /** @return the write lock for the server */
2430   ReentrantReadWriteLock.WriteLock getWriteLock() {
2431     return lock.writeLock();
2432   }
2433 
2434   public int getNumberOfOnlineRegions() {
2435     return this.onlineRegions.size();
2436   }
2437 
2438   boolean isOnlineRegionsEmpty() {
2439     return this.onlineRegions.isEmpty();
2440   }
2441 
2442   /**
2443    * For tests, web ui and metrics.
2444    * This method will only work if HRegionServer is in the same JVM as client;
2445    * HRegion cannot be serialized to cross an rpc.
2446    */
2447   public Collection<Region> getOnlineRegionsLocalContext() {
2448     Collection<Region> regions = this.onlineRegions.values();
2449     return Collections.unmodifiableCollection(regions);
2450   }
2451 
2452   @Override
2453   public void addToOnlineRegions(Region region) {
2454     this.onlineRegions.put(region.getRegionInfo().getEncodedName(), region);
2455     configurationManager.registerObserver(region);
2456   }
2457 
2458   /**
2459    * @return A new Map of online regions sorted by region size with the first entry being the
2460    * biggest.  If two regions are the same size, then the last one found wins; i.e. this method
2461    * may NOT return all regions.
2462    */
2463   SortedMap<Long, Region> getCopyOfOnlineRegionsSortedBySize() {
2464     // we'll sort the regions in reverse
2465     SortedMap<Long, Region> sortedRegions = new TreeMap<Long, Region>(
2466         new Comparator<Long>() {
2467           @Override
2468           public int compare(Long a, Long b) {
2469             return -1 * a.compareTo(b);
2470           }
2471         });
2472     // Copy over all regions. Regions are sorted by size with biggest first.
2473     for (Region region : this.onlineRegions.values()) {
2474       sortedRegions.put(region.getMemstoreSize(), region);
2475     }
2476     return sortedRegions;
2477   }
2478 
2479   /**
2480    * @return time stamp in millis of when this region server was started
2481    */
2482   public long getStartcode() {
2483     return this.startcode;
2484   }
2485 
2486   /** @return reference to FlushRequester */
2487   @Override
2488   public FlushRequester getFlushRequester() {
2489     return this.cacheFlusher;
2490   }
2491 
2492   /**
2493    * Get the top N most loaded regions this server is serving so we can tell the
2494    * master which regions it can reallocate if we're overloaded. TODO: actually
2495    * calculate which regions are most loaded. (Right now, we're just grabbing
2496    * the first N regions being served regardless of load.)
2497    */
2498   protected HRegionInfo[] getMostLoadedRegions() {
2499     ArrayList<HRegionInfo> regions = new ArrayList<HRegionInfo>();
2500     for (Region r : onlineRegions.values()) {
2501       if (!r.isAvailable()) {
2502         continue;
2503       }
2504       if (regions.size() < numRegionsToReport) {
2505         regions.add(r.getRegionInfo());
2506       } else {
2507         break;
2508       }
2509     }
2510     return regions.toArray(new HRegionInfo[regions.size()]);
2511   }
2512 
2513   @Override
2514   public Leases getLeases() {
2515     return leases;
2516   }
2517 
2518   /**
2519    * @return Return the rootDir.
2520    */
2521   protected Path getRootDir() {
2522     return rootDir;
2523   }
2524 
2525   /**
2526    * @return Return the fs.
2527    */
2528   @Override
2529   public FileSystem getFileSystem() {
2530     return fs;
2531   }
2532 
2533   @Override
2534   public String toString() {
2535     return getServerName().toString();
2536   }
2537 
2538   /**
2539    * Interval at which threads should run
2540    *
2541    * @return the interval
2542    */
2543   public int getThreadWakeFrequency() {
2544     return threadWakeFrequency;
2545   }
2546 
2547   @Override
2548   public ZooKeeperWatcher getZooKeeper() {
2549     return zooKeeper;
2550   }
2551 
2552   @Override
2553   public BaseCoordinatedStateManager getCoordinatedStateManager() {
2554     return csm;
2555   }
2556 
2557   @Override
2558   public ServerName getServerName() {
2559     return serverName;
2560   }
2561 
2562   @Override
2563   public CompactionRequestor getCompactionRequester() {
2564     return this.compactSplitThread;
2565   }
2566 
2567   public RegionServerCoprocessorHost getRegionServerCoprocessorHost(){
2568     return this.rsHost;
2569   }
2570 
2571   @Override
2572   public ConcurrentMap<byte[], Boolean> getRegionsInTransitionInRS() {
2573     return this.regionsInTransitionInRS;
2574   }
2575 
2576   @Override
2577   public ExecutorService getExecutorService() {
2578     return service;
2579   }
2580 
2581   @Override
2582   public ChoreService getChoreService() {
2583     return choreService;
2584   }
2585 
2586   @Override
2587   public RegionServerQuotaManager getRegionServerQuotaManager() {
2588     return rsQuotaManager;
2589   }
2590 
2591   //
2592   // Main program and support routines
2593   //
2594 
2595   /**
2596    * Load the replication service objects, if any
2597    */
2598   static private void createNewReplicationInstance(Configuration conf,
2599     HRegionServer server, FileSystem fs, Path logDir, Path oldLogDir) throws IOException{
2600 
2601     // If replication is not enabled, then return immediately.
2602     if (!conf.getBoolean(HConstants.REPLICATION_ENABLE_KEY,
2603         HConstants.REPLICATION_ENABLE_DEFAULT)) {
2604       return;
2605     }
2606 
2607     // read in the name of the source replication class from the config file.
2608     String sourceClassname = conf.get(HConstants.REPLICATION_SOURCE_SERVICE_CLASSNAME,
2609                                HConstants.REPLICATION_SERVICE_CLASSNAME_DEFAULT);
2610 
2611     // read in the name of the sink replication class from the config file.
2612     String sinkClassname = conf.get(HConstants.REPLICATION_SINK_SERVICE_CLASSNAME,
2613                              HConstants.REPLICATION_SERVICE_CLASSNAME_DEFAULT);
2614 
2615     // If both the sink and the source class names are the same, then instantiate
2616     // only one object.
2617     if (sourceClassname.equals(sinkClassname)) {
2618       server.replicationSourceHandler = (ReplicationSourceService)
2619                                          newReplicationInstance(sourceClassname,
2620                                          conf, server, fs, logDir, oldLogDir);
2621       server.replicationSinkHandler = (ReplicationSinkService)
2622                                          server.replicationSourceHandler;
2623     } else {
2624       server.replicationSourceHandler = (ReplicationSourceService)
2625                                          newReplicationInstance(sourceClassname,
2626                                          conf, server, fs, logDir, oldLogDir);
2627       server.replicationSinkHandler = (ReplicationSinkService)
2628                                          newReplicationInstance(sinkClassname,
2629                                          conf, server, fs, logDir, oldLogDir);
2630     }
2631   }
2632 
2633   static private ReplicationService newReplicationInstance(String classname,
2634     Configuration conf, HRegionServer server, FileSystem fs, Path logDir,
2635     Path oldLogDir) throws IOException{
2636 
2637     Class<?> clazz = null;
2638     try {
2639       ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
2640       clazz = Class.forName(classname, true, classLoader);
2641     } catch (java.lang.ClassNotFoundException nfe) {
2642       throw new IOException("Could not find class for " + classname);
2643     }
2644 
2645     // create an instance of the replication object.
2646     ReplicationService service = (ReplicationService)
2647                               ReflectionUtils.newInstance(clazz, conf);
2648     service.initialize(server, fs, logDir, oldLogDir);
2649     return service;
2650   }
2651 
2652   /**
2653    * Utility for constructing an instance of the passed HRegionServer class.
2654    *
2655    * @param regionServerClass
2656    * @param conf2
2657    * @return HRegionServer instance.
2658    */
2659   public static HRegionServer constructRegionServer(
2660       Class<? extends HRegionServer> regionServerClass,
2661       final Configuration conf2, CoordinatedStateManager cp) {
2662     try {
2663       Constructor<? extends HRegionServer> c = regionServerClass
2664           .getConstructor(Configuration.class, CoordinatedStateManager.class);
2665       return c.newInstance(conf2, cp);
2666     } catch (Exception e) {
2667       throw new RuntimeException("Failed construction of " + "Regionserver: "
2668           + regionServerClass.toString(), e);
2669     }
2670   }
2671 
2672   /**
2673    * @see org.apache.hadoop.hbase.regionserver.HRegionServerCommandLine
2674    */
2675   public static void main(String[] args) throws Exception {
2676     VersionInfo.logVersion();
2677     Configuration conf = HBaseConfiguration.create();
2678     @SuppressWarnings("unchecked")
2679     Class<? extends HRegionServer> regionServerClass = (Class<? extends HRegionServer>) conf
2680         .getClass(HConstants.REGION_SERVER_IMPL, HRegionServer.class);
2681 
2682     new HRegionServerCommandLine(regionServerClass).doMain(args);
2683   }
2684 
2685   /**
2686    * Gets the online regions of the specified table.
2687    * This method looks at the in-memory onlineRegions.  It does not go to <code>hbase:meta</code>.
2688    * Only returns <em>online</em> regions.  If a region on this table has been
2689    * closed during a disable, etc., it will not be included in the returned list.
2690    * So, the returned list may not necessarily be ALL regions in this table, its
2691    * all the ONLINE regions in the table.
2692    * @param tableName
2693    * @return Online regions from <code>tableName</code>
2694    */
2695   @Override
2696   public List<Region> getOnlineRegions(TableName tableName) {
2697      List<Region> tableRegions = new ArrayList<Region>();
2698      synchronized (this.onlineRegions) {
2699        for (Region region: this.onlineRegions.values()) {
2700          HRegionInfo regionInfo = region.getRegionInfo();
2701          if(regionInfo.getTable().equals(tableName)) {
2702            tableRegions.add(region);
2703          }
2704        }
2705      }
2706      return tableRegions;
2707    }
2708 
2709   /**
2710    * Gets the online tables in this RS.
2711    * This method looks at the in-memory onlineRegions.
2712    * @return all the online tables in this RS
2713    */
2714   @Override
2715   public Set<TableName> getOnlineTables() {
2716     Set<TableName> tables = new HashSet<TableName>();
2717     synchronized (this.onlineRegions) {
2718       for (Region region: this.onlineRegions.values()) {
2719         tables.add(region.getTableDesc().getTableName());
2720       }
2721     }
2722     return tables;
2723   }
2724 
2725   // used by org/apache/hbase/tmpl/regionserver/RSStatusTmpl.jamon (HBASE-4070).
2726   public String[] getRegionServerCoprocessors() {
2727     TreeSet<String> coprocessors = new TreeSet<String>();
2728     try {
2729       coprocessors.addAll(getWAL(null).getCoprocessorHost().getCoprocessors());
2730     } catch (IOException exception) {
2731       LOG.warn("Exception attempting to fetch wal coprocessor information for the common wal; " +
2732           "skipping.");
2733       LOG.debug("Exception details for failure to fetch wal coprocessor information.", exception);
2734     }
2735     Collection<Region> regions = getOnlineRegionsLocalContext();
2736     for (Region region: regions) {
2737       coprocessors.addAll(region.getCoprocessorHost().getCoprocessors());
2738       try {
2739         coprocessors.addAll(getWAL(region.getRegionInfo()).getCoprocessorHost().getCoprocessors());
2740       } catch (IOException exception) {
2741         LOG.warn("Exception attempting to fetch wal coprocessor information for region " + region +
2742             "; skipping.");
2743         LOG.debug("Exception details for failure to fetch wal coprocessor information.", exception);
2744       }
2745     }
2746     return coprocessors.toArray(new String[coprocessors.size()]);
2747   }
2748 
2749   /**
2750    * Try to close the region, logs a warning on failure but continues.
2751    * @param region Region to close
2752    */
2753   private void closeRegionIgnoreErrors(HRegionInfo region, final boolean abort) {
2754     try {
2755       if (!closeRegion(region.getEncodedName(), abort, null)) {
2756         LOG.warn("Failed to close " + region.getRegionNameAsString() +
2757             " - ignoring and continuing");
2758       }
2759     } catch (IOException e) {
2760       LOG.warn("Failed to close " + region.getRegionNameAsString() +
2761           " - ignoring and continuing", e);
2762     }
2763   }
2764 
2765   /**
2766    * Close asynchronously a region, can be called from the master or internally by the regionserver
2767    * when stopping. If called from the master, the region will update the znode status.
2768    *
2769    * <p>
2770    * If an opening was in progress, this method will cancel it, but will not start a new close. The
2771    * coprocessors are not called in this case. A NotServingRegionException exception is thrown.
2772    * </p>
2773 
2774    * <p>
2775    *   If a close was in progress, this new request will be ignored, and an exception thrown.
2776    * </p>
2777    *
2778    * @param encodedName Region to close
2779    * @param abort True if we are aborting
2780    * @return True if closed a region.
2781    * @throws NotServingRegionException if the region is not online
2782    */
2783   protected boolean closeRegion(String encodedName, final boolean abort, final ServerName sn)
2784       throws NotServingRegionException {
2785     //Check for permissions to close.
2786     Region actualRegion = this.getFromOnlineRegions(encodedName);
2787     if ((actualRegion != null) && (actualRegion.getCoprocessorHost() != null)) {
2788       try {
2789         actualRegion.getCoprocessorHost().preClose(false);
2790       } catch (IOException exp) {
2791         LOG.warn("Unable to close region: the coprocessor launched an error ", exp);
2792         return false;
2793       }
2794     }
2795 
2796     final Boolean previous = this.regionsInTransitionInRS.putIfAbsent(encodedName.getBytes(),
2797         Boolean.FALSE);
2798 
2799     if (Boolean.TRUE.equals(previous)) {
2800       LOG.info("Received CLOSE for the region:" + encodedName + " , which we are already " +
2801           "trying to OPEN. Cancelling OPENING.");
2802       if (!regionsInTransitionInRS.replace(encodedName.getBytes(), previous, Boolean.FALSE)){
2803         // The replace failed. That should be an exceptional case, but theoretically it can happen.
2804         // We're going to try to do a standard close then.
2805         LOG.warn("The opening for region " + encodedName + " was done before we could cancel it." +
2806             " Doing a standard close now");
2807         return closeRegion(encodedName, abort, sn);
2808       }
2809       // Let's get the region from the online region list again
2810       actualRegion = this.getFromOnlineRegions(encodedName);
2811       if (actualRegion == null) { // If already online, we still need to close it.
2812         LOG.info("The opening previously in progress has been cancelled by a CLOSE request.");
2813         // The master deletes the znode when it receives this exception.
2814         throw new NotServingRegionException("The region " + encodedName +
2815           " was opening but not yet served. Opening is cancelled.");
2816       }
2817     } else if (Boolean.FALSE.equals(previous)) {
2818       LOG.info("Received CLOSE for the region: " + encodedName +
2819         ", which we are already trying to CLOSE, but not completed yet");
2820       return true;
2821     }
2822 
2823     if (actualRegion == null) {
2824       LOG.debug("Received CLOSE for a region which is not online, and we're not opening.");
2825       this.regionsInTransitionInRS.remove(encodedName.getBytes());
2826       // The master deletes the znode when it receives this exception.
2827       throw new NotServingRegionException("The region " + encodedName +
2828           " is not online, and is not opening.");
2829     }
2830 
2831     CloseRegionHandler crh;
2832     final HRegionInfo hri = actualRegion.getRegionInfo();
2833     if (hri.isMetaRegion()) {
2834       crh = new CloseMetaHandler(this, this, hri, abort);
2835     } else {
2836       crh = new CloseRegionHandler(this, this, hri, abort, sn);
2837     }
2838     this.service.submit(crh);
2839     return true;
2840   }
2841 
2842    /**
2843    * @param regionName
2844    * @return HRegion for the passed binary <code>regionName</code> or null if
2845    *         named region is not member of the online regions.
2846    */
2847   public Region getOnlineRegion(final byte[] regionName) {
2848     String encodedRegionName = HRegionInfo.encodeRegionName(regionName);
2849     return this.onlineRegions.get(encodedRegionName);
2850   }
2851 
2852   public InetSocketAddress[] getRegionBlockLocations(final String encodedRegionName) {
2853     return this.regionFavoredNodesMap.get(encodedRegionName);
2854   }
2855 
2856   @Override
2857   public Region getFromOnlineRegions(final String encodedRegionName) {
2858     return this.onlineRegions.get(encodedRegionName);
2859   }
2860 
2861 
2862   @Override
2863   public boolean removeFromOnlineRegions(final Region r, ServerName destination) {
2864     Region toReturn = this.onlineRegions.remove(r.getRegionInfo().getEncodedName());
2865     if (destination != null) {
2866       long closeSeqNum = r.getMaxFlushedSeqId();
2867       if (closeSeqNum == HConstants.NO_SEQNUM) {
2868         // No edits in WAL for this region; get the sequence number when the region was opened.
2869         closeSeqNum = r.getOpenSeqNum();
2870         if (closeSeqNum == HConstants.NO_SEQNUM) closeSeqNum = 0;
2871       }
2872       addToMovedRegions(r.getRegionInfo().getEncodedName(), destination, closeSeqNum);
2873     }
2874     this.regionFavoredNodesMap.remove(r.getRegionInfo().getEncodedName());
2875     return toReturn != null;
2876   }
2877 
2878   /**
2879    * Protected utility method for safely obtaining an HRegion handle.
2880    *
2881    * @param regionName
2882    *          Name of online {@link HRegion} to return
2883    * @return {@link HRegion} for <code>regionName</code>
2884    * @throws NotServingRegionException
2885    */
2886   protected Region getRegion(final byte[] regionName)
2887       throws NotServingRegionException {
2888     String encodedRegionName = HRegionInfo.encodeRegionName(regionName);
2889     return getRegionByEncodedName(regionName, encodedRegionName);
2890   }
2891 
2892   public Region getRegionByEncodedName(String encodedRegionName)
2893       throws NotServingRegionException {
2894     return getRegionByEncodedName(null, encodedRegionName);
2895   }
2896 
2897   protected Region getRegionByEncodedName(byte[] regionName, String encodedRegionName)
2898     throws NotServingRegionException {
2899     Region region = this.onlineRegions.get(encodedRegionName);
2900     if (region == null) {
2901       MovedRegionInfo moveInfo = getMovedRegion(encodedRegionName);
2902       if (moveInfo != null) {
2903         throw new RegionMovedException(moveInfo.getServerName(), moveInfo.getSeqNum());
2904       }
2905       Boolean isOpening = this.regionsInTransitionInRS.get(Bytes.toBytes(encodedRegionName));
2906       String regionNameStr = regionName == null?
2907         encodedRegionName: Bytes.toStringBinary(regionName);
2908       if (isOpening != null && isOpening.booleanValue()) {
2909         throw new RegionOpeningException("Region " + regionNameStr +
2910           " is opening on " + this.serverName);
2911       }
2912       throw new NotServingRegionException("Region " + regionNameStr +
2913         " is not online on " + this.serverName);
2914     }
2915     return region;
2916   }
2917 
2918   /*
2919    * Cleanup after Throwable caught invoking method. Converts <code>t</code> to
2920    * IOE if it isn't already.
2921    *
2922    * @param t Throwable
2923    *
2924    * @param msg Message to log in error. Can be null.
2925    *
2926    * @return Throwable converted to an IOE; methods can only let out IOEs.
2927    */
2928   private Throwable cleanup(final Throwable t, final String msg) {
2929     // Don't log as error if NSRE; NSRE is 'normal' operation.
2930     if (t instanceof NotServingRegionException) {
2931       LOG.debug("NotServingRegionException; " + t.getMessage());
2932       return t;
2933     }
2934     Throwable e = t instanceof RemoteException ? ((RemoteException) t).unwrapRemoteException() : t;
2935     if (msg == null) {
2936       LOG.error("", e);
2937     } else {
2938       LOG.error(msg, e);
2939     }
2940     if (!rpcServices.checkOOME(t)) {
2941       checkFileSystem();
2942     }
2943     return t;
2944   }
2945 
2946   /*
2947    * @param t
2948    *
2949    * @param msg Message to put in new IOE if passed <code>t</code> is not an IOE
2950    *
2951    * @return Make <code>t</code> an IOE if it isn't already.
2952    */
2953   protected IOException convertThrowableToIOE(final Throwable t, final String msg) {
2954     return (t instanceof IOException ? (IOException) t : msg == null
2955         || msg.length() == 0 ? new IOException(t) : new IOException(msg, t));
2956   }
2957 
2958   /**
2959    * Checks to see if the file system is still accessible. If not, sets
2960    * abortRequested and stopRequested
2961    *
2962    * @return false if file system is not available
2963    */
2964   public boolean checkFileSystem() {
2965     if (this.fsOk && this.fs != null) {
2966       try {
2967         FSUtils.checkFileSystemAvailable(this.fs);
2968       } catch (IOException e) {
2969         abort("File System not available", e);
2970         this.fsOk = false;
2971       }
2972     }
2973     return this.fsOk;
2974   }
2975 
2976   @Override
2977   public void updateRegionFavoredNodesMapping(String encodedRegionName,
2978       List<org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.ServerName> favoredNodes) {
2979     InetSocketAddress[] addr = new InetSocketAddress[favoredNodes.size()];
2980     // Refer to the comment on the declaration of regionFavoredNodesMap on why
2981     // it is a map of region name to InetSocketAddress[]
2982     for (int i = 0; i < favoredNodes.size(); i++) {
2983       addr[i] = InetSocketAddress.createUnresolved(favoredNodes.get(i).getHostName(),
2984           favoredNodes.get(i).getPort());
2985     }
2986     regionFavoredNodesMap.put(encodedRegionName, addr);
2987   }
2988 
2989   /**
2990    * Return the favored nodes for a region given its encoded name. Look at the
2991    * comment around {@link #regionFavoredNodesMap} on why it is InetSocketAddress[]
2992    * @param encodedRegionName
2993    * @return array of favored locations
2994    */
2995   @Override
2996   public InetSocketAddress[] getFavoredNodesForRegion(String encodedRegionName) {
2997     return regionFavoredNodesMap.get(encodedRegionName);
2998   }
2999 
3000   @Override
3001   public ServerNonceManager getNonceManager() {
3002     return this.nonceManager;
3003   }
3004 
3005   private static class MovedRegionInfo {
3006     private final ServerName serverName;
3007     private final long seqNum;
3008     private final long ts;
3009 
3010     public MovedRegionInfo(ServerName serverName, long closeSeqNum) {
3011       this.serverName = serverName;
3012       this.seqNum = closeSeqNum;
3013       ts = EnvironmentEdgeManager.currentTime();
3014      }
3015 
3016     public ServerName getServerName() {
3017       return serverName;
3018     }
3019 
3020     public long getSeqNum() {
3021       return seqNum;
3022     }
3023 
3024     public long getMoveTime() {
3025       return ts;
3026     }
3027   }
3028 
3029   // This map will contains all the regions that we closed for a move.
3030   //  We add the time it was moved as we don't want to keep too old information
3031   protected Map<String, MovedRegionInfo> movedRegions =
3032       new ConcurrentHashMap<String, MovedRegionInfo>(3000);
3033 
3034   // We need a timeout. If not there is a risk of giving a wrong information: this would double
3035   //  the number of network calls instead of reducing them.
3036   private static final int TIMEOUT_REGION_MOVED = (2 * 60 * 1000);
3037 
3038   protected void addToMovedRegions(String encodedName, ServerName destination, long closeSeqNum) {
3039     if (ServerName.isSameHostnameAndPort(destination, this.getServerName())) {
3040       LOG.warn("Not adding moved region record: " + encodedName + " to self.");
3041       return;
3042     }
3043     LOG.info("Adding moved region record: "
3044       + encodedName + " to " + destination + " as of " + closeSeqNum);
3045     movedRegions.put(encodedName, new MovedRegionInfo(destination, closeSeqNum));
3046   }
3047 
3048   void removeFromMovedRegions(String encodedName) {
3049     movedRegions.remove(encodedName);
3050   }
3051 
3052   private MovedRegionInfo getMovedRegion(final String encodedRegionName) {
3053     MovedRegionInfo dest = movedRegions.get(encodedRegionName);
3054 
3055     long now = EnvironmentEdgeManager.currentTime();
3056     if (dest != null) {
3057       if (dest.getMoveTime() > (now - TIMEOUT_REGION_MOVED)) {
3058         return dest;
3059       } else {
3060         movedRegions.remove(encodedRegionName);
3061       }
3062     }
3063 
3064     return null;
3065   }
3066 
3067   /**
3068    * Remove the expired entries from the moved regions list.
3069    */
3070   protected void cleanMovedRegions() {
3071     final long cutOff = System.currentTimeMillis() - TIMEOUT_REGION_MOVED;
3072     Iterator<Entry<String, MovedRegionInfo>> it = movedRegions.entrySet().iterator();
3073 
3074     while (it.hasNext()){
3075       Map.Entry<String, MovedRegionInfo> e = it.next();
3076       if (e.getValue().getMoveTime() < cutOff) {
3077         it.remove();
3078       }
3079     }
3080   }
3081 
3082   /*
3083    * Use this to allow tests to override and schedule more frequently.
3084    */
3085 
3086   protected int movedRegionCleanerPeriod() {
3087         return TIMEOUT_REGION_MOVED;
3088   }
3089 
3090   /**
3091    * Creates a Chore thread to clean the moved region cache.
3092    */
3093 
3094   protected final static class MovedRegionsCleaner extends ScheduledChore implements Stoppable {
3095     private HRegionServer regionServer;
3096     Stoppable stoppable;
3097 
3098     private MovedRegionsCleaner(
3099       HRegionServer regionServer, Stoppable stoppable){
3100       super("MovedRegionsCleaner for region " + regionServer, stoppable,
3101           regionServer.movedRegionCleanerPeriod());
3102       this.regionServer = regionServer;
3103       this.stoppable = stoppable;
3104     }
3105 
3106     static MovedRegionsCleaner create(HRegionServer rs){
3107       Stoppable stoppable = new Stoppable() {
3108         private volatile boolean isStopped = false;
3109         @Override public void stop(String why) { isStopped = true;}
3110         @Override public boolean isStopped() {return isStopped;}
3111       };
3112 
3113       return new MovedRegionsCleaner(rs, stoppable);
3114     }
3115 
3116     @Override
3117     protected void chore() {
3118       regionServer.cleanMovedRegions();
3119     }
3120 
3121     @Override
3122     public void stop(String why) {
3123       stoppable.stop(why);
3124     }
3125 
3126     @Override
3127     public boolean isStopped() {
3128       return stoppable.isStopped();
3129     }
3130   }
3131 
3132   private String getMyEphemeralNodePath() {
3133     return ZKUtil.joinZNode(this.zooKeeper.rsZNode, getServerName().toString());
3134   }
3135 
3136   private boolean isHealthCheckerConfigured() {
3137     String healthScriptLocation = this.conf.get(HConstants.HEALTH_SCRIPT_LOC);
3138     return org.apache.commons.lang.StringUtils.isNotBlank(healthScriptLocation);
3139   }
3140 
3141   /**
3142    * @return the underlying {@link CompactSplitThread} for the servers
3143    */
3144   public CompactSplitThread getCompactSplitThread() {
3145     return this.compactSplitThread;
3146   }
3147 
3148   /**
3149    * A helper function to store the last flushed sequence Id with the previous failed RS for a
3150    * recovering region. The Id is used to skip wal edits which are flushed. Since the flushed
3151    * sequence id is only valid for each RS, we associate the Id with corresponding failed RS.
3152    * @throws KeeperException
3153    * @throws IOException
3154    */
3155   private void updateRecoveringRegionLastFlushedSequenceId(Region r) throws KeeperException,
3156       IOException {
3157     if (!r.isRecovering()) {
3158       // return immdiately for non-recovering regions
3159       return;
3160     }
3161 
3162     HRegionInfo regionInfo = r.getRegionInfo();
3163     ZooKeeperWatcher zkw = getZooKeeper();
3164     String previousRSName = this.getLastFailedRSFromZK(regionInfo.getEncodedName());
3165     Map<byte[], Long> maxSeqIdInStores = r.getMaxStoreSeqId();
3166     long minSeqIdForLogReplay = -1;
3167     for (Long storeSeqIdForReplay : maxSeqIdInStores.values()) {
3168       if (minSeqIdForLogReplay == -1 || storeSeqIdForReplay < minSeqIdForLogReplay) {
3169         minSeqIdForLogReplay = storeSeqIdForReplay;
3170       }
3171     }
3172 
3173     try {
3174       long lastRecordedFlushedSequenceId = -1;
3175       String nodePath = ZKUtil.joinZNode(this.zooKeeper.recoveringRegionsZNode,
3176         regionInfo.getEncodedName());
3177       // recovering-region level
3178       byte[] data;
3179       try {
3180         data = ZKUtil.getData(zkw, nodePath);
3181       } catch (InterruptedException e) {
3182         throw new InterruptedIOException();
3183       }
3184       if (data != null) {
3185         lastRecordedFlushedSequenceId = ZKSplitLog.parseLastFlushedSequenceIdFrom(data);
3186       }
3187       if (data == null || lastRecordedFlushedSequenceId < minSeqIdForLogReplay) {
3188         ZKUtil.setData(zkw, nodePath, ZKUtil.positionToByteArray(minSeqIdForLogReplay));
3189       }
3190       if (previousRSName != null) {
3191         // one level deeper for the failed RS
3192         nodePath = ZKUtil.joinZNode(nodePath, previousRSName);
3193         ZKUtil.setData(zkw, nodePath,
3194           ZKUtil.regionSequenceIdsToByteArray(minSeqIdForLogReplay, maxSeqIdInStores));
3195         LOG.debug("Update last flushed sequence id of region " + regionInfo.getEncodedName() +
3196           " for " + previousRSName);
3197       } else {
3198         LOG.warn("Can't find failed region server for recovering region " +
3199             regionInfo.getEncodedName());
3200       }
3201     } catch (NoNodeException ignore) {
3202       LOG.debug("Region " + regionInfo.getEncodedName() +
3203         " must have completed recovery because its recovery znode has been removed", ignore);
3204     }
3205   }
3206 
3207   /**
3208    * Return the last failed RS name under /hbase/recovering-regions/encodedRegionName
3209    * @param encodedRegionName
3210    * @throws KeeperException
3211    */
3212   private String getLastFailedRSFromZK(String encodedRegionName) throws KeeperException {
3213     String result = null;
3214     long maxZxid = 0;
3215     ZooKeeperWatcher zkw = this.getZooKeeper();
3216     String nodePath = ZKUtil.joinZNode(zkw.recoveringRegionsZNode, encodedRegionName);
3217     List<String> failedServers = ZKUtil.listChildrenNoWatch(zkw, nodePath);
3218     if (failedServers == null || failedServers.isEmpty()) {
3219       return result;
3220     }
3221     for (String failedServer : failedServers) {
3222       String rsPath = ZKUtil.joinZNode(nodePath, failedServer);
3223       Stat stat = new Stat();
3224       ZKUtil.getDataNoWatch(zkw, rsPath, stat);
3225       if (maxZxid < stat.getCzxid()) {
3226         maxZxid = stat.getCzxid();
3227         result = failedServer;
3228       }
3229     }
3230     return result;
3231   }
3232 
3233   public CoprocessorServiceResponse execRegionServerService(
3234       @SuppressWarnings("UnusedParameters") final RpcController controller,
3235       final CoprocessorServiceRequest serviceRequest) throws ServiceException {
3236     try {
3237       ServerRpcController serviceController = new ServerRpcController();
3238       CoprocessorServiceCall call = serviceRequest.getCall();
3239       String serviceName = call.getServiceName();
3240       String methodName = call.getMethodName();
3241       if (!coprocessorServiceHandlers.containsKey(serviceName)) {
3242         throw new UnknownProtocolException(null,
3243             "No registered coprocessor service found for name " + serviceName);
3244       }
3245       Service service = coprocessorServiceHandlers.get(serviceName);
3246       Descriptors.ServiceDescriptor serviceDesc = service.getDescriptorForType();
3247       Descriptors.MethodDescriptor methodDesc = serviceDesc.findMethodByName(methodName);
3248       if (methodDesc == null) {
3249         throw new UnknownProtocolException(service.getClass(), "Unknown method " + methodName
3250             + " called on service " + serviceName);
3251       }
3252       Message request =
3253           service.getRequestPrototype(methodDesc).newBuilderForType().mergeFrom(call.getRequest())
3254               .build();
3255       final Message.Builder responseBuilder =
3256           service.getResponsePrototype(methodDesc).newBuilderForType();
3257       service.callMethod(methodDesc, serviceController, request, new RpcCallback<Message>() {
3258         @Override
3259         public void run(Message message) {
3260           if (message != null) {
3261             responseBuilder.mergeFrom(message);
3262           }
3263         }
3264       });
3265       IOException exception = ResponseConverter.getControllerException(serviceController);
3266       if (exception != null) {
3267         throw exception;
3268       }
3269       Message execResult = responseBuilder.build();
3270       ClientProtos.CoprocessorServiceResponse.Builder builder =
3271           ClientProtos.CoprocessorServiceResponse.newBuilder();
3272       builder.setRegion(RequestConverter.buildRegionSpecifier(RegionSpecifierType.REGION_NAME,
3273         HConstants.EMPTY_BYTE_ARRAY));
3274       builder.setValue(builder.getValueBuilder().setName(execResult.getClass().getName())
3275           .setValue(execResult.toByteString()));
3276       return builder.build();
3277     } catch (IOException ie) {
3278       throw new ServiceException(ie);
3279     }
3280   }
3281 
3282   /**
3283    * @return The cache config instance used by the regionserver.
3284    */
3285   public CacheConfig getCacheConfig() {
3286     return this.cacheConfig;
3287   }
3288 
3289   /**
3290    * @return : Returns the ConfigurationManager object for testing purposes.
3291    */
3292   protected ConfigurationManager getConfigurationManager() {
3293     return configurationManager;
3294   }
3295 
3296   /**
3297    * @return Return table descriptors implementation.
3298    */
3299   public TableDescriptors getTableDescriptors() {
3300     return this.tableDescriptors;
3301   }
3302 
3303   /**
3304    * Reload the configuration from disk.
3305    */
3306   public void updateConfiguration() {
3307     LOG.info("Reloading the configuration from disk.");
3308     // Reload the configuration from disk.
3309     conf.reloadConfiguration();
3310     configurationManager.notifyAllObservers(conf);
3311   }
3312 
3313   @Override
3314   public HeapMemoryManager getHeapMemoryManager() {
3315     return hMemManager;
3316   }
3317 
3318   @Override
3319   public double getCompactionPressure() {
3320     double max = 0;
3321     for (Region region : onlineRegions.values()) {
3322       for (Store store : region.getStores()) {
3323         double normCount = store.getCompactionPressure();
3324         if (normCount > max) {
3325           max = normCount;
3326         }
3327       }
3328     }
3329     return max;
3330   }
3331 }