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