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