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