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