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