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