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