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