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