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