1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.master;
20
21 import java.io.IOException;
22 import java.io.InterruptedIOException;
23 import java.lang.reflect.Constructor;
24 import java.lang.reflect.InvocationTargetException;
25 import java.net.InetAddress;
26 import java.net.InetSocketAddress;
27 import java.net.UnknownHostException;
28 import java.util.ArrayList;
29 import java.util.Collection;
30 import java.util.Collections;
31 import java.util.Comparator;
32 import java.util.HashSet;
33 import java.util.Iterator;
34 import java.util.List;
35 import java.util.Map;
36 import java.util.Set;
37 import java.util.concurrent.TimeUnit;
38 import java.util.concurrent.atomic.AtomicInteger;
39 import java.util.concurrent.atomic.AtomicReference;
40 import java.util.regex.Pattern;
41
42 import javax.servlet.ServletException;
43 import javax.servlet.http.HttpServlet;
44 import javax.servlet.http.HttpServletRequest;
45 import javax.servlet.http.HttpServletResponse;
46
47 import org.apache.commons.logging.Log;
48 import org.apache.commons.logging.LogFactory;
49 import org.apache.hadoop.conf.Configuration;
50 import org.apache.hadoop.fs.Path;
51 import org.apache.hadoop.hbase.ClusterStatus;
52 import org.apache.hadoop.hbase.CoordinatedStateException;
53 import org.apache.hadoop.hbase.CoordinatedStateManager;
54 import org.apache.hadoop.hbase.DoNotRetryIOException;
55 import org.apache.hadoop.hbase.HBaseIOException;
56 import org.apache.hadoop.hbase.HBaseInterfaceAudience;
57 import org.apache.hadoop.hbase.HColumnDescriptor;
58 import org.apache.hadoop.hbase.HConstants;
59 import org.apache.hadoop.hbase.HRegionInfo;
60 import org.apache.hadoop.hbase.HTableDescriptor;
61 import org.apache.hadoop.hbase.MasterNotRunningException;
62 import org.apache.hadoop.hbase.MetaMigrationConvertingToPB;
63 import org.apache.hadoop.hbase.MetaTableAccessor;
64 import org.apache.hadoop.hbase.NamespaceDescriptor;
65 import org.apache.hadoop.hbase.NamespaceNotFoundException;
66 import org.apache.hadoop.hbase.PleaseHoldException;
67 import org.apache.hadoop.hbase.ProcedureInfo;
68 import org.apache.hadoop.hbase.RegionStateListener;
69 import org.apache.hadoop.hbase.Server;
70 import org.apache.hadoop.hbase.ServerLoad;
71 import org.apache.hadoop.hbase.ServerName;
72 import org.apache.hadoop.hbase.TableDescriptors;
73 import org.apache.hadoop.hbase.TableName;
74 import org.apache.hadoop.hbase.TableNotDisabledException;
75 import org.apache.hadoop.hbase.TableNotFoundException;
76 import org.apache.hadoop.hbase.UnknownRegionException;
77 import org.apache.hadoop.hbase.classification.InterfaceAudience;
78 import org.apache.hadoop.hbase.client.MetaScanner;
79 import org.apache.hadoop.hbase.client.MetaScanner.MetaScannerVisitor;
80 import org.apache.hadoop.hbase.client.MetaScanner.MetaScannerVisitorBase;
81 import org.apache.hadoop.hbase.client.RegionReplicaUtil;
82 import org.apache.hadoop.hbase.client.Result;
83 import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
84 import org.apache.hadoop.hbase.exceptions.DeserializationException;
85 import org.apache.hadoop.hbase.executor.ExecutorType;
86 import org.apache.hadoop.hbase.ipc.RpcServer;
87 import org.apache.hadoop.hbase.ipc.ServerNotRunningYetException;
88 import org.apache.hadoop.hbase.master.MasterRpcServices.BalanceSwitchMode;
89 import org.apache.hadoop.hbase.master.RegionState.State;
90 import org.apache.hadoop.hbase.master.balancer.BalancerChore;
91 import org.apache.hadoop.hbase.master.balancer.BaseLoadBalancer;
92 import org.apache.hadoop.hbase.master.balancer.ClusterStatusChore;
93 import org.apache.hadoop.hbase.master.balancer.LoadBalancerFactory;
94 import org.apache.hadoop.hbase.master.cleaner.HFileCleaner;
95 import org.apache.hadoop.hbase.master.cleaner.LogCleaner;
96 import org.apache.hadoop.hbase.master.handler.DispatchMergingRegionHandler;
97 import org.apache.hadoop.hbase.master.normalizer.RegionNormalizer;
98 import org.apache.hadoop.hbase.master.normalizer.RegionNormalizerChore;
99 import org.apache.hadoop.hbase.master.normalizer.RegionNormalizerFactory;
100 import org.apache.hadoop.hbase.master.procedure.AddColumnFamilyProcedure;
101 import org.apache.hadoop.hbase.master.procedure.CreateTableProcedure;
102 import org.apache.hadoop.hbase.master.procedure.DeleteColumnFamilyProcedure;
103 import org.apache.hadoop.hbase.master.procedure.DeleteTableProcedure;
104 import org.apache.hadoop.hbase.master.procedure.DisableTableProcedure;
105 import org.apache.hadoop.hbase.master.procedure.EnableTableProcedure;
106 import org.apache.hadoop.hbase.master.procedure.MasterProcedureConstants;
107 import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv;
108 import org.apache.hadoop.hbase.master.procedure.MasterProcedureScheduler.ProcedureEvent;
109 import org.apache.hadoop.hbase.master.procedure.ModifyColumnFamilyProcedure;
110 import org.apache.hadoop.hbase.master.procedure.ModifyTableProcedure;
111 import org.apache.hadoop.hbase.master.procedure.ProcedurePrepareLatch;
112 import org.apache.hadoop.hbase.master.procedure.ProcedureSyncWait;
113 import org.apache.hadoop.hbase.master.procedure.TruncateTableProcedure;
114 import org.apache.hadoop.hbase.master.snapshot.SnapshotManager;
115 import org.apache.hadoop.hbase.monitoring.MemoryBoundedLogMessageBuffer;
116 import org.apache.hadoop.hbase.monitoring.MonitoredTask;
117 import org.apache.hadoop.hbase.monitoring.TaskMonitor;
118 import org.apache.hadoop.hbase.master.normalizer.NormalizationPlan;
119 import org.apache.hadoop.hbase.procedure.MasterProcedureManagerHost;
120 import org.apache.hadoop.hbase.procedure.flush.MasterFlushTableProcedureManager;
121 import org.apache.hadoop.hbase.procedure2.ProcedureExecutor;
122 import org.apache.hadoop.hbase.procedure2.store.wal.WALProcedureStore;
123 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.GetRegionInfoResponse.CompactionState;
124 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.RegionServerInfo;
125 import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos;
126 import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos.SplitLogTask.RecoveryMode;
127 import org.apache.hadoop.hbase.quotas.MasterQuotaManager;
128 import org.apache.hadoop.hbase.regionserver.DefaultStoreEngine;
129 import org.apache.hadoop.hbase.regionserver.HRegionServer;
130 import org.apache.hadoop.hbase.regionserver.HStore;
131 import org.apache.hadoop.hbase.regionserver.RSRpcServices;
132 import org.apache.hadoop.hbase.regionserver.RegionCoprocessorHost;
133 import org.apache.hadoop.hbase.regionserver.RegionSplitPolicy;
134 import org.apache.hadoop.hbase.regionserver.compactions.ExploringCompactionPolicy;
135 import org.apache.hadoop.hbase.regionserver.compactions.FIFOCompactionPolicy;
136 import org.apache.hadoop.hbase.replication.regionserver.Replication;
137 import org.apache.hadoop.hbase.security.UserProvider;
138 import org.apache.hadoop.hbase.util.Addressing;
139 import org.apache.hadoop.hbase.util.Bytes;
140 import org.apache.hadoop.hbase.util.CompressionTest;
141 import org.apache.hadoop.hbase.util.ConfigUtil;
142 import org.apache.hadoop.hbase.util.EncryptionTest;
143 import org.apache.hadoop.hbase.util.FSUtils;
144 import org.apache.hadoop.hbase.util.HFileArchiveUtil;
145 import org.apache.hadoop.hbase.util.HasThread;
146 import org.apache.hadoop.hbase.util.ModifyRegionUtils;
147 import org.apache.hadoop.hbase.util.IdLock;
148 import org.apache.hadoop.hbase.util.Pair;
149 import org.apache.hadoop.hbase.util.Threads;
150 import org.apache.hadoop.hbase.util.VersionInfo;
151 import org.apache.hadoop.hbase.zookeeper.DrainingServerTracker;
152 import org.apache.hadoop.hbase.zookeeper.LoadBalancerTracker;
153 import org.apache.hadoop.hbase.zookeeper.MasterAddressTracker;
154 import org.apache.hadoop.hbase.zookeeper.MetaTableLocator;
155 import org.apache.hadoop.hbase.zookeeper.RegionNormalizerTracker;
156 import org.apache.hadoop.hbase.zookeeper.RegionServerTracker;
157 import org.apache.hadoop.hbase.zookeeper.ZKClusterId;
158 import org.apache.hadoop.hbase.zookeeper.ZKUtil;
159 import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
160 import org.apache.zookeeper.KeeperException;
161 import org.mortbay.jetty.Connector;
162 import org.mortbay.jetty.nio.SelectChannelConnector;
163 import org.mortbay.jetty.servlet.Context;
164
165 import com.google.common.annotations.VisibleForTesting;
166 import com.google.common.collect.Maps;
167 import com.google.protobuf.Descriptors;
168 import com.google.protobuf.Service;
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185 @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.TOOLS)
186 @SuppressWarnings("deprecation")
187 public class HMaster extends HRegionServer implements MasterServices, Server {
188 private static final Log LOG = LogFactory.getLog(HMaster.class.getName());
189
190
191
192
193
194 private static class InitializationMonitor extends HasThread {
195
196 public static final String TIMEOUT_KEY = "hbase.master.initializationmonitor.timeout";
197 public static final long TIMEOUT_DEFAULT = TimeUnit.MILLISECONDS.convert(15, TimeUnit.MINUTES);
198
199
200
201
202
203 public static final String HALT_KEY = "hbase.master.initializationmonitor.haltontimeout";
204 public static final boolean HALT_DEFAULT = false;
205
206 private final HMaster master;
207 private final long timeout;
208 private final boolean haltOnTimeout;
209
210
211 InitializationMonitor(HMaster master) {
212 super("MasterInitializationMonitor");
213 this.master = master;
214 this.timeout = master.getConfiguration().getLong(TIMEOUT_KEY, TIMEOUT_DEFAULT);
215 this.haltOnTimeout = master.getConfiguration().getBoolean(HALT_KEY, HALT_DEFAULT);
216 this.setDaemon(true);
217 }
218
219 @Override
220 public void run() {
221 try {
222 while (!master.isStopped() && master.isActiveMaster()) {
223 Thread.sleep(timeout);
224 if (master.isInitialized()) {
225 LOG.debug("Initialization completed within allotted tolerance. Monitor exiting.");
226 } else {
227 LOG.error("Master failed to complete initialization after " + timeout + "ms. Please"
228 + " consider submitting a bug report including a thread dump of this process.");
229 if (haltOnTimeout) {
230 LOG.error("Zombie Master exiting. Thread dump to stdout");
231 Threads.printThreadInfo(System.out, "Zombie HMaster");
232 System.exit(-1);
233 }
234 }
235 }
236 } catch (InterruptedException ie) {
237 LOG.trace("InitMonitor thread interrupted. Existing.");
238 }
239 }
240 }
241
242
243
244 public static final String MASTER = "master";
245
246
247 private final ActiveMasterManager activeMasterManager;
248
249 RegionServerTracker regionServerTracker;
250
251 private DrainingServerTracker drainingServerTracker;
252
253 LoadBalancerTracker loadBalancerTracker;
254
255
256 private RegionNormalizerTracker regionNormalizerTracker;
257
258
259 private TableNamespaceManager tableNamespaceManager;
260
261
262 final MetricsMaster metricsMaster;
263
264 private MasterFileSystem fileSystemManager;
265
266
267 volatile ServerManager serverManager;
268
269
270 AssignmentManager assignmentManager;
271
272
273
274
275 MemoryBoundedLogMessageBuffer rsFatals;
276
277
278 private volatile boolean isActiveMaster = false;
279
280
281
282 private final ProcedureEvent initialized = new ProcedureEvent("master initialized");
283
284
285
286 volatile boolean serviceStarted = false;
287
288
289 private final ProcedureEvent serverCrashProcessingEnabled =
290 new ProcedureEvent("server crash processing");
291
292 LoadBalancer balancer;
293 private RegionNormalizer normalizer;
294 private BalancerChore balancerChore;
295 private RegionNormalizerChore normalizerChore;
296 private ClusterStatusChore clusterStatusChore;
297 private ClusterStatusPublisher clusterStatusPublisherChore = null;
298
299 CatalogJanitor catalogJanitorChore;
300 private LogCleaner logCleaner;
301 private HFileCleaner hfileCleaner;
302 private ExpiredMobFileCleanerChore expiredMobFileCleanerChore;
303 private MobFileCompactionChore mobFileCompactChore;
304 MasterMobFileCompactionThread mobFileCompactThread;
305
306 private final IdLock mobFileCompactionLock = new IdLock();
307
308
309 private Map<TableName, AtomicInteger> mobFileCompactionStates = Maps.newConcurrentMap();
310
311 MasterCoprocessorHost cpHost;
312
313 private final boolean preLoadTableDescriptors;
314
315
316 private long masterActiveTime;
317
318
319 private final boolean masterCheckCompression;
320
321
322 private final boolean masterCheckEncryption;
323
324 Map<String, Service> coprocessorServiceHandlers = Maps.newHashMap();
325
326
327 SnapshotManager snapshotManager;
328
329 MasterProcedureManagerHost mpmHost;
330
331
332 private volatile MasterQuotaManager quotaManager;
333
334 private ProcedureExecutor<MasterProcedureEnv> procedureExecutor;
335 private WALProcedureStore procedureStore;
336
337
338 private volatile boolean initializationBeforeMetaAssignment = false;
339
340
341 private org.mortbay.jetty.Server masterJettyServer;
342
343 public static class RedirectServlet extends HttpServlet {
344 private static final long serialVersionUID = 2894774810058302472L;
345 private static int regionServerInfoPort;
346
347 @Override
348 public void doGet(HttpServletRequest request,
349 HttpServletResponse response) throws ServletException, IOException {
350 String redirectUrl = request.getScheme() + "://"
351 + request.getServerName() + ":" + regionServerInfoPort
352 + request.getRequestURI();
353 response.sendRedirect(redirectUrl);
354 }
355 }
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373 public HMaster(final Configuration conf, CoordinatedStateManager csm)
374 throws IOException, KeeperException, InterruptedException {
375 super(conf, csm);
376 this.rsFatals = new MemoryBoundedLogMessageBuffer(
377 conf.getLong("hbase.master.buffer.for.rs.fatals", 1*1024*1024));
378
379 LOG.info("hbase.rootdir=" + FSUtils.getRootDir(this.conf) +
380 ", hbase.cluster.distributed=" + this.conf.getBoolean(HConstants.CLUSTER_DISTRIBUTED, false));
381
382
383 this.conf.setBoolean(HConstants.USE_META_REPLICAS, false);
384
385 Replication.decorateMasterConfiguration(this.conf);
386
387
388
389 if (this.conf.get("mapreduce.task.attempt.id") == null) {
390 this.conf.set("mapreduce.task.attempt.id", "hb_m_" + this.serverName.toString());
391 }
392
393
394 this.masterCheckCompression = conf.getBoolean("hbase.master.check.compression", true);
395
396
397 this.masterCheckEncryption = conf.getBoolean("hbase.master.check.encryption", true);
398
399 this.metricsMaster = new MetricsMaster(new MetricsMasterWrapperImpl(this));
400
401
402 this.preLoadTableDescriptors = conf.getBoolean("hbase.master.preload.tabledescriptors", true);
403
404
405
406 boolean shouldPublish = conf.getBoolean(HConstants.STATUS_PUBLISHED,
407 HConstants.STATUS_PUBLISHED_DEFAULT);
408 Class<? extends ClusterStatusPublisher.Publisher> publisherClass =
409 conf.getClass(ClusterStatusPublisher.STATUS_PUBLISHER_CLASS,
410 ClusterStatusPublisher.DEFAULT_STATUS_PUBLISHER_CLASS,
411 ClusterStatusPublisher.Publisher.class);
412
413 if (shouldPublish) {
414 if (publisherClass == null) {
415 LOG.warn(HConstants.STATUS_PUBLISHED + " is true, but " +
416 ClusterStatusPublisher.DEFAULT_STATUS_PUBLISHER_CLASS +
417 " is not set - not publishing status");
418 } else {
419 clusterStatusPublisherChore = new ClusterStatusPublisher(this, conf, publisherClass);
420 getChoreService().scheduleChore(clusterStatusPublisherChore);
421 }
422 }
423
424
425 if (!conf.getBoolean("hbase.testing.nocluster", false)) {
426 activeMasterManager = new ActiveMasterManager(zooKeeper, this.serverName, this);
427 int infoPort = putUpJettyServer();
428 startActiveMasterManager(infoPort);
429 } else {
430 activeMasterManager = null;
431 }
432 }
433
434
435 private int putUpJettyServer() throws IOException {
436 if (!conf.getBoolean("hbase.master.infoserver.redirect", true)) {
437 return -1;
438 }
439 int infoPort = conf.getInt("hbase.master.info.port.orig",
440 HConstants.DEFAULT_MASTER_INFOPORT);
441
442 if (infoPort < 0 || infoServer == null) {
443 return -1;
444 }
445 String addr = conf.get("hbase.master.info.bindAddress", "0.0.0.0");
446 if (!Addressing.isLocalAddress(InetAddress.getByName(addr))) {
447 String msg =
448 "Failed to start redirecting jetty server. Address " + addr
449 + " does not belong to this host. Correct configuration parameter: "
450 + "hbase.master.info.bindAddress";
451 LOG.error(msg);
452 throw new IOException(msg);
453 }
454
455 RedirectServlet.regionServerInfoPort = infoServer.getPort();
456 if(RedirectServlet.regionServerInfoPort == infoPort) {
457 return infoPort;
458 }
459 masterJettyServer = new org.mortbay.jetty.Server();
460 Connector connector = new SelectChannelConnector();
461 connector.setHost(addr);
462 connector.setPort(infoPort);
463 masterJettyServer.addConnector(connector);
464 masterJettyServer.setStopAtShutdown(true);
465 Context context = new Context(masterJettyServer, "/", Context.NO_SESSIONS);
466 context.addServlet(RedirectServlet.class, "/*");
467 try {
468 masterJettyServer.start();
469 } catch (Exception e) {
470 throw new IOException("Failed to start redirecting jetty server", e);
471 }
472 return connector.getLocalPort();
473 }
474
475
476
477
478 @Override
479 protected void login(UserProvider user, String host) throws IOException {
480 try {
481 super.login(user, host);
482 } catch (IOException ie) {
483 user.login("hbase.master.keytab.file",
484 "hbase.master.kerberos.principal", host);
485 }
486 }
487
488
489
490
491
492
493 @Override
494 protected void waitForMasterActive(){
495 boolean tablesOnMaster = BaseLoadBalancer.tablesOnMaster(conf);
496 while (!(tablesOnMaster && isActiveMaster)
497 && !isStopped() && !isAborted()) {
498 sleeper.sleep();
499 }
500 }
501
502 @VisibleForTesting
503 public MasterRpcServices getMasterRpcServices() {
504 return (MasterRpcServices)rpcServices;
505 }
506
507 public boolean balanceSwitch(final boolean b) throws IOException {
508 return getMasterRpcServices().switchBalancer(b, BalanceSwitchMode.ASYNC);
509 }
510
511 @Override
512 protected String getProcessName() {
513 return MASTER;
514 }
515
516 @Override
517 protected boolean canCreateBaseZNode() {
518 return true;
519 }
520
521 @Override
522 protected boolean canUpdateTableDescriptor() {
523 return true;
524 }
525
526 @Override
527 protected RSRpcServices createRpcServices() throws IOException {
528 return new MasterRpcServices(this);
529 }
530
531 @Override
532 protected void configureInfoServer() {
533 infoServer.addServlet("master-status", "/master-status", MasterStatusServlet.class);
534 infoServer.setAttribute(MASTER, this);
535 if (BaseLoadBalancer.tablesOnMaster(conf)) {
536 super.configureInfoServer();
537 }
538 }
539
540 @Override
541 protected Class<? extends HttpServlet> getDumpServlet() {
542 return MasterDumpServlet.class;
543 }
544
545
546
547
548
549 @Override
550 protected void doMetrics() {
551 try {
552 if (assignmentManager != null) {
553 assignmentManager.updateRegionsInTransitionMetrics();
554 }
555 } catch (Throwable e) {
556 LOG.error("Couldn't update metrics: " + e.getMessage());
557 }
558 }
559
560 MetricsMaster getMasterMetrics() {
561 return metricsMaster;
562 }
563
564
565
566
567
568
569
570
571 void initializeZKBasedSystemTrackers() throws IOException,
572 InterruptedException, KeeperException, CoordinatedStateException {
573 this.balancer = LoadBalancerFactory.getLoadBalancer(conf);
574 this.normalizer = RegionNormalizerFactory.getRegionNormalizer(conf);
575 this.normalizer.setMasterServices(this);
576 this.loadBalancerTracker = new LoadBalancerTracker(zooKeeper, this);
577 this.loadBalancerTracker.start();
578 this.regionNormalizerTracker = new RegionNormalizerTracker(zooKeeper, this);
579 this.regionNormalizerTracker.start();
580 this.assignmentManager = new AssignmentManager(this, serverManager,
581 this.balancer, this.service, this.metricsMaster,
582 this.tableLockManager);
583 zooKeeper.registerListenerFirst(assignmentManager);
584
585 this.regionServerTracker = new RegionServerTracker(zooKeeper, this,
586 this.serverManager);
587 this.regionServerTracker.start();
588
589 this.drainingServerTracker = new DrainingServerTracker(zooKeeper, this,
590 this.serverManager);
591 this.drainingServerTracker.start();
592
593
594
595 boolean wasUp = this.clusterStatusTracker.isClusterUp();
596 if (!wasUp) this.clusterStatusTracker.setClusterUp();
597
598 LOG.info("Server active/primary master=" + this.serverName +
599 ", sessionid=0x" +
600 Long.toHexString(this.zooKeeper.getRecoverableZooKeeper().getSessionId()) +
601 ", setting cluster-up flag (Was=" + wasUp + ")");
602
603
604 this.snapshotManager = new SnapshotManager();
605 this.mpmHost = new MasterProcedureManagerHost();
606 this.mpmHost.register(this.snapshotManager);
607 this.mpmHost.register(new MasterFlushTableProcedureManager());
608 this.mpmHost.loadProcedures(conf);
609 this.mpmHost.initialize(this, this.metricsMaster);
610 }
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632 private void finishActiveMasterInitialization(MonitoredTask status)
633 throws IOException, InterruptedException, KeeperException, CoordinatedStateException {
634
635 isActiveMaster = true;
636 Thread zombieDetector = new Thread(new InitializationMonitor(this));
637 zombieDetector.start();
638
639
640
641
642
643
644
645 status.setStatus("Initializing Master file system");
646
647 this.masterActiveTime = System.currentTimeMillis();
648
649 this.fileSystemManager = new MasterFileSystem(this, this);
650
651
652 this.tableDescriptors.setCacheOn();
653
654 this.tableDescriptors.get(TableName.META_TABLE_NAME).setRegionReplication(
655 conf.getInt(HConstants.META_REPLICAS_NUM, HConstants.DEFAULT_META_REPLICA_NUM));
656
657 if (preLoadTableDescriptors) {
658 status.setStatus("Pre-loading table descriptors");
659 this.tableDescriptors.getAll();
660 }
661
662
663 status.setStatus("Publishing Cluster ID in ZooKeeper");
664 ZKClusterId.setClusterId(this.zooKeeper, fileSystemManager.getClusterId());
665 this.serverManager = createServerManager(this, this);
666
667 setupClusterConnection();
668
669
670 this.tableLockManager.reapWriteLocks();
671
672 status.setStatus("Initializing ZK system trackers");
673 initializeZKBasedSystemTrackers();
674
675
676 status.setStatus("Initializing master coprocessors");
677 this.cpHost = new MasterCoprocessorHost(this, this.conf);
678
679
680 status.setStatus("Initializing master service threads");
681 startServiceThreads();
682
683
684 sleeper.skipSleepCycle();
685
686
687 this.serverManager.waitForRegionServers(status);
688
689 for (ServerName sn: this.regionServerTracker.getOnlineServers()) {
690
691 if (!this.serverManager.isServerOnline(sn)
692 && serverManager.checkAndRecordNewServer(sn, ServerLoad.EMPTY_SERVERLOAD)) {
693 LOG.info("Registered server found up in zk but who has not yet reported in: " + sn);
694 }
695 }
696
697
698
699
700 Set<ServerName> previouslyFailedServers =
701 this.fileSystemManager.getFailedServersFromLogFolders();
702
703
704 ServerName oldMetaServerLocation = metaTableLocator.getMetaRegionLocation(this.getZooKeeper());
705 if (oldMetaServerLocation != null && previouslyFailedServers.contains(oldMetaServerLocation)) {
706 splitMetaLogBeforeAssignment(oldMetaServerLocation);
707
708
709 }
710 Set<ServerName> previouslyFailedMetaRSs = getPreviouselyFailedMetaServersFromZK();
711
712
713
714
715
716
717
718
719 previouslyFailedMetaRSs.addAll(previouslyFailedServers);
720
721 this.initializationBeforeMetaAssignment = true;
722
723
724 if (BaseLoadBalancer.tablesOnMaster(conf)) {
725 waitForServerOnline();
726 }
727
728
729 this.balancer.setClusterStatus(getClusterStatus());
730 this.balancer.setMasterServices(this);
731 this.balancer.initialize();
732
733
734
735 if (isStopped()) return;
736
737
738 status.setStatus("Assigning Meta Region");
739 assignMeta(status, previouslyFailedMetaRSs, HRegionInfo.DEFAULT_REPLICA_ID);
740
741
742 if (isStopped()) return;
743
744 status.setStatus("Submitting log splitting work for previously failed region servers");
745
746
747 for (ServerName tmpServer : previouslyFailedServers) {
748 this.serverManager.processDeadServer(tmpServer, true);
749 }
750
751
752
753 if (this.conf.getBoolean("hbase.MetaMigrationConvertingToPB", true)) {
754 MetaMigrationConvertingToPB.updateMetaIfNecessary(this);
755 }
756
757
758 status.setStatus("Starting assignment manager");
759 this.assignmentManager.joinCluster();
760
761
762 this.balancer.setClusterStatus(getClusterStatus());
763
764
765 status.setStatus("Starting balancer and catalog janitor");
766 this.clusterStatusChore = new ClusterStatusChore(this, balancer);
767 getChoreService().scheduleChore(clusterStatusChore);
768 this.balancerChore = new BalancerChore(this);
769 getChoreService().scheduleChore(balancerChore);
770 this.normalizerChore = new RegionNormalizerChore(this);
771 getChoreService().scheduleChore(normalizerChore);
772 this.catalogJanitorChore = new CatalogJanitor(this, this);
773 getChoreService().scheduleChore(catalogJanitorChore);
774
775 status.setStatus("Starting namespace manager");
776 initNamespace();
777
778 if (this.cpHost != null) {
779 try {
780 this.cpHost.preMasterInitialization();
781 } catch (IOException e) {
782 LOG.error("Coprocessor preMasterInitialization() hook failed", e);
783 }
784 }
785
786 status.markComplete("Initialization successful");
787 LOG.info("Master has completed initialization");
788 configurationManager.registerObserver(this.balancer);
789
790
791 setInitialized(true);
792
793 status.setStatus("Starting quota manager");
794 initQuotaManager();
795
796
797 Set<ServerName> EMPTY_SET = new HashSet<ServerName>();
798 int numReplicas = conf.getInt(HConstants.META_REPLICAS_NUM,
799 HConstants.DEFAULT_META_REPLICA_NUM);
800 for (int i = 1; i < numReplicas; i++) {
801 assignMeta(status, EMPTY_SET, i);
802 }
803 unassignExcessMetaReplica(zooKeeper, numReplicas);
804
805
806
807
808 this.serverManager.clearDeadServersWithSameHostNameAndPortOfOnlineServer();
809
810
811 status.setStatus("Checking ZNode ACLs");
812 zooKeeper.checkAndSetZNodeAcls();
813
814 status.setStatus("Calling postStartMaster coprocessors");
815
816 this.expiredMobFileCleanerChore = new ExpiredMobFileCleanerChore(this);
817 getChoreService().scheduleChore(this.expiredMobFileCleanerChore);
818 this.mobFileCompactChore = new MobFileCompactionChore(this);
819 getChoreService().scheduleChore(this.mobFileCompactChore);
820 this.mobFileCompactThread = new MasterMobFileCompactionThread(this);
821
822 if (this.cpHost != null) {
823
824 try {
825 this.cpHost.postStartMaster();
826 } catch (IOException ioe) {
827 LOG.error("Coprocessor postStartMaster() hook failed", ioe);
828 }
829 }
830
831 zombieDetector.interrupt();
832 }
833
834 private void initQuotaManager() throws IOException {
835 quotaManager = new MasterQuotaManager(this);
836 this.assignmentManager.setRegionStateListener((RegionStateListener) quotaManager);
837 quotaManager.start();
838 }
839
840
841
842
843
844
845
846
847
848 ServerManager createServerManager(final Server master,
849 final MasterServices services)
850 throws IOException {
851
852
853 return new ServerManager(master, services);
854 }
855
856 private void unassignExcessMetaReplica(ZooKeeperWatcher zkw, int numMetaReplicasConfigured) {
857
858
859 try {
860 List<String> metaReplicaZnodes = zooKeeper.getMetaReplicaNodes();
861 for (String metaReplicaZnode : metaReplicaZnodes) {
862 int replicaId = zooKeeper.getMetaReplicaIdFromZnode(metaReplicaZnode);
863 if (replicaId >= numMetaReplicasConfigured) {
864 RegionState r = MetaTableLocator.getMetaRegionState(zkw, replicaId);
865 LOG.info("Closing excess replica of meta region " + r.getRegion());
866
867 ServerManager.closeRegionSilentlyAndWait(getConnection(), r.getServerName(),
868 r.getRegion(), 30000);
869 ZKUtil.deleteNode(zkw, zkw.getZNodeForReplica(replicaId));
870 }
871 }
872 } catch (Exception ex) {
873
874
875 LOG.warn("Ignoring exception " + ex);
876 }
877 }
878
879
880
881
882
883
884
885
886
887
888 void assignMeta(MonitoredTask status, Set<ServerName> previouslyFailedMetaRSs, int replicaId)
889 throws InterruptedException, IOException, KeeperException {
890
891 int assigned = 0;
892 long timeout = this.conf.getLong("hbase.catalog.verification.timeout", 1000);
893 if (replicaId == HRegionInfo.DEFAULT_REPLICA_ID) {
894 status.setStatus("Assigning hbase:meta region");
895 } else {
896 status.setStatus("Assigning hbase:meta region, replicaId " + replicaId);
897 }
898
899 RegionStates regionStates = assignmentManager.getRegionStates();
900 RegionState metaState = MetaTableLocator.getMetaRegionState(getZooKeeper(), replicaId);
901 HRegionInfo hri = RegionReplicaUtil.getRegionInfoForReplica(HRegionInfo.FIRST_META_REGIONINFO,
902 replicaId);
903 ServerName currentMetaServer = metaState.getServerName();
904 if (!ConfigUtil.useZKForAssignment(conf)) {
905 regionStates.createRegionState(hri, metaState.getState(),
906 currentMetaServer, null);
907 } else {
908 regionStates.createRegionState(hri);
909 }
910 boolean rit = this.assignmentManager.
911 processRegionInTransitionAndBlockUntilAssigned(hri);
912 boolean metaRegionLocation = metaTableLocator.verifyMetaRegionLocation(
913 this.getConnection(), this.getZooKeeper(), timeout, replicaId);
914 if (!metaRegionLocation || !metaState.isOpened()) {
915
916
917 assigned++;
918 if (!ConfigUtil.useZKForAssignment(conf)) {
919 assignMetaZkLess(regionStates, metaState, timeout, previouslyFailedMetaRSs);
920 } else if (!rit) {
921
922 if (currentMetaServer != null) {
923
924
925
926
927
928
929
930 if (serverManager.isServerOnline(currentMetaServer)) {
931 LOG.info("Forcing expire of " + currentMetaServer);
932 serverManager.expireServer(currentMetaServer);
933 }
934 if (replicaId == HRegionInfo.DEFAULT_REPLICA_ID) {
935 splitMetaLogBeforeAssignment(currentMetaServer);
936 previouslyFailedMetaRSs.add(currentMetaServer);
937 }
938 }
939 assignmentManager.assignMeta(hri);
940 }
941 } else {
942
943 regionStates.updateRegionState(
944 HRegionInfo.FIRST_META_REGIONINFO, State.OPEN, currentMetaServer);
945 this.assignmentManager.regionOnline(
946 HRegionInfo.FIRST_META_REGIONINFO, currentMetaServer);
947 }
948
949 if (replicaId == HRegionInfo.DEFAULT_REPLICA_ID) enableMeta(TableName.META_TABLE_NAME);
950
951 if ((RecoveryMode.LOG_REPLAY == this.getMasterFileSystem().getLogRecoveryMode())
952 && (!previouslyFailedMetaRSs.isEmpty())) {
953
954 status.setStatus("replaying log for Meta Region");
955 this.fileSystemManager.splitMetaLog(previouslyFailedMetaRSs);
956 }
957
958
959
960
961
962 if (replicaId == HRegionInfo.DEFAULT_REPLICA_ID) enableCrashedServerProcessing(assigned != 0);
963 LOG.info("hbase:meta with replicaId " + replicaId + " assigned=" + assigned + ", rit=" + rit +
964 ", location=" + metaTableLocator.getMetaRegionLocation(this.getZooKeeper(), replicaId));
965 status.setStatus("META assigned.");
966 }
967
968 private void assignMetaZkLess(RegionStates regionStates, RegionState regionState, long timeout,
969 Set<ServerName> previouslyFailedRs) throws IOException, KeeperException {
970 ServerName currentServer = regionState.getServerName();
971 if (serverManager.isServerOnline(currentServer)) {
972 LOG.info("Meta was in transition on " + currentServer);
973 assignmentManager.processRegionInTransitionZkLess();
974 } else {
975 if (currentServer != null) {
976 if (regionState.getRegion().getReplicaId() == HRegionInfo.DEFAULT_REPLICA_ID) {
977 splitMetaLogBeforeAssignment(currentServer);
978 regionStates.logSplit(HRegionInfo.FIRST_META_REGIONINFO);
979 previouslyFailedRs.add(currentServer);
980 }
981 }
982 LOG.info("Re-assigning hbase:meta, it was on " + currentServer);
983 regionStates.updateRegionState(regionState.getRegion(), State.OFFLINE);
984 assignmentManager.assignMeta(regionState.getRegion());
985 }
986 }
987
988 void initNamespace() throws IOException {
989
990 tableNamespaceManager = new TableNamespaceManager(this);
991 tableNamespaceManager.start();
992 }
993
994 boolean isCatalogJanitorEnabled() {
995 return catalogJanitorChore != null ?
996 catalogJanitorChore.getEnabled() : false;
997 }
998
999 private void splitMetaLogBeforeAssignment(ServerName currentMetaServer) throws IOException {
1000 if (RecoveryMode.LOG_REPLAY == this.getMasterFileSystem().getLogRecoveryMode()) {
1001
1002 Set<HRegionInfo> regions = new HashSet<HRegionInfo>();
1003 regions.add(HRegionInfo.FIRST_META_REGIONINFO);
1004 this.fileSystemManager.prepareLogReplay(currentMetaServer, regions);
1005 } else {
1006
1007 this.fileSystemManager.splitMetaLog(currentMetaServer);
1008 }
1009 }
1010
1011 private void enableCrashedServerProcessing(final boolean waitForMeta)
1012 throws IOException, InterruptedException {
1013
1014
1015
1016
1017 if (!isServerCrashProcessingEnabled()) {
1018 setServerCrashProcessingEnabled(true);
1019 this.serverManager.processQueuedDeadServers();
1020 }
1021
1022 if (waitForMeta) {
1023 metaTableLocator.waitMetaRegionLocation(this.getZooKeeper());
1024
1025
1026 this.assignmentManager.waitForAssignment(HRegionInfo.FIRST_META_REGIONINFO);
1027 }
1028 }
1029
1030 private void enableMeta(TableName metaTableName) {
1031 if (!this.assignmentManager.getTableStateManager().isTableState(metaTableName,
1032 ZooKeeperProtos.Table.State.ENABLED)) {
1033 this.assignmentManager.setEnabledTable(metaTableName);
1034 }
1035 }
1036
1037
1038
1039
1040
1041
1042 private Set<ServerName> getPreviouselyFailedMetaServersFromZK() throws KeeperException {
1043 Set<ServerName> result = new HashSet<ServerName>();
1044 String metaRecoveringZNode = ZKUtil.joinZNode(zooKeeper.recoveringRegionsZNode,
1045 HRegionInfo.FIRST_META_REGIONINFO.getEncodedName());
1046 List<String> regionFailedServers = ZKUtil.listChildrenNoWatch(zooKeeper, metaRecoveringZNode);
1047 if (regionFailedServers == null) return result;
1048
1049 for(String failedServer : regionFailedServers) {
1050 ServerName server = ServerName.parseServerName(failedServer);
1051 result.add(server);
1052 }
1053 return result;
1054 }
1055
1056 @Override
1057 public TableDescriptors getTableDescriptors() {
1058 return this.tableDescriptors;
1059 }
1060
1061 @Override
1062 public ServerManager getServerManager() {
1063 return this.serverManager;
1064 }
1065
1066 @Override
1067 public MasterFileSystem getMasterFileSystem() {
1068 return this.fileSystemManager;
1069 }
1070
1071
1072
1073
1074
1075
1076
1077
1078 private void startServiceThreads() throws IOException{
1079
1080 this.service.startExecutorService(ExecutorType.MASTER_OPEN_REGION,
1081 conf.getInt("hbase.master.executor.openregion.threads", 5));
1082 this.service.startExecutorService(ExecutorType.MASTER_CLOSE_REGION,
1083 conf.getInt("hbase.master.executor.closeregion.threads", 5));
1084 this.service.startExecutorService(ExecutorType.MASTER_SERVER_OPERATIONS,
1085 conf.getInt("hbase.master.executor.serverops.threads", 5));
1086 this.service.startExecutorService(ExecutorType.MASTER_META_SERVER_OPERATIONS,
1087 conf.getInt("hbase.master.executor.serverops.threads", 5));
1088 this.service.startExecutorService(ExecutorType.M_LOG_REPLAY_OPS,
1089 conf.getInt("hbase.master.executor.logreplayops.threads", 10));
1090
1091
1092
1093
1094
1095
1096 this.service.startExecutorService(ExecutorType.MASTER_TABLE_OPERATIONS, 1);
1097 startProcedureExecutor();
1098
1099
1100 int cleanerInterval = conf.getInt("hbase.master.cleaner.interval", 60 * 1000);
1101 this.logCleaner =
1102 new LogCleaner(cleanerInterval,
1103 this, conf, getMasterFileSystem().getFileSystem(),
1104 getMasterFileSystem().getOldLogDir());
1105 getChoreService().scheduleChore(logCleaner);
1106
1107
1108 Path archiveDir = HFileArchiveUtil.getArchivePath(conf);
1109 this.hfileCleaner = new HFileCleaner(cleanerInterval, this, conf, getMasterFileSystem()
1110 .getFileSystem(), archiveDir);
1111 getChoreService().scheduleChore(hfileCleaner);
1112 serviceStarted = true;
1113 if (LOG.isTraceEnabled()) {
1114 LOG.trace("Started service threads");
1115 }
1116 }
1117
1118 @Override
1119 protected void sendShutdownInterrupt() {
1120 super.sendShutdownInterrupt();
1121 stopProcedureExecutor();
1122 }
1123
1124 @Override
1125 protected void stopServiceThreads() {
1126 if (masterJettyServer != null) {
1127 LOG.info("Stopping master jetty server");
1128 try {
1129 masterJettyServer.stop();
1130 } catch (Exception e) {
1131 LOG.error("Failed to stop master jetty server", e);
1132 }
1133 }
1134 super.stopServiceThreads();
1135 stopChores();
1136
1137
1138
1139 if (!isAborted() && this.serverManager != null &&
1140 this.serverManager.isClusterShutdown()) {
1141 this.serverManager.letRegionServersShutdown();
1142 }
1143 if (LOG.isDebugEnabled()) {
1144 LOG.debug("Stopping service threads");
1145 }
1146
1147 if (this.logCleaner != null) this.logCleaner.cancel(true);
1148 if (this.hfileCleaner != null) this.hfileCleaner.cancel(true);
1149 if (this.quotaManager != null) this.quotaManager.stop();
1150 if (this.activeMasterManager != null) this.activeMasterManager.stop();
1151 if (this.serverManager != null) this.serverManager.stop();
1152 if (this.assignmentManager != null) this.assignmentManager.stop();
1153 if (this.fileSystemManager != null) this.fileSystemManager.stop();
1154 if (this.mpmHost != null) this.mpmHost.stop("server shutting down.");
1155 }
1156
1157 private void startProcedureExecutor() throws IOException {
1158 final MasterProcedureEnv procEnv = new MasterProcedureEnv(this);
1159 final Path logDir = new Path(fileSystemManager.getRootDir(),
1160 MasterProcedureConstants.MASTER_PROCEDURE_LOGDIR);
1161
1162 procedureStore = new WALProcedureStore(conf, fileSystemManager.getFileSystem(), logDir,
1163 new MasterProcedureEnv.WALStoreLeaseRecovery(this));
1164 procedureStore.registerListener(new MasterProcedureEnv.MasterProcedureStoreListener(this));
1165 procedureExecutor = new ProcedureExecutor(conf, procEnv, procedureStore,
1166 procEnv.getProcedureQueue());
1167
1168 final int numThreads = conf.getInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS,
1169 Math.max(Runtime.getRuntime().availableProcessors(),
1170 MasterProcedureConstants.DEFAULT_MIN_MASTER_PROCEDURE_THREADS));
1171 final boolean abortOnCorruption = conf.getBoolean(
1172 MasterProcedureConstants.EXECUTOR_ABORT_ON_CORRUPTION,
1173 MasterProcedureConstants.DEFAULT_EXECUTOR_ABORT_ON_CORRUPTION);
1174 procedureStore.start(numThreads);
1175 procedureExecutor.start(numThreads, abortOnCorruption);
1176 }
1177
1178 private void stopProcedureExecutor() {
1179 if (procedureExecutor != null) {
1180 procedureExecutor.stop();
1181 }
1182
1183 if (procedureStore != null) {
1184 procedureStore.stop(isAborted());
1185 }
1186 }
1187
1188 private void stopChores() {
1189 if (this.expiredMobFileCleanerChore != null) {
1190 this.expiredMobFileCleanerChore.cancel(true);
1191 }
1192 if (this.mobFileCompactChore != null) {
1193 this.mobFileCompactChore.cancel(true);
1194 }
1195 if (this.balancerChore != null) {
1196 this.balancerChore.cancel(true);
1197 }
1198 if (this.normalizerChore != null) {
1199 this.normalizerChore.cancel(true);
1200 }
1201 if (this.clusterStatusChore != null) {
1202 this.clusterStatusChore.cancel(true);
1203 }
1204 if (this.catalogJanitorChore != null) {
1205 this.catalogJanitorChore.cancel(true);
1206 }
1207 if (this.clusterStatusPublisherChore != null){
1208 clusterStatusPublisherChore.cancel(true);
1209 }
1210 if (this.mobFileCompactThread != null) {
1211 this.mobFileCompactThread.close();
1212 }
1213 }
1214
1215
1216
1217
1218
1219 InetAddress getRemoteInetAddress(final int port,
1220 final long serverStartCode) throws UnknownHostException {
1221
1222
1223 InetAddress ia = RpcServer.getRemoteIp();
1224
1225
1226
1227 if (ia == null && serverStartCode == startcode) {
1228 InetSocketAddress isa = rpcServices.getSocketAddress();
1229 if (isa != null && isa.getPort() == port) {
1230 ia = isa.getAddress();
1231 }
1232 }
1233 return ia;
1234 }
1235
1236
1237
1238
1239 private int getBalancerCutoffTime() {
1240 int balancerCutoffTime =
1241 getConfiguration().getInt("hbase.balancer.max.balancing", -1);
1242 if (balancerCutoffTime == -1) {
1243
1244 int balancerPeriod =
1245 getConfiguration().getInt("hbase.balancer.period", 300000);
1246 balancerCutoffTime = balancerPeriod;
1247
1248 if (balancerCutoffTime <= 0) balancerCutoffTime = balancerPeriod;
1249 }
1250 return balancerCutoffTime;
1251 }
1252
1253 public boolean balance() throws IOException {
1254
1255 if (!isInitialized()) {
1256 LOG.debug("Master has not been initialized, don't run balancer.");
1257 return false;
1258 }
1259
1260 int maximumBalanceTime = getBalancerCutoffTime();
1261 synchronized (this.balancer) {
1262
1263 if (!this.loadBalancerTracker.isBalancerOn()) return false;
1264
1265 if (this.assignmentManager.getRegionStates().isRegionsInTransition()) {
1266 Map<String, RegionState> regionsInTransition =
1267 this.assignmentManager.getRegionStates().getRegionsInTransition();
1268 LOG.debug("Not running balancer because " + regionsInTransition.size() +
1269 " region(s) in transition: " + org.apache.commons.lang.StringUtils.
1270 abbreviate(regionsInTransition.toString(), 256));
1271 return false;
1272 }
1273 if (this.serverManager.areDeadServersInProgress()) {
1274 LOG.debug("Not running balancer because processing dead regionserver(s): " +
1275 this.serverManager.getDeadServers());
1276 return false;
1277 }
1278
1279 if (this.cpHost != null) {
1280 try {
1281 if (this.cpHost.preBalance()) {
1282 LOG.debug("Coprocessor bypassing balancer request");
1283 return false;
1284 }
1285 } catch (IOException ioe) {
1286 LOG.error("Error invoking master coprocessor preBalance()", ioe);
1287 return false;
1288 }
1289 }
1290
1291 Map<TableName, Map<ServerName, List<HRegionInfo>>> assignmentsByTable =
1292 this.assignmentManager.getRegionStates().getAssignmentsByTable();
1293
1294 List<RegionPlan> plans = new ArrayList<RegionPlan>();
1295
1296 this.balancer.setClusterStatus(getClusterStatus());
1297 for (Map<ServerName, List<HRegionInfo>> assignments : assignmentsByTable.values()) {
1298 List<RegionPlan> partialPlans = this.balancer.balanceCluster(assignments);
1299 if (partialPlans != null) plans.addAll(partialPlans);
1300 }
1301 long cutoffTime = System.currentTimeMillis() + maximumBalanceTime;
1302 int rpCount = 0;
1303 long totalRegPlanExecTime = 0;
1304 if (plans != null && !plans.isEmpty()) {
1305 for (RegionPlan plan: plans) {
1306 LOG.info("balance " + plan);
1307 long balStartTime = System.currentTimeMillis();
1308
1309 this.assignmentManager.balance(plan);
1310 totalRegPlanExecTime += System.currentTimeMillis()-balStartTime;
1311 rpCount++;
1312 if (rpCount < plans.size() &&
1313
1314 (System.currentTimeMillis() + (totalRegPlanExecTime / rpCount)) > cutoffTime) {
1315
1316 LOG.debug("No more balancing till next balance run; maximumBalanceTime=" +
1317 maximumBalanceTime);
1318 break;
1319 }
1320 }
1321 }
1322 if (this.cpHost != null) {
1323 try {
1324 this.cpHost.postBalance(rpCount < plans.size() ? plans.subList(0, rpCount) : plans);
1325 } catch (IOException ioe) {
1326
1327 LOG.error("Error invoking master coprocessor postBalance()", ioe);
1328 }
1329 }
1330 }
1331
1332
1333 return true;
1334 }
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345 public boolean normalizeRegions() throws IOException, CoordinatedStateException {
1346 if (!isInitialized()) {
1347 LOG.debug("Master has not been initialized, don't run region normalizer.");
1348 return false;
1349 }
1350
1351 if (!this.regionNormalizerTracker.isNormalizerOn()) {
1352 LOG.debug("Region normalization is disabled, don't run region normalizer.");
1353 return false;
1354 }
1355
1356 synchronized (this.normalizer) {
1357
1358 List<TableName> allEnabledTables = new ArrayList<>(
1359 this.assignmentManager.getTableStateManager().getTablesInStates(
1360 ZooKeeperProtos.Table.State.ENABLED));
1361
1362 Collections.shuffle(allEnabledTables);
1363
1364 for (TableName table : allEnabledTables) {
1365 if (quotaManager.getNamespaceQuotaManager() != null &&
1366 quotaManager.getNamespaceQuotaManager().getState(table.getNamespaceAsString()) != null){
1367 LOG.debug("Skipping normalizing " + table + " since its namespace has quota");
1368 continue;
1369 }
1370 if (table.isSystemTable() || (getTableDescriptors().get(table) != null &&
1371 !getTableDescriptors().get(table).isNormalizationEnabled())) {
1372 LOG.debug("Skipping normalization for table: " + table + ", as it's either system"
1373 + " table or doesn't have auto normalization turned on");
1374 continue;
1375 }
1376 List<NormalizationPlan> plans = this.normalizer.computePlanForTable(table);
1377 if (plans != null) {
1378 for (NormalizationPlan plan : plans) {
1379 plan.execute(clusterConnection.getAdmin());
1380 }
1381 }
1382 }
1383 }
1384
1385
1386 return true;
1387 }
1388
1389
1390
1391
1392 String getClientIdAuditPrefix() {
1393 return "Client=" + RpcServer.getRequestUserName() + "/" + RpcServer.getRemoteAddress();
1394 }
1395
1396
1397
1398
1399
1400
1401
1402 public void setCatalogJanitorEnabled(final boolean b) {
1403 this.catalogJanitorChore.setEnabled(b);
1404 }
1405
1406 @Override
1407 public void dispatchMergingRegions(final HRegionInfo region_a,
1408 final HRegionInfo region_b, final boolean forcible) throws IOException {
1409 checkInitialized();
1410 this.service.submit(new DispatchMergingRegionHandler(this,
1411 this.catalogJanitorChore, region_a, region_b, forcible));
1412 }
1413
1414 void move(final byte[] encodedRegionName,
1415 final byte[] destServerName) throws HBaseIOException {
1416 RegionState regionState = assignmentManager.getRegionStates().
1417 getRegionState(Bytes.toString(encodedRegionName));
1418 if (regionState == null) {
1419 throw new UnknownRegionException(Bytes.toStringBinary(encodedRegionName));
1420 }
1421
1422 HRegionInfo hri = regionState.getRegion();
1423 ServerName dest;
1424 if (destServerName == null || destServerName.length == 0) {
1425 LOG.info("Passed destination servername is null/empty so " +
1426 "choosing a server at random");
1427 final List<ServerName> destServers = this.serverManager.createDestinationServersList(
1428 regionState.getServerName());
1429 dest = balancer.randomAssignment(hri, destServers);
1430 if (dest == null) {
1431 LOG.debug("Unable to determine a plan to assign " + hri);
1432 return;
1433 }
1434 } else {
1435 dest = ServerName.valueOf(Bytes.toString(destServerName));
1436 if (dest.equals(serverName) && balancer instanceof BaseLoadBalancer
1437 && !((BaseLoadBalancer)balancer).shouldBeOnMaster(hri)) {
1438
1439
1440
1441 LOG.debug("Skipping move of region " + hri.getRegionNameAsString()
1442 + " to avoid unnecessary region moving later by load balancer,"
1443 + " because it should not be on master");
1444 return;
1445 }
1446 }
1447
1448 if (dest.equals(regionState.getServerName())) {
1449 LOG.debug("Skipping move of region " + hri.getRegionNameAsString()
1450 + " because region already assigned to the same server " + dest + ".");
1451 return;
1452 }
1453
1454
1455 RegionPlan rp = new RegionPlan(hri, regionState.getServerName(), dest);
1456
1457 try {
1458 checkInitialized();
1459 if (this.cpHost != null) {
1460 if (this.cpHost.preMove(hri, rp.getSource(), rp.getDestination())) {
1461 return;
1462 }
1463 }
1464
1465
1466
1467 serverManager.sendRegionWarmup(rp.getDestination(), hri);
1468
1469 LOG.info(getClientIdAuditPrefix() + " move " + rp + ", running balancer");
1470 this.assignmentManager.balance(rp);
1471 if (this.cpHost != null) {
1472 this.cpHost.postMove(hri, rp.getSource(), rp.getDestination());
1473 }
1474 } catch (IOException ioe) {
1475 if (ioe instanceof HBaseIOException) {
1476 throw (HBaseIOException)ioe;
1477 }
1478 throw new HBaseIOException(ioe);
1479 }
1480 }
1481
1482 @Override
1483 public long createTable(
1484 final HTableDescriptor hTableDescriptor,
1485 final byte [][] splitKeys,
1486 final long nonceGroup,
1487 final long nonce) throws IOException {
1488 if (isStopped()) {
1489 throw new MasterNotRunningException();
1490 }
1491
1492 String namespace = hTableDescriptor.getTableName().getNamespaceAsString();
1493 ensureNamespaceExists(namespace);
1494
1495 HRegionInfo[] newRegions = ModifyRegionUtils.createHRegionInfos(hTableDescriptor, splitKeys);
1496 checkInitialized();
1497 sanityCheckTableDescriptor(hTableDescriptor);
1498
1499 if (cpHost != null) {
1500 cpHost.preCreateTable(hTableDescriptor, newRegions);
1501 }
1502 LOG.info(getClientIdAuditPrefix() + " create " + hTableDescriptor);
1503
1504
1505
1506 ProcedurePrepareLatch latch = ProcedurePrepareLatch.createLatch();
1507 long procId = this.procedureExecutor.submitProcedure(
1508 new CreateTableProcedure(
1509 procedureExecutor.getEnvironment(), hTableDescriptor, newRegions, latch),
1510 nonceGroup,
1511 nonce);
1512 latch.await();
1513
1514 if (cpHost != null) {
1515 cpHost.postCreateTable(hTableDescriptor, newRegions);
1516 }
1517
1518 return procId;
1519 }
1520
1521
1522
1523
1524
1525
1526 private void sanityCheckTableDescriptor(final HTableDescriptor htd) throws IOException {
1527 final String CONF_KEY = "hbase.table.sanity.checks";
1528 boolean logWarn = false;
1529 if (!conf.getBoolean(CONF_KEY, true)) {
1530 logWarn = true;
1531 }
1532 String tableVal = htd.getConfigurationValue(CONF_KEY);
1533 if (tableVal != null && !Boolean.valueOf(tableVal)) {
1534 logWarn = true;
1535 }
1536
1537
1538 long maxFileSizeLowerLimit = 2 * 1024 * 1024L;
1539 long maxFileSize = htd.getMaxFileSize();
1540 if (maxFileSize < 0) {
1541 maxFileSize = conf.getLong(HConstants.HREGION_MAX_FILESIZE, maxFileSizeLowerLimit);
1542 }
1543 if (maxFileSize < conf.getLong("hbase.hregion.max.filesize.limit", maxFileSizeLowerLimit)) {
1544 String message = "MAX_FILESIZE for table descriptor or "
1545 + "\"hbase.hregion.max.filesize\" (" + maxFileSize
1546 + ") is too small, which might cause over splitting into unmanageable "
1547 + "number of regions.";
1548 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null);
1549 }
1550
1551
1552 long flushSizeLowerLimit = 1024 * 1024L;
1553 long flushSize = htd.getMemStoreFlushSize();
1554 if (flushSize < 0) {
1555 flushSize = conf.getLong(HConstants.HREGION_MEMSTORE_FLUSH_SIZE, flushSizeLowerLimit);
1556 }
1557 if (flushSize < conf.getLong("hbase.hregion.memstore.flush.size.limit", flushSizeLowerLimit)) {
1558 String message = "MEMSTORE_FLUSHSIZE for table descriptor or "
1559 + "\"hbase.hregion.memstore.flush.size\" ("+flushSize+") is too small, which might cause"
1560 + " very frequent flushing.";
1561 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null);
1562 }
1563
1564
1565 try {
1566 checkClassLoading(conf, htd);
1567 } catch (Exception ex) {
1568 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, ex.getMessage(), null);
1569 }
1570
1571
1572 try {
1573 checkCompression(htd);
1574 } catch (IOException e) {
1575 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, e.getMessage(), e);
1576 }
1577
1578
1579 try {
1580 checkEncryption(conf, htd);
1581 } catch (IOException e) {
1582 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, e.getMessage(), e);
1583 }
1584
1585 try{
1586 checkCompactionPolicy(conf, htd);
1587 } catch(IOException e){
1588 warnOrThrowExceptionForFailure(false, CONF_KEY, e.getMessage(), e);
1589 }
1590
1591 if (htd.getColumnFamilies().length == 0) {
1592 String message = "Table should have at least one column family.";
1593 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null);
1594 }
1595
1596 for (HColumnDescriptor hcd : htd.getColumnFamilies()) {
1597 if (hcd.getTimeToLive() <= 0) {
1598 String message = "TTL for column family " + hcd.getNameAsString() + " must be positive.";
1599 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null);
1600 }
1601
1602
1603 if (hcd.getBlocksize() < 1024 || hcd.getBlocksize() > 16 * 1024 * 1024) {
1604 String message = "Block size for column family " + hcd.getNameAsString()
1605 + " must be between 1K and 16MB.";
1606 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null);
1607 }
1608
1609
1610 if (hcd.getMinVersions() < 0) {
1611 String message = "Min versions for column family " + hcd.getNameAsString()
1612 + " must be positive.";
1613 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null);
1614 }
1615
1616
1617
1618
1619
1620 if (hcd.getMinVersions() > hcd.getMaxVersions()) {
1621 String message = "Min versions for column family " + hcd.getNameAsString()
1622 + " must be less than the Max versions.";
1623 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null);
1624 }
1625
1626
1627 if (hcd.getScope() < 0) {
1628 String message = "Replication scope for column family "
1629 + hcd.getNameAsString() + " must be positive.";
1630 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null);
1631 }
1632
1633
1634
1635 if (hcd.getDFSReplication() < 0) {
1636 String message = "HFile Replication for column family " + hcd.getNameAsString()
1637 + " must be greater than zero.";
1638 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null);
1639 }
1640
1641
1642 }
1643 }
1644
1645 private void checkCompactionPolicy(Configuration conf, HTableDescriptor htd)
1646 throws IOException {
1647
1648
1649 String className =
1650 htd.getConfigurationValue(DefaultStoreEngine.DEFAULT_COMPACTION_POLICY_CLASS_KEY);
1651 if (className == null) {
1652 className =
1653 conf.get(DefaultStoreEngine.DEFAULT_COMPACTION_POLICY_CLASS_KEY,
1654 ExploringCompactionPolicy.class.getName());
1655 }
1656
1657 int blockingFileCount = HStore.DEFAULT_BLOCKING_STOREFILE_COUNT;
1658 String sv = htd.getConfigurationValue(HStore.BLOCKING_STOREFILES_KEY);
1659 if (sv != null) {
1660 blockingFileCount = Integer.parseInt(sv);
1661 } else {
1662 blockingFileCount = conf.getInt(HStore.BLOCKING_STOREFILES_KEY, blockingFileCount);
1663 }
1664
1665 for (HColumnDescriptor hcd : htd.getColumnFamilies()) {
1666 String compactionPolicy =
1667 hcd.getConfigurationValue(DefaultStoreEngine.DEFAULT_COMPACTION_POLICY_CLASS_KEY);
1668 if (compactionPolicy == null) {
1669 compactionPolicy = className;
1670 }
1671 if (!compactionPolicy.equals(FIFOCompactionPolicy.class.getName())) {
1672 continue;
1673 }
1674
1675 String message = null;
1676
1677
1678 if (hcd.getTimeToLive() == HColumnDescriptor.DEFAULT_TTL) {
1679 message = "Default TTL is not supported for FIFO compaction";
1680 throw new IOException(message);
1681 }
1682
1683
1684 if (hcd.getMinVersions() > 0) {
1685 message = "MIN_VERSION > 0 is not supported for FIFO compaction";
1686 throw new IOException(message);
1687 }
1688
1689
1690 String sbfc = htd.getConfigurationValue(HStore.BLOCKING_STOREFILES_KEY);
1691 if (sbfc != null) {
1692 blockingFileCount = Integer.parseInt(sbfc);
1693 }
1694 if (blockingFileCount < 1000) {
1695 message =
1696 "blocking file count '" + HStore.BLOCKING_STOREFILES_KEY + "' " + blockingFileCount
1697 + " is below recommended minimum of 1000";
1698 throw new IOException(message);
1699 }
1700 }
1701 }
1702
1703
1704 private static void warnOrThrowExceptionForFailure(boolean logWarn, String confKey,
1705 String message, Exception cause) throws IOException {
1706 if (!logWarn) {
1707 throw new DoNotRetryIOException(message + " Set " + confKey +
1708 " to false at conf or table descriptor if you want to bypass sanity checks", cause);
1709 }
1710 LOG.warn(message);
1711 }
1712
1713 private void startActiveMasterManager(int infoPort) throws KeeperException {
1714 String backupZNode = ZKUtil.joinZNode(
1715 zooKeeper.backupMasterAddressesZNode, serverName.toString());
1716
1717
1718
1719
1720
1721
1722
1723
1724
1725
1726 LOG.info("Adding backup master ZNode " + backupZNode);
1727 if (!MasterAddressTracker.setMasterAddress(zooKeeper, backupZNode,
1728 serverName, infoPort)) {
1729 LOG.warn("Failed create of " + backupZNode + " by " + serverName);
1730 }
1731
1732 activeMasterManager.setInfoPort(infoPort);
1733
1734 Threads.setDaemonThreadRunning(new Thread(new Runnable() {
1735 @Override
1736 public void run() {
1737 int timeout = conf.getInt(HConstants.ZK_SESSION_TIMEOUT,
1738 HConstants.DEFAULT_ZK_SESSION_TIMEOUT);
1739
1740 if (conf.getBoolean(HConstants.MASTER_TYPE_BACKUP,
1741 HConstants.DEFAULT_MASTER_TYPE_BACKUP)) {
1742 LOG.debug("HMaster started in backup mode. "
1743 + "Stalling until master znode is written.");
1744
1745
1746 while (!activeMasterManager.hasActiveMaster()) {
1747 LOG.debug("Waiting for master address ZNode to be written "
1748 + "(Also watching cluster state node)");
1749 Threads.sleep(timeout);
1750 }
1751 }
1752 MonitoredTask status = TaskMonitor.get().createStatus("Master startup");
1753 status.setDescription("Master startup");
1754 try {
1755 if (activeMasterManager.blockUntilBecomingActiveMaster(timeout, status)) {
1756 finishActiveMasterInitialization(status);
1757 }
1758 } catch (Throwable t) {
1759 status.setStatus("Failed to become active: " + t.getMessage());
1760 LOG.fatal("Failed to become active master", t);
1761
1762 if (t instanceof NoClassDefFoundError &&
1763 t.getMessage()
1764 .contains("org/apache/hadoop/hdfs/protocol/HdfsConstants$SafeModeAction")) {
1765
1766 abort("HBase is having a problem with its Hadoop jars. You may need to "
1767 + "recompile HBase against Hadoop version "
1768 + org.apache.hadoop.util.VersionInfo.getVersion()
1769 + " or change your hadoop jars to start properly", t);
1770 } else {
1771 abort("Unhandled exception. Starting shutdown.", t);
1772 }
1773 } finally {
1774 status.cleanup();
1775 }
1776 }
1777 }, getServerName().toShortString() + ".activeMasterManager"));
1778 }
1779
1780 private void checkCompression(final HTableDescriptor htd)
1781 throws IOException {
1782 if (!this.masterCheckCompression) return;
1783 for (HColumnDescriptor hcd : htd.getColumnFamilies()) {
1784 checkCompression(hcd);
1785 }
1786 }
1787
1788 private void checkCompression(final HColumnDescriptor hcd)
1789 throws IOException {
1790 if (!this.masterCheckCompression) return;
1791 CompressionTest.testCompression(hcd.getCompression());
1792 CompressionTest.testCompression(hcd.getCompactionCompression());
1793 }
1794
1795 private void checkEncryption(final Configuration conf, final HTableDescriptor htd)
1796 throws IOException {
1797 if (!this.masterCheckEncryption) return;
1798 for (HColumnDescriptor hcd : htd.getColumnFamilies()) {
1799 checkEncryption(conf, hcd);
1800 }
1801 }
1802
1803 private void checkEncryption(final Configuration conf, final HColumnDescriptor hcd)
1804 throws IOException {
1805 if (!this.masterCheckEncryption) return;
1806 EncryptionTest.testEncryption(conf, hcd.getEncryptionType(), hcd.getEncryptionKey());
1807 }
1808
1809 private void checkClassLoading(final Configuration conf, final HTableDescriptor htd)
1810 throws IOException {
1811 RegionSplitPolicy.getSplitPolicyClass(htd, conf);
1812 RegionCoprocessorHost.testTableCoprocessorAttrs(conf, htd);
1813 }
1814
1815 private static boolean isCatalogTable(final TableName tableName) {
1816 return tableName.equals(TableName.META_TABLE_NAME);
1817 }
1818
1819 @Override
1820 public long deleteTable(
1821 final TableName tableName,
1822 final long nonceGroup,
1823 final long nonce) throws IOException {
1824 checkInitialized();
1825 if (cpHost != null) {
1826 cpHost.preDeleteTable(tableName);
1827 }
1828 LOG.info(getClientIdAuditPrefix() + " delete " + tableName);
1829
1830
1831 ProcedurePrepareLatch latch = ProcedurePrepareLatch.createLatch();
1832 long procId = this.procedureExecutor.submitProcedure(
1833 new DeleteTableProcedure(procedureExecutor.getEnvironment(), tableName, latch),
1834 nonceGroup,
1835 nonce);
1836 latch.await();
1837
1838 if (cpHost != null) {
1839 cpHost.postDeleteTable(tableName);
1840 }
1841
1842 return procId;
1843 }
1844
1845 @Override
1846 public void truncateTable(
1847 final TableName tableName,
1848 final boolean preserveSplits,
1849 final long nonceGroup,
1850 final long nonce) throws IOException {
1851 checkInitialized();
1852 if (cpHost != null) {
1853 cpHost.preTruncateTable(tableName);
1854 }
1855 LOG.info(getClientIdAuditPrefix() + " truncate " + tableName);
1856
1857 long procId = this.procedureExecutor.submitProcedure(
1858 new TruncateTableProcedure(procedureExecutor.getEnvironment(), tableName, preserveSplits),
1859 nonceGroup,
1860 nonce);
1861 ProcedureSyncWait.waitForProcedureToComplete(procedureExecutor, procId);
1862
1863 if (cpHost != null) {
1864 cpHost.postTruncateTable(tableName);
1865 }
1866 }
1867
1868 @Override
1869 public void addColumn(
1870 final TableName tableName,
1871 final HColumnDescriptor columnDescriptor,
1872 final long nonceGroup,
1873 final long nonce)
1874 throws IOException {
1875 checkInitialized();
1876 checkCompression(columnDescriptor);
1877 checkEncryption(conf, columnDescriptor);
1878 if (cpHost != null) {
1879 if (cpHost.preAddColumn(tableName, columnDescriptor)) {
1880 return;
1881 }
1882 }
1883
1884 long procId = this.procedureExecutor.submitProcedure(
1885 new AddColumnFamilyProcedure(procedureExecutor.getEnvironment(), tableName, columnDescriptor),
1886 nonceGroup,
1887 nonce);
1888 ProcedureSyncWait.waitForProcedureToComplete(procedureExecutor, procId);
1889 if (cpHost != null) {
1890 cpHost.postAddColumn(tableName, columnDescriptor);
1891 }
1892 }
1893
1894 @Override
1895 public void modifyColumn(
1896 final TableName tableName,
1897 final HColumnDescriptor descriptor,
1898 final long nonceGroup,
1899 final long nonce)
1900 throws IOException {
1901 checkInitialized();
1902 checkCompression(descriptor);
1903 checkEncryption(conf, descriptor);
1904 if (cpHost != null) {
1905 if (cpHost.preModifyColumn(tableName, descriptor)) {
1906 return;
1907 }
1908 }
1909 LOG.info(getClientIdAuditPrefix() + " modify " + descriptor);
1910
1911
1912 long procId = this.procedureExecutor.submitProcedure(
1913 new ModifyColumnFamilyProcedure(procedureExecutor.getEnvironment(), tableName, descriptor),
1914 nonceGroup,
1915 nonce);
1916 ProcedureSyncWait.waitForProcedureToComplete(procedureExecutor, procId);
1917
1918 if (cpHost != null) {
1919 cpHost.postModifyColumn(tableName, descriptor);
1920 }
1921 }
1922
1923 @Override
1924 public void deleteColumn(
1925 final TableName tableName,
1926 final byte[] columnName,
1927 final long nonceGroup,
1928 final long nonce)
1929 throws IOException {
1930 checkInitialized();
1931 if (cpHost != null) {
1932 if (cpHost.preDeleteColumn(tableName, columnName)) {
1933 return;
1934 }
1935 }
1936 LOG.info(getClientIdAuditPrefix() + " delete " + Bytes.toString(columnName));
1937
1938
1939 long procId = this.procedureExecutor.submitProcedure(
1940 new DeleteColumnFamilyProcedure(procedureExecutor.getEnvironment(), tableName, columnName),
1941 nonceGroup,
1942 nonce);
1943 ProcedureSyncWait.waitForProcedureToComplete(procedureExecutor, procId);
1944
1945 if (cpHost != null) {
1946 cpHost.postDeleteColumn(tableName, columnName);
1947 }
1948 }
1949
1950 @Override
1951 public long enableTable(
1952 final TableName tableName,
1953 final long nonceGroup,
1954 final long nonce) throws IOException {
1955 checkInitialized();
1956 if (cpHost != null) {
1957 cpHost.preEnableTable(tableName);
1958 }
1959 LOG.info(getClientIdAuditPrefix() + " enable " + tableName);
1960
1961
1962 final ProcedurePrepareLatch prepareLatch = ProcedurePrepareLatch.createLatch();
1963 long procId = this.procedureExecutor.submitProcedure(
1964 new EnableTableProcedure(procedureExecutor.getEnvironment(), tableName, false, prepareLatch),
1965 nonceGroup,
1966 nonce);
1967
1968
1969
1970
1971 prepareLatch.await();
1972
1973 if (cpHost != null) {
1974 cpHost.postEnableTable(tableName);
1975 }
1976
1977 return procId;
1978 }
1979
1980 @Override
1981 public long disableTable(
1982 final TableName tableName,
1983 final long nonceGroup,
1984 final long nonce) throws IOException {
1985 checkInitialized();
1986 if (cpHost != null) {
1987 cpHost.preDisableTable(tableName);
1988 }
1989 LOG.info(getClientIdAuditPrefix() + " disable " + tableName);
1990
1991
1992 final ProcedurePrepareLatch prepareLatch = ProcedurePrepareLatch.createLatch();
1993
1994 long procId = this.procedureExecutor.submitProcedure(
1995 new DisableTableProcedure(procedureExecutor.getEnvironment(), tableName, false, prepareLatch),
1996 nonceGroup,
1997 nonce);
1998
1999
2000
2001
2002 prepareLatch.await();
2003
2004 if (cpHost != null) {
2005 cpHost.postDisableTable(tableName);
2006 }
2007
2008 return procId;
2009 }
2010
2011
2012
2013
2014
2015
2016
2017 @VisibleForTesting
2018 Pair<HRegionInfo, ServerName> getTableRegionForRow(
2019 final TableName tableName, final byte [] rowKey)
2020 throws IOException {
2021 final AtomicReference<Pair<HRegionInfo, ServerName>> result =
2022 new AtomicReference<Pair<HRegionInfo, ServerName>>(null);
2023
2024 MetaScannerVisitor visitor =
2025 new MetaScannerVisitorBase() {
2026 @Override
2027 public boolean processRow(Result data) throws IOException {
2028 if (data == null || data.size() <= 0) {
2029 return true;
2030 }
2031 Pair<HRegionInfo, ServerName> pair = HRegionInfo.getHRegionInfoAndServerName(data);
2032 if (pair == null) {
2033 return false;
2034 }
2035 if (!pair.getFirst().getTable().equals(tableName)) {
2036 return false;
2037 }
2038 result.set(pair);
2039 return true;
2040 }
2041 };
2042
2043 MetaScanner.metaScan(clusterConnection, visitor, tableName, rowKey, 1);
2044 return result.get();
2045 }
2046
2047 @Override
2048 public void modifyTable(
2049 final TableName tableName,
2050 final HTableDescriptor descriptor,
2051 final long nonceGroup,
2052 final long nonce)
2053 throws IOException {
2054 checkInitialized();
2055 sanityCheckTableDescriptor(descriptor);
2056 if (cpHost != null) {
2057 cpHost.preModifyTable(tableName, descriptor);
2058 }
2059
2060 LOG.info(getClientIdAuditPrefix() + " modify " + tableName);
2061
2062
2063 long procId = this.procedureExecutor.submitProcedure(
2064 new ModifyTableProcedure(procedureExecutor.getEnvironment(), descriptor),
2065 nonceGroup,
2066 nonce);
2067
2068 ProcedureSyncWait.waitForProcedureToComplete(procedureExecutor, procId);
2069
2070 if (cpHost != null) {
2071 cpHost.postModifyTable(tableName, descriptor);
2072 }
2073 }
2074
2075 @Override
2076 public void checkTableModifiable(final TableName tableName)
2077 throws IOException, TableNotFoundException, TableNotDisabledException {
2078 if (isCatalogTable(tableName)) {
2079 throw new IOException("Can't modify catalog tables");
2080 }
2081 if (!MetaTableAccessor.tableExists(getConnection(), tableName)) {
2082 throw new TableNotFoundException(tableName);
2083 }
2084 if (!getAssignmentManager().getTableStateManager().
2085 isTableState(tableName, ZooKeeperProtos.Table.State.DISABLED)) {
2086 throw new TableNotDisabledException(tableName);
2087 }
2088 }
2089
2090
2091
2092
2093 public ClusterStatus getClusterStatus() throws InterruptedIOException {
2094
2095 List<String> backupMasterStrings;
2096 try {
2097 backupMasterStrings = ZKUtil.listChildrenNoWatch(this.zooKeeper,
2098 this.zooKeeper.backupMasterAddressesZNode);
2099 } catch (KeeperException e) {
2100 LOG.warn(this.zooKeeper.prefix("Unable to list backup servers"), e);
2101 backupMasterStrings = null;
2102 }
2103
2104 List<ServerName> backupMasters = null;
2105 if (backupMasterStrings != null && !backupMasterStrings.isEmpty()) {
2106 backupMasters = new ArrayList<ServerName>(backupMasterStrings.size());
2107 for (String s: backupMasterStrings) {
2108 try {
2109 byte [] bytes;
2110 try {
2111 bytes = ZKUtil.getData(this.zooKeeper, ZKUtil.joinZNode(
2112 this.zooKeeper.backupMasterAddressesZNode, s));
2113 } catch (InterruptedException e) {
2114 throw new InterruptedIOException();
2115 }
2116 if (bytes != null) {
2117 ServerName sn;
2118 try {
2119 sn = ServerName.parseFrom(bytes);
2120 } catch (DeserializationException e) {
2121 LOG.warn("Failed parse, skipping registering backup server", e);
2122 continue;
2123 }
2124 backupMasters.add(sn);
2125 }
2126 } catch (KeeperException e) {
2127 LOG.warn(this.zooKeeper.prefix("Unable to get information about " +
2128 "backup servers"), e);
2129 }
2130 }
2131 Collections.sort(backupMasters, new Comparator<ServerName>() {
2132 @Override
2133 public int compare(ServerName s1, ServerName s2) {
2134 return s1.getServerName().compareTo(s2.getServerName());
2135 }});
2136 }
2137
2138 String clusterId = fileSystemManager != null ?
2139 fileSystemManager.getClusterId().toString() : null;
2140 Map<String, RegionState> regionsInTransition = assignmentManager != null ?
2141 assignmentManager.getRegionStates().getRegionsInTransition() : null;
2142 String[] coprocessors = cpHost != null ? getMasterCoprocessors() : null;
2143 boolean balancerOn = loadBalancerTracker != null ?
2144 loadBalancerTracker.isBalancerOn() : false;
2145 Map<ServerName, ServerLoad> onlineServers = null;
2146 Set<ServerName> deadServers = null;
2147 if (serverManager != null) {
2148 deadServers = serverManager.getDeadServers().copyServerNames();
2149 onlineServers = serverManager.getOnlineServers();
2150 }
2151 return new ClusterStatus(VersionInfo.getVersion(), clusterId,
2152 onlineServers, deadServers, serverName, backupMasters,
2153 regionsInTransition, coprocessors, balancerOn);
2154 }
2155
2156
2157
2158
2159
2160
2161
2162
2163 public static String getLoadedCoprocessors() {
2164 return CoprocessorHost.getLoadedCoprocessors().toString();
2165 }
2166
2167
2168
2169
2170 public long getMasterStartTime() {
2171 return startcode;
2172 }
2173
2174
2175
2176
2177 public long getMasterActiveTime() {
2178 return masterActiveTime;
2179 }
2180
2181 public int getNumWALFiles() {
2182 return procedureStore != null ? procedureStore.getActiveLogs().size() : 0;
2183 }
2184
2185 public WALProcedureStore getWalProcedureStore() {
2186 return procedureStore;
2187 }
2188
2189 public int getRegionServerInfoPort(final ServerName sn) {
2190 RegionServerInfo info = this.regionServerTracker.getRegionServerInfo(sn);
2191 if (info == null || info.getInfoPort() == 0) {
2192 return conf.getInt(HConstants.REGIONSERVER_INFO_PORT,
2193 HConstants.DEFAULT_REGIONSERVER_INFOPORT);
2194 }
2195 return info.getInfoPort();
2196 }
2197
2198 public String getRegionServerVersion(final ServerName sn) {
2199 RegionServerInfo info = this.regionServerTracker.getRegionServerInfo(sn);
2200 if (info != null && info.hasVersionInfo()) {
2201 return info.getVersionInfo().getVersion();
2202 }
2203 return "Unknown";
2204 }
2205
2206
2207
2208
2209 public String[] getMasterCoprocessors() {
2210 Set<String> masterCoprocessors = getMasterCoprocessorHost().getCoprocessors();
2211 return masterCoprocessors.toArray(new String[masterCoprocessors.size()]);
2212 }
2213
2214 @Override
2215 public void abort(final String msg, final Throwable t) {
2216 if (isAborted() || isStopped()) {
2217 return;
2218 }
2219 if (cpHost != null) {
2220
2221 LOG.fatal("Master server abort: loaded coprocessors are: " +
2222 getLoadedCoprocessors());
2223 }
2224 if (t != null) LOG.fatal(msg, t);
2225 stop(msg);
2226 }
2227
2228 @Override
2229 public ZooKeeperWatcher getZooKeeper() {
2230 return zooKeeper;
2231 }
2232
2233 @Override
2234 public MasterCoprocessorHost getMasterCoprocessorHost() {
2235 return cpHost;
2236 }
2237
2238 @Override
2239 public MasterQuotaManager getMasterQuotaManager() {
2240 return quotaManager;
2241 }
2242
2243 @Override
2244 public ProcedureExecutor<MasterProcedureEnv> getMasterProcedureExecutor() {
2245 return procedureExecutor;
2246 }
2247
2248 @Override
2249 public ServerName getServerName() {
2250 return this.serverName;
2251 }
2252
2253 @Override
2254 public AssignmentManager getAssignmentManager() {
2255 return this.assignmentManager;
2256 }
2257
2258 public MemoryBoundedLogMessageBuffer getRegionServerFatalLogBuffer() {
2259 return rsFatals;
2260 }
2261
2262 public void shutdown() {
2263 if (cpHost != null) {
2264 try {
2265 cpHost.preShutdown();
2266 } catch (IOException ioe) {
2267 LOG.error("Error call master coprocessor preShutdown()", ioe);
2268 }
2269 }
2270
2271 if (this.serverManager != null) {
2272 this.serverManager.shutdownCluster();
2273 }
2274 if (this.clusterStatusTracker != null){
2275 try {
2276 this.clusterStatusTracker.setClusterDown();
2277 } catch (KeeperException e) {
2278 LOG.error("ZooKeeper exception trying to set cluster as down in ZK", e);
2279 }
2280 }
2281 }
2282
2283 public void stopMaster() {
2284 if (cpHost != null) {
2285 try {
2286 cpHost.preStopMaster();
2287 } catch (IOException ioe) {
2288 LOG.error("Error call master coprocessor preStopMaster()", ioe);
2289 }
2290 }
2291 stop("Stopped by " + Thread.currentThread().getName());
2292 }
2293
2294 void checkServiceStarted() throws ServerNotRunningYetException {
2295 if (!serviceStarted) {
2296 throw new ServerNotRunningYetException("Server is not running yet");
2297 }
2298 }
2299
2300 void checkInitialized() throws PleaseHoldException, ServerNotRunningYetException {
2301 checkServiceStarted();
2302 if (!isInitialized()) {
2303 throw new PleaseHoldException("Master is initializing");
2304 }
2305 }
2306
2307 void checkNamespaceManagerReady() throws IOException {
2308 checkInitialized();
2309 if (tableNamespaceManager == null ||
2310 !tableNamespaceManager.isTableAvailableAndInitialized()) {
2311 throw new IOException("Table Namespace Manager not ready yet, try again later");
2312 }
2313 }
2314
2315
2316
2317
2318
2319
2320
2321
2322 public boolean isActiveMaster() {
2323 return isActiveMaster;
2324 }
2325
2326
2327
2328
2329
2330
2331
2332
2333
2334
2335 @Override
2336 public boolean isInitialized() {
2337 return initialized.isReady();
2338 }
2339
2340 @VisibleForTesting
2341 public void setInitialized(boolean isInitialized) {
2342 procedureExecutor.getEnvironment().setEventReady(initialized, isInitialized);
2343 }
2344
2345 public ProcedureEvent getInitializedEvent() {
2346 return initialized;
2347 }
2348
2349
2350
2351
2352
2353
2354 @Override
2355 public boolean isServerCrashProcessingEnabled() {
2356 return serverCrashProcessingEnabled.isReady();
2357 }
2358
2359 @VisibleForTesting
2360 public void setServerCrashProcessingEnabled(final boolean b) {
2361 procedureExecutor.getEnvironment().setEventReady(serverCrashProcessingEnabled, b);
2362 }
2363
2364 public ProcedureEvent getServerCrashProcessingEnabledEvent() {
2365 return serverCrashProcessingEnabled;
2366 }
2367
2368
2369
2370
2371
2372 public boolean isInitializationStartsMetaRegionAssignment() {
2373 return this.initializationBeforeMetaAssignment;
2374 }
2375
2376 public void assignRegion(HRegionInfo hri) {
2377 assignmentManager.assign(hri, true);
2378 }
2379
2380
2381
2382
2383
2384
2385
2386 public double getAverageLoad() {
2387 if (this.assignmentManager == null) {
2388 return 0;
2389 }
2390
2391 RegionStates regionStates = this.assignmentManager.getRegionStates();
2392 if (regionStates == null) {
2393 return 0;
2394 }
2395 return regionStates.getAverageLoad();
2396 }
2397
2398 @Override
2399 public boolean registerService(Service instance) {
2400
2401
2402
2403 Descriptors.ServiceDescriptor serviceDesc = instance.getDescriptorForType();
2404 if (coprocessorServiceHandlers.containsKey(serviceDesc.getFullName())) {
2405 LOG.error("Coprocessor service "+serviceDesc.getFullName()+
2406 " already registered, rejecting request from "+instance
2407 );
2408 return false;
2409 }
2410
2411 coprocessorServiceHandlers.put(serviceDesc.getFullName(), instance);
2412 if (LOG.isDebugEnabled()) {
2413 LOG.debug("Registered master coprocessor service: service="+serviceDesc.getFullName());
2414 }
2415 return true;
2416 }
2417
2418
2419
2420
2421
2422
2423
2424 public static HMaster constructMaster(Class<? extends HMaster> masterClass,
2425 final Configuration conf, final CoordinatedStateManager cp) {
2426 try {
2427 Constructor<? extends HMaster> c =
2428 masterClass.getConstructor(Configuration.class, CoordinatedStateManager.class);
2429 return c.newInstance(conf, cp);
2430 } catch(Exception e) {
2431 Throwable error = e;
2432 if (e instanceof InvocationTargetException &&
2433 ((InvocationTargetException)e).getTargetException() != null) {
2434 error = ((InvocationTargetException)e).getTargetException();
2435 }
2436 throw new RuntimeException("Failed construction of Master: " + masterClass.toString() + ". "
2437 , error);
2438 }
2439 }
2440
2441
2442
2443
2444 public static void main(String [] args) {
2445 VersionInfo.logVersion();
2446 new HMasterCommandLine(HMaster.class).doMain(args);
2447 }
2448
2449 public HFileCleaner getHFileCleaner() {
2450 return this.hfileCleaner;
2451 }
2452
2453
2454
2455
2456
2457 public SnapshotManager getSnapshotManagerForTesting() {
2458 return this.snapshotManager;
2459 }
2460
2461 @Override
2462 public void createNamespace(NamespaceDescriptor descriptor) throws IOException {
2463 TableName.isLegalNamespaceName(Bytes.toBytes(descriptor.getName()));
2464 checkNamespaceManagerReady();
2465 if (cpHost != null) {
2466 if (cpHost.preCreateNamespace(descriptor)) {
2467 return;
2468 }
2469 }
2470 LOG.info(getClientIdAuditPrefix() + " creating " + descriptor);
2471 tableNamespaceManager.create(descriptor);
2472 if (cpHost != null) {
2473 cpHost.postCreateNamespace(descriptor);
2474 }
2475 }
2476
2477 @Override
2478 public void modifyNamespace(NamespaceDescriptor descriptor) throws IOException {
2479 TableName.isLegalNamespaceName(Bytes.toBytes(descriptor.getName()));
2480 checkNamespaceManagerReady();
2481 if (cpHost != null) {
2482 if (cpHost.preModifyNamespace(descriptor)) {
2483 return;
2484 }
2485 }
2486 LOG.info(getClientIdAuditPrefix() + " modify " + descriptor);
2487 tableNamespaceManager.update(descriptor);
2488 if (cpHost != null) {
2489 cpHost.postModifyNamespace(descriptor);
2490 }
2491 }
2492
2493 @Override
2494 public void deleteNamespace(String name) throws IOException {
2495 checkNamespaceManagerReady();
2496 if (cpHost != null) {
2497 if (cpHost.preDeleteNamespace(name)) {
2498 return;
2499 }
2500 }
2501 LOG.info(getClientIdAuditPrefix() + " delete " + name);
2502 tableNamespaceManager.remove(name);
2503 if (cpHost != null) {
2504 cpHost.postDeleteNamespace(name);
2505 }
2506 }
2507
2508
2509
2510
2511
2512
2513
2514
2515 private void ensureNamespaceExists(final String name)
2516 throws IOException, NamespaceNotFoundException {
2517 checkNamespaceManagerReady();
2518 NamespaceDescriptor nsd = tableNamespaceManager.get(name);
2519 if (nsd == null) {
2520 throw new NamespaceNotFoundException(name);
2521 }
2522 }
2523
2524 @Override
2525 public NamespaceDescriptor getNamespaceDescriptor(String name) throws IOException {
2526 checkNamespaceManagerReady();
2527
2528 if (cpHost != null) {
2529 cpHost.preGetNamespaceDescriptor(name);
2530 }
2531
2532 NamespaceDescriptor nsd = tableNamespaceManager.get(name);
2533 if (nsd == null) {
2534 throw new NamespaceNotFoundException(name);
2535 }
2536
2537 if (cpHost != null) {
2538 cpHost.postGetNamespaceDescriptor(nsd);
2539 }
2540
2541 return nsd;
2542 }
2543
2544 @Override
2545 public List<NamespaceDescriptor> listNamespaceDescriptors() throws IOException {
2546 checkNamespaceManagerReady();
2547
2548 final List<NamespaceDescriptor> descriptors = new ArrayList<NamespaceDescriptor>();
2549 boolean bypass = false;
2550 if (cpHost != null) {
2551 bypass = cpHost.preListNamespaceDescriptors(descriptors);
2552 }
2553
2554 if (!bypass) {
2555 descriptors.addAll(tableNamespaceManager.list());
2556
2557 if (cpHost != null) {
2558 cpHost.postListNamespaceDescriptors(descriptors);
2559 }
2560 }
2561 return descriptors;
2562 }
2563
2564 @Override
2565 public boolean abortProcedure(final long procId, final boolean mayInterruptIfRunning)
2566 throws IOException {
2567 if (cpHost != null) {
2568 cpHost.preAbortProcedure(this.procedureExecutor, procId);
2569 }
2570
2571 final boolean result = this.procedureExecutor.abort(procId, mayInterruptIfRunning);
2572
2573 if (cpHost != null) {
2574 cpHost.postAbortProcedure();
2575 }
2576
2577 return result;
2578 }
2579
2580 @Override
2581 public List<ProcedureInfo> listProcedures() throws IOException {
2582 if (cpHost != null) {
2583 cpHost.preListProcedures();
2584 }
2585
2586 final List<ProcedureInfo> procInfoList = this.procedureExecutor.listProcedures();
2587
2588 if (cpHost != null) {
2589 cpHost.postListProcedures(procInfoList);
2590 }
2591
2592 return procInfoList;
2593 }
2594
2595 @Override
2596 public List<HTableDescriptor> listTableDescriptorsByNamespace(String name) throws IOException {
2597 ensureNamespaceExists(name);
2598 return listTableDescriptors(name, null, null, true);
2599 }
2600
2601 @Override
2602 public List<TableName> listTableNamesByNamespace(String name) throws IOException {
2603 ensureNamespaceExists(name);
2604 return listTableNames(name, null, true);
2605 }
2606
2607
2608
2609
2610
2611
2612
2613
2614
2615
2616 public List<HTableDescriptor> listTableDescriptors(final String namespace, final String regex,
2617 final List<TableName> tableNameList, final boolean includeSysTables)
2618 throws IOException {
2619 final List<HTableDescriptor> descriptors = new ArrayList<HTableDescriptor>();
2620
2621 boolean bypass = false;
2622 if (cpHost != null) {
2623 bypass = cpHost.preGetTableDescriptors(tableNameList, descriptors);
2624
2625 bypass |= cpHost.preGetTableDescriptors(tableNameList, descriptors, regex);
2626 }
2627
2628 if (!bypass) {
2629 if (tableNameList == null || tableNameList.size() == 0) {
2630
2631 Collection<HTableDescriptor> htds;
2632 if (namespace != null && namespace.length() > 0) {
2633 htds = tableDescriptors.getByNamespace(namespace).values();
2634 } else {
2635 htds = tableDescriptors.getAll().values();
2636 }
2637
2638 for (HTableDescriptor desc: htds) {
2639 if (includeSysTables || !desc.getTableName().isSystemTable()) {
2640 descriptors.add(desc);
2641 }
2642 }
2643 } else {
2644 for (TableName s: tableNameList) {
2645 HTableDescriptor desc = tableDescriptors.get(s);
2646 if (desc != null) {
2647 descriptors.add(desc);
2648 }
2649 }
2650 }
2651
2652
2653 if (regex != null) {
2654 filterTablesByRegex(descriptors, Pattern.compile(regex));
2655 }
2656
2657 if (cpHost != null) {
2658 cpHost.postGetTableDescriptors(descriptors);
2659
2660 cpHost.postGetTableDescriptors(tableNameList, descriptors, regex);
2661 }
2662 }
2663 return descriptors;
2664 }
2665
2666
2667
2668
2669
2670
2671
2672
2673 public List<TableName> listTableNames(final String namespace, final String regex,
2674 final boolean includeSysTables) throws IOException {
2675 final List<HTableDescriptor> descriptors = new ArrayList<HTableDescriptor>();
2676
2677 boolean bypass = false;
2678 if (cpHost != null) {
2679 bypass = cpHost.preGetTableNames(descriptors, regex);
2680 }
2681
2682 if (!bypass) {
2683
2684 Collection<HTableDescriptor> htds;
2685 if (namespace != null && namespace.length() > 0) {
2686 htds = tableDescriptors.getByNamespace(namespace).values();
2687 } else {
2688 htds = tableDescriptors.getAll().values();
2689 }
2690
2691 for (HTableDescriptor htd: htds) {
2692 if (includeSysTables || !htd.getTableName().isSystemTable()) {
2693 descriptors.add(htd);
2694 }
2695 }
2696
2697
2698 if (regex != null) {
2699 filterTablesByRegex(descriptors, Pattern.compile(regex));
2700 }
2701
2702 if (cpHost != null) {
2703 cpHost.postGetTableNames(descriptors, regex);
2704 }
2705 }
2706
2707 List<TableName> result = new ArrayList<TableName>(descriptors.size());
2708 for (HTableDescriptor htd: descriptors) {
2709 result.add(htd.getTableName());
2710 }
2711 return result;
2712 }
2713
2714
2715
2716
2717
2718
2719
2720 private static void filterTablesByRegex(final Collection<HTableDescriptor> descriptors,
2721 final Pattern pattern) {
2722 final String defaultNS = NamespaceDescriptor.DEFAULT_NAMESPACE_NAME_STR;
2723 Iterator<HTableDescriptor> itr = descriptors.iterator();
2724 while (itr.hasNext()) {
2725 HTableDescriptor htd = itr.next();
2726 String tableName = htd.getTableName().getNameAsString();
2727 boolean matched = pattern.matcher(tableName).matches();
2728 if (!matched && htd.getTableName().getNamespaceAsString().equals(defaultNS)) {
2729 matched = pattern.matcher(defaultNS + TableName.NAMESPACE_DELIM + tableName).matches();
2730 }
2731 if (!matched) {
2732 itr.remove();
2733 }
2734 }
2735 }
2736
2737 @Override
2738 public long getLastMajorCompactionTimestamp(TableName table) throws IOException {
2739 return getClusterStatus().getLastMajorCompactionTsForTable(table);
2740 }
2741
2742 @Override
2743 public long getLastMajorCompactionTimestampForRegion(byte[] regionName) throws IOException {
2744 return getClusterStatus().getLastMajorCompactionTsForRegion(regionName);
2745 }
2746
2747
2748
2749
2750
2751
2752
2753 public boolean isBalancerOn() {
2754 if (null == loadBalancerTracker) return false;
2755 return loadBalancerTracker.isBalancerOn();
2756 }
2757
2758
2759
2760
2761
2762 public boolean isNormalizerOn() {
2763 if (null == regionNormalizerTracker) {
2764 return false;
2765 }
2766 return regionNormalizerTracker.isNormalizerOn();
2767 }
2768
2769
2770
2771
2772
2773
2774 public String getLoadBalancerClassName() {
2775 return conf.get(HConstants.HBASE_MASTER_LOADBALANCER_CLASS, LoadBalancerFactory
2776 .getDefaultLoadBalancerClass().getName());
2777 }
2778
2779
2780
2781
2782 public RegionNormalizerTracker getRegionNormalizerTracker() {
2783 return regionNormalizerTracker;
2784 }
2785
2786
2787
2788
2789
2790
2791
2792
2793
2794
2795 public CompactionState getMobCompactionState(TableName tableName) {
2796 AtomicInteger compactionsCount = mobFileCompactionStates.get(tableName);
2797 if (compactionsCount != null && compactionsCount.get() != 0) {
2798 return CompactionState.MAJOR_AND_MINOR;
2799 }
2800 return CompactionState.NONE;
2801 }
2802
2803 public void reportMobFileCompactionStart(TableName tableName) throws IOException {
2804 IdLock.Entry lockEntry = null;
2805 try {
2806 lockEntry = mobFileCompactionLock.getLockEntry(tableName.hashCode());
2807 AtomicInteger compactionsCount = mobFileCompactionStates.get(tableName);
2808 if (compactionsCount == null) {
2809 compactionsCount = new AtomicInteger(0);
2810 mobFileCompactionStates.put(tableName, compactionsCount);
2811 }
2812 compactionsCount.incrementAndGet();
2813 } finally {
2814 if (lockEntry != null) {
2815 mobFileCompactionLock.releaseLockEntry(lockEntry);
2816 }
2817 }
2818 }
2819
2820 public void reportMobFileCompactionEnd(TableName tableName) throws IOException {
2821 IdLock.Entry lockEntry = null;
2822 try {
2823 lockEntry = mobFileCompactionLock.getLockEntry(tableName.hashCode());
2824 AtomicInteger compactionsCount = mobFileCompactionStates.get(tableName);
2825 if (compactionsCount != null) {
2826 int count = compactionsCount.decrementAndGet();
2827
2828 if (count == 0) {
2829 mobFileCompactionStates.remove(tableName);
2830 }
2831 }
2832 } finally {
2833 if (lockEntry != null) {
2834 mobFileCompactionLock.releaseLockEntry(lockEntry);
2835 }
2836 }
2837 }
2838 }