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.util.ArrayList;
24 import java.util.Arrays;
25 import java.util.Collection;
26 import java.util.Collections;
27 import java.util.HashMap;
28 import java.util.HashSet;
29 import java.util.Iterator;
30 import java.util.List;
31 import java.util.Map;
32 import java.util.NavigableMap;
33 import java.util.Random;
34 import java.util.Set;
35 import java.util.TreeMap;
36 import java.util.concurrent.Callable;
37 import java.util.concurrent.ConcurrentHashMap;
38 import java.util.concurrent.CopyOnWriteArrayList;
39 import java.util.concurrent.ThreadFactory;
40 import java.util.concurrent.TimeUnit;
41 import java.util.concurrent.atomic.AtomicBoolean;
42 import java.util.concurrent.atomic.AtomicInteger;
43 import java.util.concurrent.locks.Lock;
44 import java.util.concurrent.locks.ReentrantLock;
45
46 import org.apache.commons.logging.Log;
47 import org.apache.commons.logging.LogFactory;
48 import org.apache.hadoop.conf.Configuration;
49 import org.apache.hadoop.fs.FileSystem;
50 import org.apache.hadoop.fs.Path;
51 import org.apache.hadoop.hbase.CoordinatedStateException;
52 import org.apache.hadoop.hbase.HBaseIOException;
53 import org.apache.hadoop.hbase.HConstants;
54 import org.apache.hadoop.hbase.HRegionInfo;
55 import org.apache.hadoop.hbase.HRegionLocation;
56 import org.apache.hadoop.hbase.HTableDescriptor;
57 import org.apache.hadoop.hbase.MetaTableAccessor;
58 import org.apache.hadoop.hbase.NotServingRegionException;
59 import org.apache.hadoop.hbase.RegionLocations;
60 import org.apache.hadoop.hbase.RegionStateListener;
61 import org.apache.hadoop.hbase.RegionTransition;
62 import org.apache.hadoop.hbase.ServerName;
63 import org.apache.hadoop.hbase.TableName;
64 import org.apache.hadoop.hbase.TableNotFoundException;
65 import org.apache.hadoop.hbase.TableStateManager;
66 import org.apache.hadoop.hbase.classification.InterfaceAudience;
67 import org.apache.hadoop.hbase.client.Admin;
68 import org.apache.hadoop.hbase.client.RegionReplicaUtil;
69 import org.apache.hadoop.hbase.client.Result;
70 import org.apache.hadoop.hbase.coordination.BaseCoordinatedStateManager;
71 import org.apache.hadoop.hbase.coordination.OpenRegionCoordination;
72 import org.apache.hadoop.hbase.coordination.RegionMergeCoordination;
73 import org.apache.hadoop.hbase.coordination.SplitTransactionCoordination.SplitTransactionDetails;
74 import org.apache.hadoop.hbase.coordination.ZkOpenRegionCoordination;
75 import org.apache.hadoop.hbase.coordination.ZkRegionMergeCoordination;
76 import org.apache.hadoop.hbase.exceptions.DeserializationException;
77 import org.apache.hadoop.hbase.executor.EventHandler;
78 import org.apache.hadoop.hbase.executor.EventType;
79 import org.apache.hadoop.hbase.executor.ExecutorService;
80 import org.apache.hadoop.hbase.ipc.FailedServerException;
81 import org.apache.hadoop.hbase.ipc.RpcClient;
82 import org.apache.hadoop.hbase.ipc.ServerNotRunningYetException;
83 import org.apache.hadoop.hbase.master.RegionState.State;
84 import org.apache.hadoop.hbase.master.balancer.FavoredNodeAssignmentHelper;
85 import org.apache.hadoop.hbase.master.balancer.FavoredNodeLoadBalancer;
86 import org.apache.hadoop.hbase.master.handler.ClosedRegionHandler;
87 import org.apache.hadoop.hbase.master.handler.DisableTableHandler;
88 import org.apache.hadoop.hbase.master.handler.EnableTableHandler;
89 import org.apache.hadoop.hbase.master.handler.OpenedRegionHandler;
90 import org.apache.hadoop.hbase.protobuf.generated.RegionServerStatusProtos.RegionStateTransition;
91 import org.apache.hadoop.hbase.protobuf.generated.RegionServerStatusProtos.RegionStateTransition.TransitionCode;
92 import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos;
93 import org.apache.hadoop.hbase.regionserver.RegionAlreadyInTransitionException;
94 import org.apache.hadoop.hbase.regionserver.RegionOpeningState;
95 import org.apache.hadoop.hbase.regionserver.RegionServerAbortedException;
96 import org.apache.hadoop.hbase.regionserver.RegionServerStoppedException;
97 import org.apache.hadoop.hbase.util.ConfigUtil;
98 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
99 import org.apache.hadoop.hbase.util.FSUtils;
100 import org.apache.hadoop.hbase.util.KeyLocker;
101 import org.apache.hadoop.hbase.util.Pair;
102 import org.apache.hadoop.hbase.util.PairOfSameType;
103 import org.apache.hadoop.hbase.util.Threads;
104 import org.apache.hadoop.hbase.util.Triple;
105 import org.apache.hadoop.hbase.wal.DefaultWALProvider;
106 import org.apache.hadoop.hbase.zookeeper.MetaTableLocator;
107 import org.apache.hadoop.hbase.zookeeper.ZKAssign;
108 import org.apache.hadoop.hbase.zookeeper.ZKUtil;
109 import org.apache.hadoop.hbase.zookeeper.ZooKeeperListener;
110 import org.apache.hadoop.ipc.RemoteException;
111 import org.apache.hadoop.util.StringUtils;
112 import org.apache.zookeeper.AsyncCallback;
113 import org.apache.zookeeper.KeeperException;
114 import org.apache.zookeeper.KeeperException.NoNodeException;
115 import org.apache.zookeeper.KeeperException.NodeExistsException;
116 import org.apache.zookeeper.data.Stat;
117
118 import com.google.common.annotations.VisibleForTesting;
119 import com.google.common.collect.LinkedHashMultimap;
120
121
122
123
124
125
126
127
128 @InterfaceAudience.Private
129 public class AssignmentManager extends ZooKeeperListener {
130 private static final Log LOG = LogFactory.getLog(AssignmentManager.class);
131
132 public static final ServerName HBCK_CODE_SERVERNAME = ServerName.valueOf(HConstants.HBCK_CODE_NAME,
133 -1, -1L);
134
135 static final String ALREADY_IN_TRANSITION_WAITTIME
136 = "hbase.assignment.already.intransition.waittime";
137 static final int DEFAULT_ALREADY_IN_TRANSITION_WAITTIME = 60000;
138
139 protected final MasterServices server;
140
141 private ServerManager serverManager;
142
143 private boolean shouldAssignRegionsWithFavoredNodes;
144
145 private LoadBalancer balancer;
146
147 private final MetricsAssignmentManager metricsAssignmentManager;
148
149 private final TableLockManager tableLockManager;
150
151 private AtomicInteger numRegionsOpened = new AtomicInteger(0);
152
153 final private KeyLocker<String> locker = new KeyLocker<String>();
154
155 Set<HRegionInfo> replicasToClose = Collections.synchronizedSet(new HashSet<HRegionInfo>());
156
157
158
159
160
161 private final Map <String, HRegionInfo> regionsToReopen;
162
163
164
165
166
167 private final int maximumAttempts;
168
169
170
171
172 private final Map<String, PairOfSameType<HRegionInfo>> mergingRegions
173 = new HashMap<String, PairOfSameType<HRegionInfo>>();
174
175 private final Map<HRegionInfo, PairOfSameType<HRegionInfo>> splitRegions
176 = new HashMap<HRegionInfo, PairOfSameType<HRegionInfo>>();
177
178
179
180
181
182 private final long sleepTimeBeforeRetryingMetaAssignment;
183
184
185
186
187
188 final NavigableMap<String, RegionPlan> regionPlans =
189 new TreeMap<String, RegionPlan>();
190
191 private final TableStateManager tableStateManager;
192
193 private final ExecutorService executorService;
194
195
196 private Map<HRegionInfo, AtomicBoolean> closedRegionHandlerCalled = null;
197
198
199 private Map<HRegionInfo, AtomicBoolean> openedRegionHandlerCalled = null;
200
201
202 private java.util.concurrent.ExecutorService threadPoolExecutorService;
203
204
205 private final java.util.concurrent.ExecutorService zkEventWorkers;
206
207 private List<EventType> ignoreStatesRSOffline = Arrays.asList(
208 EventType.RS_ZK_REGION_FAILED_OPEN, EventType.RS_ZK_REGION_CLOSED);
209
210 private final RegionStates regionStates;
211
212
213
214
215
216 private final int bulkAssignThresholdRegions;
217 private final int bulkAssignThresholdServers;
218 private final int bulkPerRegionOpenTimeGuesstimate;
219
220
221
222
223 private final boolean bulkAssignWaitTillAllAssigned;
224
225
226
227
228
229
230
231
232
233 protected final AtomicBoolean failoverCleanupDone = new AtomicBoolean(false);
234
235
236
237
238
239
240
241
242 private final ConcurrentHashMap<String, AtomicInteger>
243 failedOpenTracker = new ConcurrentHashMap<String, AtomicInteger>();
244
245
246 private final boolean useZKForAssignment;
247
248
249
250 private final RegionStateStore regionStateStore;
251
252
253
254
255 @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="MS_SHOULD_BE_FINAL")
256 public static boolean TEST_SKIP_SPLIT_HANDLING = false;
257
258
259 private List<AssignmentListener> listeners = new CopyOnWriteArrayList<AssignmentListener>();
260
261 private RegionStateListener regionStateListener;
262
263 public enum ServerHostRegion {
264 NOT_HOSTING_REGION, HOSTING_REGION, UNKNOWN,
265 }
266
267
268
269
270
271
272
273
274
275
276
277
278
279 public AssignmentManager(MasterServices server, ServerManager serverManager,
280 final LoadBalancer balancer,
281 final ExecutorService service, MetricsMaster metricsMaster,
282 final TableLockManager tableLockManager) throws KeeperException,
283 IOException, CoordinatedStateException {
284 super(server.getZooKeeper());
285 this.server = server;
286 this.serverManager = serverManager;
287 this.executorService = service;
288 this.regionStateStore = new RegionStateStore(server);
289 this.regionsToReopen = Collections.synchronizedMap
290 (new HashMap<String, HRegionInfo> ());
291 Configuration conf = server.getConfiguration();
292
293 this.shouldAssignRegionsWithFavoredNodes = conf.getClass(
294 HConstants.HBASE_MASTER_LOADBALANCER_CLASS, Object.class).equals(
295 FavoredNodeLoadBalancer.class);
296 try {
297 if (server.getCoordinatedStateManager() != null) {
298 this.tableStateManager = server.getCoordinatedStateManager().getTableStateManager();
299 } else {
300 this.tableStateManager = null;
301 }
302 } catch (InterruptedException e) {
303 throw new InterruptedIOException();
304 }
305
306 this.maximumAttempts = Math.max(1,
307 this.server.getConfiguration().getInt("hbase.assignment.maximum.attempts", 10));
308 this.sleepTimeBeforeRetryingMetaAssignment = this.server.getConfiguration().getLong(
309 "hbase.meta.assignment.retry.sleeptime", 1000l);
310 this.balancer = balancer;
311 int maxThreads = conf.getInt("hbase.assignment.threads.max", 30);
312 this.threadPoolExecutorService = Threads.getBoundedCachedThreadPool(
313 maxThreads, 60L, TimeUnit.SECONDS, Threads.newDaemonThreadFactory("AM."));
314 this.regionStates = new RegionStates(
315 server, tableStateManager, serverManager, regionStateStore);
316
317 this.bulkAssignWaitTillAllAssigned =
318 conf.getBoolean("hbase.bulk.assignment.waittillallassigned", false);
319 this.bulkAssignThresholdRegions = conf.getInt("hbase.bulk.assignment.threshold.regions", 7);
320 this.bulkAssignThresholdServers = conf.getInt("hbase.bulk.assignment.threshold.servers", 3);
321 this.bulkPerRegionOpenTimeGuesstimate =
322 conf.getInt("hbase.bulk.assignment.perregion.open.time", 10000);
323
324 int workers = conf.getInt("hbase.assignment.zkevent.workers", 20);
325 ThreadFactory threadFactory = Threads.newDaemonThreadFactory("AM.ZK.Worker");
326 zkEventWorkers = Threads.getBoundedCachedThreadPool(workers, 60L,
327 TimeUnit.SECONDS, threadFactory);
328 this.tableLockManager = tableLockManager;
329
330 this.metricsAssignmentManager = new MetricsAssignmentManager();
331 useZKForAssignment = ConfigUtil.useZKForAssignment(conf);
332 }
333
334 MetricsAssignmentManager getAssignmentManagerMetrics() {
335 return this.metricsAssignmentManager;
336 }
337
338
339
340
341
342 public void registerListener(final AssignmentListener listener) {
343 this.listeners.add(listener);
344 }
345
346
347
348
349
350 public boolean unregisterListener(final AssignmentListener listener) {
351 return this.listeners.remove(listener);
352 }
353
354
355
356
357 public TableStateManager getTableStateManager() {
358
359
360 return this.tableStateManager;
361 }
362
363
364
365
366
367
368
369 public RegionStates getRegionStates() {
370 return regionStates;
371 }
372
373
374
375
376 @VisibleForTesting
377 RegionStateStore getRegionStateStore() {
378 return regionStateStore;
379 }
380
381 public RegionPlan getRegionReopenPlan(HRegionInfo hri) {
382 return new RegionPlan(hri, null, regionStates.getRegionServerOfRegion(hri));
383 }
384
385
386
387
388
389
390 public void addPlan(String encodedName, RegionPlan plan) {
391 synchronized (regionPlans) {
392 regionPlans.put(encodedName, plan);
393 }
394 }
395
396
397
398
399 public void addPlans(Map<String, RegionPlan> plans) {
400 synchronized (regionPlans) {
401 regionPlans.putAll(plans);
402 }
403 }
404
405
406
407
408
409
410
411
412 public void setRegionsToReopen(List <HRegionInfo> regions) {
413 for(HRegionInfo hri : regions) {
414 regionsToReopen.put(hri.getEncodedName(), hri);
415 }
416 }
417
418
419
420
421
422
423
424
425 public Pair<Integer, Integer> getReopenStatus(TableName tableName)
426 throws IOException {
427 List<HRegionInfo> hris;
428 if (TableName.META_TABLE_NAME.equals(tableName)) {
429 hris = new MetaTableLocator().getMetaRegions(server.getZooKeeper());
430 } else {
431 hris = MetaTableAccessor.getTableRegions(server.getZooKeeper(),
432 server.getConnection(), tableName, true);
433 }
434
435 Integer pending = 0;
436 for (HRegionInfo hri : hris) {
437 String name = hri.getEncodedName();
438
439 if (regionsToReopen.containsKey(name)
440 || regionStates.isRegionInTransition(name)) {
441 pending++;
442 }
443 }
444 return new Pair<Integer, Integer>(pending, hris.size());
445 }
446
447
448
449
450
451
452 public boolean isFailoverCleanupDone() {
453 return failoverCleanupDone.get();
454 }
455
456
457
458
459
460 public Lock acquireRegionLock(final String encodedName) {
461 return locker.acquireLock(encodedName);
462 }
463
464
465
466
467
468 void failoverCleanupDone() {
469 failoverCleanupDone.set(true);
470 serverManager.processQueuedDeadServers();
471 }
472
473
474
475
476
477
478
479
480
481 void joinCluster() throws IOException,
482 KeeperException, InterruptedException, CoordinatedStateException {
483 long startTime = System.currentTimeMillis();
484
485
486
487
488
489
490
491
492
493
494
495 Set<ServerName> deadServers = rebuildUserRegions();
496
497
498
499 boolean failover = processDeadServersAndRegionsInTransition(deadServers);
500
501 if (!useZKForAssignment) {
502
503 ZKUtil.deleteNodeRecursively(watcher, watcher.assignmentZNode);
504 }
505 recoverTableInDisablingState();
506 recoverTableInEnablingState();
507 LOG.info("Joined the cluster in " + (System.currentTimeMillis()
508 - startTime) + "ms, failover=" + failover);
509 }
510
511
512
513
514
515
516
517
518
519
520
521
522 boolean processDeadServersAndRegionsInTransition(final Set<ServerName> deadServers)
523 throws KeeperException, IOException, InterruptedException, CoordinatedStateException {
524 List<String> nodes = ZKUtil.listChildrenNoWatch(watcher, watcher.assignmentZNode);
525
526 if (useZKForAssignment && nodes == null) {
527 String errorMessage = "Failed to get the children from ZK";
528 server.abort(errorMessage, new IOException(errorMessage));
529 return true;
530 }
531
532 boolean failover = !serverManager.getDeadServers().isEmpty();
533 if (failover) {
534
535 if (LOG.isDebugEnabled()) {
536 LOG.debug("Found dead servers out on cluster " + serverManager.getDeadServers());
537 }
538 } else {
539
540 Set<ServerName> onlineServers = serverManager.getOnlineServers().keySet();
541 for (Map.Entry<HRegionInfo, ServerName> en:
542 regionStates.getRegionAssignments().entrySet()) {
543 HRegionInfo hri = en.getKey();
544 if (!hri.isMetaTable()
545 && onlineServers.contains(en.getValue())) {
546 LOG.debug("Found " + hri + " out on cluster");
547 failover = true;
548 break;
549 }
550 }
551 if (!failover && nodes != null) {
552
553 for (String encodedName: nodes) {
554 RegionState regionState = regionStates.getRegionState(encodedName);
555 if (regionState != null && !regionState.getRegion().isMetaRegion()) {
556 LOG.debug("Found " + regionState + " in RITs");
557 failover = true;
558 break;
559 }
560 }
561 }
562 }
563 if (!failover && !useZKForAssignment) {
564
565 Map<String, RegionState> regionsInTransition = regionStates.getRegionsInTransition();
566 if (!regionsInTransition.isEmpty()) {
567 Set<ServerName> onlineServers = serverManager.getOnlineServers().keySet();
568 for (RegionState regionState: regionsInTransition.values()) {
569 ServerName serverName = regionState.getServerName();
570 if (!regionState.getRegion().isMetaRegion()
571 && serverName != null && onlineServers.contains(serverName)) {
572 LOG.debug("Found " + regionState + " in RITs");
573 failover = true;
574 break;
575 }
576 }
577 }
578 }
579 if (!failover) {
580
581
582
583
584 Set<ServerName> queuedDeadServers = serverManager.getRequeuedDeadServers().keySet();
585 if (!queuedDeadServers.isEmpty()) {
586 Configuration conf = server.getConfiguration();
587 Path rootdir = FSUtils.getRootDir(conf);
588 FileSystem fs = rootdir.getFileSystem(conf);
589 for (ServerName serverName: queuedDeadServers) {
590
591
592 Path logDir = new Path(rootdir,
593 DefaultWALProvider.getWALDirectoryName(serverName.toString()));
594 Path splitDir = logDir.suffix(DefaultWALProvider.SPLITTING_EXT);
595 if (fs.exists(logDir) || fs.exists(splitDir)) {
596 LOG.debug("Found queued dead server " + serverName);
597 failover = true;
598 break;
599 }
600 }
601 if (!failover) {
602
603
604 LOG.info("AM figured that it's not a failover and cleaned up "
605 + queuedDeadServers.size() + " queued dead servers");
606 serverManager.removeRequeuedDeadServers();
607 }
608 }
609 }
610
611 Set<TableName> disabledOrDisablingOrEnabling = null;
612 Map<HRegionInfo, ServerName> allRegions = null;
613
614 if (!failover) {
615 disabledOrDisablingOrEnabling = tableStateManager.getTablesInStates(
616 ZooKeeperProtos.Table.State.DISABLED, ZooKeeperProtos.Table.State.DISABLING,
617 ZooKeeperProtos.Table.State.ENABLING);
618
619
620 allRegions = regionStates.closeAllUserRegions(
621 disabledOrDisablingOrEnabling);
622 }
623
624
625 regionStateStore.start();
626
627
628 if (failover) {
629 LOG.info("Found regions out on cluster or in RIT; presuming failover");
630
631
632 processDeadServersAndRecoverLostRegions(deadServers);
633 }
634
635 if (!failover && useZKForAssignment) {
636
637 ZKAssign.deleteAllNodes(watcher);
638 ZKUtil.listChildrenAndWatchForNewChildren(this.watcher,
639 this.watcher.assignmentZNode);
640 }
641
642
643
644
645
646 failoverCleanupDone();
647 if (!failover) {
648
649 LOG.info("Clean cluster startup. Assigning user regions");
650 assignAllUserRegions(allRegions);
651 }
652
653
654
655 for (HRegionInfo h : replicasToClose) {
656 unassign(h);
657 }
658 replicasToClose.clear();
659 return failover;
660 }
661
662
663
664
665
666
667
668
669
670
671
672
673 boolean processRegionInTransitionAndBlockUntilAssigned(final HRegionInfo hri)
674 throws InterruptedException, KeeperException, IOException {
675 String encodedRegionName = hri.getEncodedName();
676 if (!processRegionInTransition(encodedRegionName, hri)) {
677 return false;
678 }
679 LOG.debug("Waiting on " + HRegionInfo.prettyPrint(encodedRegionName));
680 while (!this.server.isStopped() &&
681 this.regionStates.isRegionInTransition(encodedRegionName)) {
682 RegionState state = this.regionStates.getRegionTransitionState(encodedRegionName);
683 if (state == null || !serverManager.isServerOnline(state.getServerName())) {
684
685
686
687 break;
688 }
689 this.regionStates.waitForUpdate(100);
690 }
691 return true;
692 }
693
694
695
696
697
698
699
700
701
702
703 boolean processRegionInTransition(final String encodedRegionName,
704 final HRegionInfo regionInfo) throws KeeperException, IOException {
705
706
707
708
709 Lock lock = locker.acquireLock(encodedRegionName);
710 try {
711 Stat stat = new Stat();
712 byte [] data = ZKAssign.getDataAndWatch(watcher, encodedRegionName, stat);
713 if (data == null) return false;
714 RegionTransition rt;
715 try {
716 rt = RegionTransition.parseFrom(data);
717 } catch (DeserializationException e) {
718 LOG.warn("Failed parse znode data", e);
719 return false;
720 }
721 HRegionInfo hri = regionInfo;
722 if (hri == null) {
723
724
725
726
727
728 hri = regionStates.getRegionInfo(rt.getRegionName());
729 EventType et = rt.getEventType();
730 if (hri == null && et != EventType.RS_ZK_REGION_MERGING
731 && et != EventType.RS_ZK_REQUEST_REGION_MERGE) {
732 LOG.warn("Couldn't find the region in recovering " + rt);
733 return false;
734 }
735 }
736
737
738
739 BaseCoordinatedStateManager cp =
740 (BaseCoordinatedStateManager) this.server.getCoordinatedStateManager();
741 OpenRegionCoordination openRegionCoordination = cp.getOpenRegionCoordination();
742
743 ZkOpenRegionCoordination.ZkOpenRegionDetails zkOrd =
744 new ZkOpenRegionCoordination.ZkOpenRegionDetails();
745 zkOrd.setVersion(stat.getVersion());
746 zkOrd.setServerName(cp.getServer().getServerName());
747
748 return processRegionsInTransition(
749 rt, hri, openRegionCoordination, zkOrd);
750 } finally {
751 lock.unlock();
752 }
753 }
754
755
756
757
758
759
760
761
762
763 boolean processRegionsInTransition(
764 final RegionTransition rt, final HRegionInfo regionInfo,
765 OpenRegionCoordination coordination,
766 final OpenRegionCoordination.OpenRegionDetails ord) throws KeeperException {
767 EventType et = rt.getEventType();
768
769 final ServerName sn = rt.getServerName();
770 final byte[] regionName = rt.getRegionName();
771 final String encodedName = HRegionInfo.encodeRegionName(regionName);
772 final String prettyPrintedRegionName = HRegionInfo.prettyPrint(encodedName);
773 LOG.info("Processing " + prettyPrintedRegionName + " in state: " + et);
774
775 if (regionStates.isRegionInTransition(encodedName)
776 && (regionInfo.isMetaRegion() || !useZKForAssignment)) {
777 LOG.info("Processed region " + prettyPrintedRegionName + " in state: "
778 + et + ", does nothing since the region is already in transition "
779 + regionStates.getRegionTransitionState(encodedName));
780
781 return true;
782 }
783 if (!serverManager.isServerOnline(sn)) {
784
785
786
787 LOG.debug("RIT " + encodedName + " in state=" + rt.getEventType() +
788 " was on deadserver; forcing offline");
789 if (regionStates.isRegionOnline(regionInfo)) {
790
791
792
793 regionStates.regionOffline(regionInfo);
794 sendRegionClosedNotification(regionInfo);
795 }
796
797 regionStates.updateRegionState(regionInfo, State.OFFLINE, sn);
798
799 if (regionInfo.isMetaRegion()) {
800
801
802 MetaTableLocator.setMetaLocation(watcher, sn, State.OPEN);
803 } else {
804
805
806 regionStates.setLastRegionServerOfRegion(sn, encodedName);
807
808 if (!serverManager.isServerDead(sn)) {
809 serverManager.expireServer(sn);
810 }
811 }
812 return false;
813 }
814 switch (et) {
815 case M_ZK_REGION_CLOSING:
816
817
818 final RegionState rsClosing = regionStates.updateRegionState(rt, State.CLOSING);
819 this.executorService.submit(
820 new EventHandler(server, EventType.M_MASTER_RECOVERY) {
821 @Override
822 public void process() throws IOException {
823 ReentrantLock lock = locker.acquireLock(regionInfo.getEncodedName());
824 try {
825 final int expectedVersion = ((ZkOpenRegionCoordination.ZkOpenRegionDetails) ord)
826 .getVersion();
827 unassign(regionInfo, rsClosing, expectedVersion, null, useZKForAssignment, null);
828 if (regionStates.isRegionOffline(regionInfo)) {
829 assign(regionInfo, true);
830 }
831 } finally {
832 lock.unlock();
833 }
834 }
835 });
836 break;
837
838 case RS_ZK_REGION_CLOSED:
839 case RS_ZK_REGION_FAILED_OPEN:
840
841 regionStates.setRegionStateTOCLOSED(regionInfo, sn);
842 if (!replicasToClose.contains(regionInfo)) {
843 invokeAssign(regionInfo);
844 } else {
845 offlineDisabledRegion(regionInfo);
846 }
847 break;
848
849 case M_ZK_REGION_OFFLINE:
850
851 regionStates.updateRegionState(rt, State.OFFLINE);
852 final RegionState rsOffline = regionStates.getRegionState(regionInfo);
853 this.executorService.submit(
854 new EventHandler(server, EventType.M_MASTER_RECOVERY) {
855 @Override
856 public void process() throws IOException {
857 ReentrantLock lock = locker.acquireLock(regionInfo.getEncodedName());
858 try {
859 RegionPlan plan = new RegionPlan(regionInfo, null, sn);
860 addPlan(encodedName, plan);
861 assign(rsOffline, true, false);
862 } finally {
863 lock.unlock();
864 }
865 }
866 });
867 break;
868
869 case RS_ZK_REGION_OPENING:
870 regionStates.updateRegionState(rt, State.OPENING);
871 break;
872
873 case RS_ZK_REGION_OPENED:
874
875
876
877 regionStates.updateRegionState(rt, State.OPEN);
878 new OpenedRegionHandler(server, this, regionInfo, coordination, ord).process();
879 break;
880 case RS_ZK_REQUEST_REGION_SPLIT:
881 case RS_ZK_REGION_SPLITTING:
882 case RS_ZK_REGION_SPLIT:
883
884
885
886 regionStates.regionOnline(regionInfo, sn);
887 regionStates.updateRegionState(rt, State.SPLITTING);
888 if (!handleRegionSplitting(
889 rt, encodedName, prettyPrintedRegionName, sn)) {
890 deleteSplittingNode(encodedName, sn);
891 }
892 break;
893 case RS_ZK_REQUEST_REGION_MERGE:
894 case RS_ZK_REGION_MERGING:
895 case RS_ZK_REGION_MERGED:
896 if (!handleRegionMerging(
897 rt, encodedName, prettyPrintedRegionName, sn)) {
898 deleteMergingNode(encodedName, sn);
899 }
900 break;
901 default:
902 throw new IllegalStateException("Received region in state:" + et + " is not valid.");
903 }
904 LOG.info("Processed region " + prettyPrintedRegionName + " in state "
905 + et + ", on " + (serverManager.isServerOnline(sn) ? "" : "dead ")
906 + "server: " + sn);
907 return true;
908 }
909
910
911
912
913
914 public void removeClosedRegion(HRegionInfo hri) {
915 if (regionsToReopen.remove(hri.getEncodedName()) != null) {
916 LOG.debug("Removed region from reopening regions because it was closed");
917 }
918 }
919
920
921
922
923
924
925
926
927
928
929
930
931 @edu.umd.cs.findbugs.annotations.SuppressWarnings(
932 value="AT_OPERATION_SEQUENCE_ON_CONCURRENT_ABSTRACTION",
933 justification="Needs work; says access to ConcurrentHashMaps not ATOMIC!!!")
934 void handleRegion(final RegionTransition rt, OpenRegionCoordination coordination,
935 OpenRegionCoordination.OpenRegionDetails ord) {
936 if (rt == null) {
937 LOG.warn("Unexpected NULL input for RegionTransition rt");
938 return;
939 }
940 final ServerName sn = rt.getServerName();
941
942 if (sn.equals(HBCK_CODE_SERVERNAME)) {
943 handleHBCK(rt);
944 return;
945 }
946 final long createTime = rt.getCreateTime();
947 final byte[] regionName = rt.getRegionName();
948 String encodedName = HRegionInfo.encodeRegionName(regionName);
949 String prettyPrintedRegionName = HRegionInfo.prettyPrint(encodedName);
950
951 if (!serverManager.isServerOnline(sn)
952 && !ignoreStatesRSOffline.contains(rt.getEventType())) {
953 LOG.warn("Attempted to handle region transition for server but " +
954 "it is not online: " + prettyPrintedRegionName + ", " + rt);
955 return;
956 }
957
958 RegionState regionState =
959 regionStates.getRegionState(encodedName);
960 long startTime = System.currentTimeMillis();
961 if (LOG.isDebugEnabled()) {
962 boolean lateEvent = createTime < (startTime - 15000);
963 LOG.debug("Handling " + rt.getEventType() +
964 ", server=" + sn + ", region=" +
965 (prettyPrintedRegionName == null ? "null" : prettyPrintedRegionName) +
966 (lateEvent ? ", which is more than 15 seconds late" : "") +
967 ", current_state=" + regionState);
968 }
969
970
971 if (rt.getEventType() == EventType.M_ZK_REGION_OFFLINE) {
972 return;
973 }
974
975
976 Lock lock = locker.acquireLock(encodedName);
977 try {
978 RegionState latestState =
979 regionStates.getRegionState(encodedName);
980 if ((regionState == null && latestState != null)
981 || (regionState != null && latestState == null)
982 || (regionState != null && latestState != null
983 && latestState.getState() != regionState.getState())) {
984 LOG.warn("Region state changed from " + regionState + " to "
985 + latestState + ", while acquiring lock");
986 }
987 long waitedTime = System.currentTimeMillis() - startTime;
988 if (waitedTime > 5000) {
989 LOG.warn("Took " + waitedTime + "ms to acquire the lock");
990 }
991 regionState = latestState;
992 switch (rt.getEventType()) {
993 case RS_ZK_REQUEST_REGION_SPLIT:
994 case RS_ZK_REGION_SPLITTING:
995 case RS_ZK_REGION_SPLIT:
996 if (!handleRegionSplitting(
997 rt, encodedName, prettyPrintedRegionName, sn)) {
998 deleteSplittingNode(encodedName, sn);
999 }
1000 break;
1001
1002 case RS_ZK_REQUEST_REGION_MERGE:
1003 case RS_ZK_REGION_MERGING:
1004 case RS_ZK_REGION_MERGED:
1005
1006
1007 if (!handleRegionMerging(
1008 rt, encodedName, prettyPrintedRegionName, sn)) {
1009 deleteMergingNode(encodedName, sn);
1010 }
1011 break;
1012
1013 case M_ZK_REGION_CLOSING:
1014
1015
1016 if (regionState == null
1017 || !regionState.isPendingCloseOrClosingOnServer(sn)) {
1018 LOG.warn("Received CLOSING for " + prettyPrintedRegionName
1019 + " from " + sn + " but the region isn't PENDING_CLOSE/CLOSING here: "
1020 + regionStates.getRegionState(encodedName));
1021 return;
1022 }
1023
1024 regionStates.updateRegionState(rt, State.CLOSING);
1025 break;
1026
1027 case RS_ZK_REGION_CLOSED:
1028
1029 if (regionState == null
1030 || !regionState.isPendingCloseOrClosingOnServer(sn)) {
1031 LOG.warn("Received CLOSED for " + prettyPrintedRegionName
1032 + " from " + sn + " but the region isn't PENDING_CLOSE/CLOSING here: "
1033 + regionStates.getRegionState(encodedName));
1034 return;
1035 }
1036
1037
1038
1039 new ClosedRegionHandler(server, this, regionState.getRegion()).process();
1040 updateClosedRegionHandlerTracker(regionState.getRegion());
1041 break;
1042
1043 case RS_ZK_REGION_FAILED_OPEN:
1044 if (regionState == null
1045 || !regionState.isPendingOpenOrOpeningOnServer(sn)) {
1046 LOG.warn("Received FAILED_OPEN for " + prettyPrintedRegionName
1047 + " from " + sn + " but the region isn't PENDING_OPEN/OPENING here: "
1048 + regionStates.getRegionState(encodedName));
1049 return;
1050 }
1051 AtomicInteger failedOpenCount = failedOpenTracker.get(encodedName);
1052 if (failedOpenCount == null) {
1053 failedOpenCount = new AtomicInteger();
1054
1055
1056
1057
1058 failedOpenTracker.put(encodedName, failedOpenCount);
1059 }
1060 if (failedOpenCount.incrementAndGet() >= maximumAttempts) {
1061
1062 regionStates.updateRegionState(rt, State.FAILED_OPEN);
1063
1064
1065 failedOpenTracker.remove(encodedName);
1066 } else {
1067
1068 regionState = regionStates.setRegionStateTOCLOSED(rt.getRegionName(), sn);
1069 if (regionState != null) {
1070
1071
1072 try {
1073 getRegionPlan(regionState.getRegion(), sn, true);
1074 new ClosedRegionHandler(server, this, regionState.getRegion()).process();
1075 } catch (HBaseIOException e) {
1076 LOG.warn("Failed to get region plan", e);
1077 }
1078 }
1079 }
1080 break;
1081
1082 case RS_ZK_REGION_OPENING:
1083
1084
1085 if (regionState == null
1086 || !regionState.isPendingOpenOrOpeningOnServer(sn)) {
1087 LOG.warn("Received OPENING for " + prettyPrintedRegionName
1088 + " from " + sn + " but the region isn't PENDING_OPEN/OPENING here: "
1089 + regionStates.getRegionState(encodedName));
1090 return;
1091 }
1092
1093 regionStates.updateRegionState(rt, State.OPENING);
1094 break;
1095
1096 case RS_ZK_REGION_OPENED:
1097
1098 if (regionState == null
1099 || !regionState.isPendingOpenOrOpeningOnServer(sn)) {
1100 LOG.warn("Received OPENED for " + prettyPrintedRegionName
1101 + " from " + sn + " but the region isn't PENDING_OPEN/OPENING here: "
1102 + regionStates.getRegionState(encodedName));
1103
1104 if (regionState != null) {
1105 if(regionState.isOpened() && regionState.getServerName().equals(sn)) {
1106
1107
1108 failedOpenTracker.remove(encodedName);
1109 new OpenedRegionHandler(
1110 server, this, regionState.getRegion(), coordination, ord).process();
1111 updateOpenedRegionHandlerTracker(regionState.getRegion());
1112 } else {
1113
1114
1115
1116 unassign(regionState.getRegion(), null, -1, null, false, sn);
1117 }
1118 }
1119 return;
1120 }
1121
1122 regionState =
1123 regionStates.transitionOpenFromPendingOpenOrOpeningOnServer(rt,regionState, sn);
1124 if (regionState != null) {
1125 failedOpenTracker.remove(encodedName);
1126 new OpenedRegionHandler(
1127 server, this, regionState.getRegion(), coordination, ord).process();
1128 updateOpenedRegionHandlerTracker(regionState.getRegion());
1129 }
1130 break;
1131
1132 default:
1133 throw new IllegalStateException("Received event is not valid.");
1134 }
1135 } finally {
1136 lock.unlock();
1137 }
1138 }
1139
1140
1141 boolean wasClosedHandlerCalled(HRegionInfo hri) {
1142 AtomicBoolean b = closedRegionHandlerCalled.get(hri);
1143
1144
1145
1146 return b == null ? false : b.compareAndSet(true, false);
1147 }
1148
1149
1150 boolean wasOpenedHandlerCalled(HRegionInfo hri) {
1151 AtomicBoolean b = openedRegionHandlerCalled.get(hri);
1152
1153
1154
1155 return b == null ? false : b.compareAndSet(true, false);
1156 }
1157
1158
1159 void initializeHandlerTrackers() {
1160 closedRegionHandlerCalled = new HashMap<HRegionInfo, AtomicBoolean>();
1161 openedRegionHandlerCalled = new HashMap<HRegionInfo, AtomicBoolean>();
1162 }
1163
1164 void updateClosedRegionHandlerTracker(HRegionInfo hri) {
1165 if (closedRegionHandlerCalled != null) {
1166 closedRegionHandlerCalled.put(hri, new AtomicBoolean(true));
1167 }
1168 }
1169
1170 void updateOpenedRegionHandlerTracker(HRegionInfo hri) {
1171 if (openedRegionHandlerCalled != null) {
1172 openedRegionHandlerCalled.put(hri, new AtomicBoolean(true));
1173 }
1174 }
1175
1176
1177
1178
1179
1180
1181 void processFavoredNodes(List<HRegionInfo> regions) throws IOException {
1182 if (!shouldAssignRegionsWithFavoredNodes) return;
1183
1184
1185 Map<HRegionInfo, List<ServerName>> regionToFavoredNodes =
1186 new HashMap<HRegionInfo, List<ServerName>>();
1187 for (HRegionInfo region : regions) {
1188 regionToFavoredNodes.put(region,
1189 ((FavoredNodeLoadBalancer)this.balancer).getFavoredNodes(region));
1190 }
1191 FavoredNodeAssignmentHelper.updateMetaWithFavoredNodesInfo(regionToFavoredNodes,
1192 this.server.getConnection());
1193 }
1194
1195
1196
1197
1198
1199
1200
1201 @SuppressWarnings("deprecation")
1202 private void handleHBCK(RegionTransition rt) {
1203 String encodedName = HRegionInfo.encodeRegionName(rt.getRegionName());
1204 LOG.info("Handling HBCK triggered transition=" + rt.getEventType() +
1205 ", server=" + rt.getServerName() + ", region=" +
1206 HRegionInfo.prettyPrint(encodedName));
1207 RegionState regionState = regionStates.getRegionTransitionState(encodedName);
1208 switch (rt.getEventType()) {
1209 case M_ZK_REGION_OFFLINE:
1210 HRegionInfo regionInfo;
1211 if (regionState != null) {
1212 regionInfo = regionState.getRegion();
1213 } else {
1214 try {
1215 byte [] name = rt.getRegionName();
1216 Pair<HRegionInfo, ServerName> p = MetaTableAccessor.getRegion(
1217 this.server.getConnection(), name);
1218 regionInfo = p.getFirst();
1219 } catch (IOException e) {
1220 LOG.info("Exception reading hbase:meta doing HBCK repair operation", e);
1221 return;
1222 }
1223 }
1224 LOG.info("HBCK repair is triggering assignment of region=" +
1225 regionInfo.getRegionNameAsString());
1226
1227 assign(regionInfo, false);
1228 break;
1229
1230 default:
1231 LOG.warn("Received unexpected region state from HBCK: " + rt.toString());
1232 break;
1233 }
1234
1235 }
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251 @Override
1252 public void nodeCreated(String path) {
1253 handleAssignmentEvent(path);
1254 }
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268 @Override
1269 public void nodeDataChanged(String path) {
1270 handleAssignmentEvent(path);
1271 }
1272
1273
1274
1275
1276
1277 private final Set<String> regionsInProgress = new HashSet<String>();
1278
1279
1280 private final LinkedHashMultimap <String, RegionRunnable>
1281 zkEventWorkerWaitingList = LinkedHashMultimap.create();
1282
1283
1284
1285
1286 private interface RegionRunnable extends Runnable{
1287
1288
1289
1290 String getRegionName();
1291 }
1292
1293
1294
1295
1296
1297 protected void zkEventWorkersSubmit(final RegionRunnable regRunnable) {
1298
1299 synchronized (regionsInProgress) {
1300
1301
1302 if (regionsInProgress.contains(regRunnable.getRegionName())) {
1303 synchronized (zkEventWorkerWaitingList){
1304 zkEventWorkerWaitingList.put(regRunnable.getRegionName(), regRunnable);
1305 }
1306 return;
1307 }
1308
1309
1310 regionsInProgress.add(regRunnable.getRegionName());
1311 zkEventWorkers.submit(new Runnable() {
1312 @Override
1313 public void run() {
1314 try {
1315 regRunnable.run();
1316 } finally {
1317
1318
1319 synchronized (regionsInProgress) {
1320 regionsInProgress.remove(regRunnable.getRegionName());
1321 synchronized (zkEventWorkerWaitingList) {
1322 java.util.Set<RegionRunnable> waiting = zkEventWorkerWaitingList.get(
1323 regRunnable.getRegionName());
1324 if (!waiting.isEmpty()) {
1325
1326 RegionRunnable toSubmit = waiting.iterator().next();
1327 zkEventWorkerWaitingList.remove(toSubmit.getRegionName(), toSubmit);
1328 zkEventWorkersSubmit(toSubmit);
1329 }
1330 }
1331 }
1332 }
1333 }
1334 });
1335 }
1336 }
1337
1338 @Override
1339 public void nodeDeleted(final String path) {
1340 if (path.startsWith(watcher.assignmentZNode)) {
1341 final String regionName = ZKAssign.getRegionName(watcher, path);
1342 zkEventWorkersSubmit(new RegionRunnable() {
1343 @Override
1344 public String getRegionName() {
1345 return regionName;
1346 }
1347
1348 @Override
1349 public void run() {
1350 Lock lock = locker.acquireLock(regionName);
1351 try {
1352 RegionState rs = regionStates.getRegionTransitionState(regionName);
1353 if (rs == null) {
1354 rs = regionStates.getRegionState(regionName);
1355 if (rs == null || !rs.isMergingNew()) {
1356
1357 return;
1358 }
1359 }
1360
1361 HRegionInfo regionInfo = rs.getRegion();
1362 String regionNameStr = regionInfo.getRegionNameAsString();
1363 LOG.debug("Znode " + regionNameStr + " deleted, state: " + rs);
1364
1365 boolean disabled = getTableStateManager().isTableState(regionInfo.getTable(),
1366 ZooKeeperProtos.Table.State.DISABLED, ZooKeeperProtos.Table.State.DISABLING);
1367
1368 ServerName serverName = rs.getServerName();
1369 if (serverManager.isServerOnline(serverName)) {
1370 if (rs.isOnServer(serverName) && (rs.isOpened() || rs.isSplitting())) {
1371 synchronized (regionStates) {
1372 regionOnline(regionInfo, serverName);
1373 if (rs.isSplitting() && splitRegions.containsKey(regionInfo)) {
1374
1375
1376 HRegionInfo hri_a = splitRegions.get(regionInfo).getFirst();
1377 HRegionInfo hri_b = splitRegions.get(regionInfo).getSecond();
1378 if (!regionStates.isRegionInTransition(hri_a.getEncodedName())) {
1379 LOG.warn("Split daughter region not in transition " + hri_a);
1380 }
1381 if (!regionStates.isRegionInTransition(hri_b.getEncodedName())) {
1382 LOG.warn("Split daughter region not in transition" + hri_b);
1383 }
1384 regionOffline(hri_a);
1385 regionOffline(hri_b);
1386 splitRegions.remove(regionInfo);
1387 }
1388 if (disabled) {
1389
1390 LOG.info("Opened " + regionNameStr
1391 + "but this table is disabled, triggering close of region");
1392 unassign(regionInfo);
1393 }
1394 }
1395 } else if (rs.isMergingNew()) {
1396 synchronized (regionStates) {
1397 String p = regionInfo.getEncodedName();
1398 PairOfSameType<HRegionInfo> regions = mergingRegions.get(p);
1399 if (regions != null) {
1400 onlineMergingRegion(disabled, regions.getFirst(), serverName);
1401 onlineMergingRegion(disabled, regions.getSecond(), serverName);
1402 }
1403 }
1404 }
1405 }
1406 } finally {
1407 lock.unlock();
1408 }
1409 }
1410
1411 private void onlineMergingRegion(boolean disabled,
1412 final HRegionInfo hri, final ServerName serverName) {
1413 RegionState regionState = regionStates.getRegionState(hri);
1414 if (regionState != null && regionState.isMerging()
1415 && regionState.isOnServer(serverName)) {
1416 regionOnline(regionState.getRegion(), serverName);
1417 if (disabled) {
1418 unassign(hri);
1419 }
1420 }
1421 }
1422 });
1423 }
1424 }
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438 @Override
1439 public void nodeChildrenChanged(String path) {
1440 if (path.equals(watcher.assignmentZNode)) {
1441 zkEventWorkers.submit(new Runnable() {
1442 @Override
1443 public void run() {
1444 try {
1445
1446 List<String> children =
1447 ZKUtil.listChildrenAndWatchForNewChildren(
1448 watcher, watcher.assignmentZNode);
1449 if (children != null) {
1450 Stat stat = new Stat();
1451 for (String child : children) {
1452
1453
1454
1455 if (!regionStates.isRegionInTransition(child)) {
1456 ZKAssign.getDataAndWatch(watcher, child, stat);
1457 }
1458 }
1459 }
1460 } catch (KeeperException e) {
1461 server.abort("Unexpected ZK exception reading unassigned children", e);
1462 }
1463 }
1464 });
1465 }
1466 }
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477 void regionOnline(HRegionInfo regionInfo, ServerName sn) {
1478 regionOnline(regionInfo, sn, HConstants.NO_SEQNUM);
1479 }
1480
1481 void regionOnline(HRegionInfo regionInfo, ServerName sn, long openSeqNum) {
1482 numRegionsOpened.incrementAndGet();
1483 regionStates.regionOnline(regionInfo, sn, openSeqNum);
1484
1485
1486 clearRegionPlan(regionInfo);
1487 balancer.regionOnline(regionInfo, sn);
1488
1489
1490 sendRegionOpenedNotification(regionInfo, sn);
1491 }
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501 private void handleAssignmentEvent(final String path) {
1502 if (path.startsWith(watcher.assignmentZNode)) {
1503 final String regionName = ZKAssign.getRegionName(watcher, path);
1504
1505 zkEventWorkersSubmit(new RegionRunnable() {
1506 @Override
1507 public String getRegionName() {
1508 return regionName;
1509 }
1510
1511 @Override
1512 public void run() {
1513 try {
1514 Stat stat = new Stat();
1515 byte [] data = ZKAssign.getDataAndWatch(watcher, path, stat);
1516 if (data == null) return;
1517
1518 RegionTransition rt = RegionTransition.parseFrom(data);
1519
1520
1521
1522 BaseCoordinatedStateManager csm =
1523 (BaseCoordinatedStateManager) server.getCoordinatedStateManager();
1524 OpenRegionCoordination openRegionCoordination = csm.getOpenRegionCoordination();
1525
1526 ZkOpenRegionCoordination.ZkOpenRegionDetails zkOrd =
1527 new ZkOpenRegionCoordination.ZkOpenRegionDetails();
1528 zkOrd.setVersion(stat.getVersion());
1529 zkOrd.setServerName(csm.getServer().getServerName());
1530
1531 handleRegion(rt, openRegionCoordination, zkOrd);
1532 } catch (KeeperException e) {
1533 server.abort("Unexpected ZK exception reading unassigned node data", e);
1534 } catch (DeserializationException e) {
1535 server.abort("Unexpected exception deserializing node data", e);
1536 }
1537 }
1538 });
1539 }
1540 }
1541
1542
1543
1544
1545
1546
1547
1548
1549 public void regionOffline(final HRegionInfo regionInfo) {
1550 regionOffline(regionInfo, null);
1551 }
1552
1553 public void offlineDisabledRegion(HRegionInfo regionInfo) {
1554 if (useZKForAssignment) {
1555
1556 LOG.debug("Table being disabled so deleting ZK node and removing from " +
1557 "regions in transition, skipping assignment of region " +
1558 regionInfo.getRegionNameAsString());
1559 String encodedName = regionInfo.getEncodedName();
1560 deleteNodeInStates(encodedName, "closed", null,
1561 EventType.RS_ZK_REGION_CLOSED, EventType.M_ZK_REGION_OFFLINE);
1562 }
1563 replicasToClose.remove(regionInfo);
1564 regionOffline(regionInfo);
1565 }
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587 public void assign(HRegionInfo region, boolean setOfflineInZK) {
1588 assign(region, setOfflineInZK, false);
1589 }
1590
1591
1592
1593
1594 @VisibleForTesting
1595 public void assign(HRegionInfo region,
1596 boolean setOfflineInZK, boolean forceNewPlan) {
1597 if (isDisabledorDisablingRegionInRIT(region)) {
1598 return;
1599 }
1600 String encodedName = region.getEncodedName();
1601 Lock lock = locker.acquireLock(encodedName);
1602 try {
1603 RegionState state = forceRegionStateToOffline(region, forceNewPlan);
1604 if (state != null) {
1605 if (regionStates.wasRegionOnDeadServer(encodedName)) {
1606 LOG.info("Skip assigning " + region.getRegionNameAsString()
1607 + ", it's host " + regionStates.getLastRegionServerOfRegion(encodedName)
1608 + " is dead but not processed yet");
1609 return;
1610 }
1611 assign(state, setOfflineInZK && useZKForAssignment, forceNewPlan);
1612 }
1613 } finally {
1614 lock.unlock();
1615 }
1616 }
1617
1618
1619
1620
1621
1622
1623
1624 boolean assign(final ServerName destination, final List<HRegionInfo> regions)
1625 throws InterruptedException {
1626 long startTime = EnvironmentEdgeManager.currentTime();
1627 try {
1628 int regionCount = regions.size();
1629 if (regionCount == 0) {
1630 return true;
1631 }
1632 LOG.info("Assigning " + regionCount + " region(s) to " + destination.toString());
1633 Set<String> encodedNames = new HashSet<String>(regionCount);
1634 for (HRegionInfo region : regions) {
1635 encodedNames.add(region.getEncodedName());
1636 }
1637
1638 List<HRegionInfo> failedToOpenRegions = new ArrayList<HRegionInfo>();
1639 Map<String, Lock> locks = locker.acquireLocks(encodedNames);
1640 try {
1641 AtomicInteger counter = new AtomicInteger(0);
1642 Map<String, Integer> offlineNodesVersions = new ConcurrentHashMap<String, Integer>();
1643 OfflineCallback cb = new OfflineCallback(
1644 watcher, destination, counter, offlineNodesVersions);
1645 Map<String, RegionPlan> plans = new HashMap<String, RegionPlan>(regions.size());
1646 List<RegionState> states = new ArrayList<RegionState>(regions.size());
1647 for (HRegionInfo region : regions) {
1648 String encodedName = region.getEncodedName();
1649 if (!isDisabledorDisablingRegionInRIT(region)) {
1650 RegionState state = forceRegionStateToOffline(region, false);
1651 boolean onDeadServer = false;
1652 if (state != null) {
1653 if (regionStates.wasRegionOnDeadServer(encodedName)) {
1654 LOG.info("Skip assigning " + region.getRegionNameAsString()
1655 + ", it's host " + regionStates.getLastRegionServerOfRegion(encodedName)
1656 + " is dead but not processed yet");
1657 onDeadServer = true;
1658 } else if (!useZKForAssignment
1659 || asyncSetOfflineInZooKeeper(state, cb, destination)) {
1660 RegionPlan plan = new RegionPlan(region, state.getServerName(), destination);
1661 plans.put(encodedName, plan);
1662 states.add(state);
1663 continue;
1664 }
1665 }
1666
1667 if (!onDeadServer) {
1668 LOG.info("failed to force region state to offline or "
1669 + "failed to set it offline in ZK, will reassign later: " + region);
1670 failedToOpenRegions.add(region);
1671 }
1672 }
1673
1674
1675 Lock lock = locks.remove(encodedName);
1676 lock.unlock();
1677 }
1678
1679 if (useZKForAssignment) {
1680
1681 int total = states.size();
1682 for (int oldCounter = 0; !server.isStopped();) {
1683 int count = counter.get();
1684 if (oldCounter != count) {
1685 LOG.debug(destination.toString() + " unassigned znodes=" + count +
1686 " of total=" + total + "; oldCounter=" + oldCounter);
1687 oldCounter = count;
1688 }
1689 if (count >= total) break;
1690 Thread.sleep(5);
1691 }
1692 }
1693
1694 if (server.isStopped()) {
1695 return false;
1696 }
1697
1698
1699
1700 this.addPlans(plans);
1701
1702 List<Triple<HRegionInfo, Integer, List<ServerName>>> regionOpenInfos =
1703 new ArrayList<Triple<HRegionInfo, Integer, List<ServerName>>>(states.size());
1704 for (RegionState state: states) {
1705 HRegionInfo region = state.getRegion();
1706 String encodedRegionName = region.getEncodedName();
1707 Integer nodeVersion = offlineNodesVersions.get(encodedRegionName);
1708 if (useZKForAssignment && (nodeVersion == null || nodeVersion == -1)) {
1709 LOG.warn("failed to offline in zookeeper: " + region);
1710 failedToOpenRegions.add(region);
1711 Lock lock = locks.remove(encodedRegionName);
1712 lock.unlock();
1713 } else {
1714 regionStates.updateRegionState(
1715 region, State.PENDING_OPEN, destination);
1716 List<ServerName> favoredNodes = ServerName.EMPTY_SERVER_LIST;
1717 if (this.shouldAssignRegionsWithFavoredNodes) {
1718 favoredNodes = ((FavoredNodeLoadBalancer)this.balancer).getFavoredNodes(region);
1719 }
1720 regionOpenInfos.add(new Triple<HRegionInfo, Integer, List<ServerName>>(
1721 region, nodeVersion, favoredNodes));
1722 }
1723 }
1724
1725
1726 try {
1727
1728
1729 long maxWaitTime = System.currentTimeMillis() +
1730 this.server.getConfiguration().
1731 getLong("hbase.regionserver.rpc.startup.waittime", 60000);
1732 for (int i = 1; i <= maximumAttempts && !server.isStopped(); i++) {
1733 try {
1734
1735 if (regionOpenInfos.isEmpty()) {
1736 break;
1737 }
1738 List<RegionOpeningState> regionOpeningStateList = serverManager
1739 .sendRegionOpen(destination, regionOpenInfos);
1740 if (regionOpeningStateList == null) {
1741
1742 return false;
1743 }
1744 for (int k = 0, n = regionOpeningStateList.size(); k < n; k++) {
1745 RegionOpeningState openingState = regionOpeningStateList.get(k);
1746 if (openingState != RegionOpeningState.OPENED) {
1747 HRegionInfo region = regionOpenInfos.get(k).getFirst();
1748 if (openingState == RegionOpeningState.ALREADY_OPENED) {
1749 processAlreadyOpenedRegion(region, destination);
1750 } else if (openingState == RegionOpeningState.FAILED_OPENING) {
1751
1752 failedToOpenRegions.add(region);
1753 } else {
1754 LOG.warn("THIS SHOULD NOT HAPPEN: unknown opening state "
1755 + openingState + " in assigning region " + region);
1756 }
1757 }
1758 }
1759 break;
1760 } catch (IOException e) {
1761 if (e instanceof RemoteException) {
1762 e = ((RemoteException)e).unwrapRemoteException();
1763 }
1764 if (e instanceof RegionServerStoppedException) {
1765 LOG.warn("The region server was shut down, ", e);
1766
1767 return false;
1768 } else if (e instanceof ServerNotRunningYetException) {
1769 long now = System.currentTimeMillis();
1770 if (now < maxWaitTime) {
1771 LOG.debug("Server is not yet up; waiting up to " +
1772 (maxWaitTime - now) + "ms", e);
1773 Thread.sleep(100);
1774 i--;
1775 continue;
1776 }
1777 } else if (e instanceof java.net.SocketTimeoutException
1778 && this.serverManager.isServerOnline(destination)) {
1779
1780
1781
1782
1783 if (LOG.isDebugEnabled()) {
1784 LOG.debug("Bulk assigner openRegion() to " + destination
1785 + " has timed out, but the regions might"
1786 + " already be opened on it.", e);
1787 }
1788
1789 Thread.sleep(100);
1790 i--;
1791 continue;
1792 }
1793 throw e;
1794 }
1795 }
1796 } catch (IOException e) {
1797
1798 LOG.info("Unable to communicate with " + destination
1799 + " in order to assign regions, ", e);
1800 return false;
1801 }
1802 } finally {
1803 for (Lock lock : locks.values()) {
1804 lock.unlock();
1805 }
1806 }
1807
1808 if (!failedToOpenRegions.isEmpty()) {
1809 for (HRegionInfo region : failedToOpenRegions) {
1810 if (!regionStates.isRegionOnline(region)) {
1811 invokeAssign(region);
1812 }
1813 }
1814 }
1815
1816
1817 ArrayList<HRegionInfo> userRegionSet = new ArrayList<HRegionInfo>(regions.size());
1818 for (HRegionInfo region: regions) {
1819 if (!region.getTable().isSystemTable()) {
1820 userRegionSet.add(region);
1821 }
1822 }
1823 if (!waitForAssignment(userRegionSet, true, userRegionSet.size(),
1824 System.currentTimeMillis())) {
1825 LOG.debug("some user regions are still in transition: " + userRegionSet);
1826 }
1827 LOG.debug("Bulk assigning done for " + destination);
1828 return true;
1829 } finally {
1830 metricsAssignmentManager.updateBulkAssignTime(EnvironmentEdgeManager.currentTime() - startTime);
1831 }
1832 }
1833
1834
1835
1836
1837
1838
1839
1840
1841
1842
1843
1844 private void unassign(final HRegionInfo region,
1845 final RegionState state, final int versionOfClosingNode,
1846 final ServerName dest, final boolean transitionInZK,
1847 final ServerName src) {
1848 ServerName server = src;
1849 if (state != null) {
1850 server = state.getServerName();
1851 }
1852 long maxWaitTime = -1;
1853 for (int i = 1; i <= this.maximumAttempts; i++) {
1854 if (this.server.isStopped() || this.server.isAborted()) {
1855 LOG.debug("Server stopped/aborted; skipping unassign of " + region);
1856 return;
1857 }
1858
1859 if (!serverManager.isServerOnline(server)) {
1860 LOG.debug("Offline " + region.getRegionNameAsString()
1861 + ", no need to unassign since it's on a dead server: " + server);
1862 if (transitionInZK) {
1863
1864 deleteClosingOrClosedNode(region, server);
1865 }
1866 if (state != null) {
1867 regionOffline(region);
1868 }
1869 return;
1870 }
1871 try {
1872
1873 if (serverManager.sendRegionClose(server, region,
1874 versionOfClosingNode, dest, transitionInZK)) {
1875 LOG.debug("Sent CLOSE to " + server + " for region " +
1876 region.getRegionNameAsString());
1877 if (useZKForAssignment && !transitionInZK && state != null) {
1878
1879
1880 unassign(region, state, versionOfClosingNode,
1881 dest, transitionInZK, src);
1882 }
1883 return;
1884 }
1885
1886
1887 LOG.warn("Server " + server + " region CLOSE RPC returned false for " +
1888 region.getRegionNameAsString());
1889 } catch (Throwable t) {
1890 long sleepTime = 0;
1891 Configuration conf = this.server.getConfiguration();
1892 if (t instanceof RemoteException) {
1893 t = ((RemoteException)t).unwrapRemoteException();
1894 }
1895 boolean logRetries = true;
1896 if (t instanceof RegionServerAbortedException
1897 || t instanceof RegionServerStoppedException
1898 || t instanceof ServerNotRunningYetException) {
1899
1900
1901 sleepTime = 1 + conf.getInt(RpcClient.FAILED_SERVER_EXPIRY_KEY,
1902 RpcClient.FAILED_SERVER_EXPIRY_DEFAULT);
1903
1904 } else if (t instanceof NotServingRegionException) {
1905 LOG.debug("Offline " + region.getRegionNameAsString()
1906 + ", it's not any more on " + server, t);
1907 if (transitionInZK) {
1908 deleteClosingOrClosedNode(region, server);
1909 }
1910 if (state != null) {
1911 regionOffline(region);
1912 }
1913 return;
1914 } else if ((t instanceof FailedServerException) || (state != null &&
1915 t instanceof RegionAlreadyInTransitionException)) {
1916 if(t instanceof FailedServerException) {
1917 sleepTime = 1 + conf.getInt(RpcClient.FAILED_SERVER_EXPIRY_KEY,
1918 RpcClient.FAILED_SERVER_EXPIRY_DEFAULT);
1919 } else {
1920
1921 LOG.debug("update " + state + " the timestamp.");
1922 state.updateTimestampToNow();
1923 if (maxWaitTime < 0) {
1924 maxWaitTime =
1925 EnvironmentEdgeManager.currentTime()
1926 + conf.getLong(ALREADY_IN_TRANSITION_WAITTIME,
1927 DEFAULT_ALREADY_IN_TRANSITION_WAITTIME);
1928 }
1929 long now = EnvironmentEdgeManager.currentTime();
1930 if (now < maxWaitTime) {
1931 LOG.debug("Region is already in transition; "
1932 + "waiting up to " + (maxWaitTime - now) + "ms", t);
1933 sleepTime = 100;
1934 i--;
1935 logRetries = false;
1936 }
1937 }
1938 }
1939
1940 try {
1941 if (sleepTime > 0) {
1942 Thread.sleep(sleepTime);
1943 }
1944 } catch (InterruptedException ie) {
1945 LOG.warn("Failed to unassign "
1946 + region.getRegionNameAsString() + " since interrupted", ie);
1947 Thread.currentThread().interrupt();
1948 if (state != null) {
1949 regionStates.updateRegionState(region, State.FAILED_CLOSE);
1950 }
1951 return;
1952 }
1953
1954 if (logRetries) {
1955 LOG.info("Server " + server + " returned " + t + " for "
1956 + region.getRegionNameAsString() + ", try=" + i
1957 + " of " + this.maximumAttempts, t);
1958
1959 }
1960 }
1961 }
1962
1963 if (state != null) {
1964 regionStates.updateRegionState(region, State.FAILED_CLOSE);
1965 }
1966 }
1967
1968
1969
1970
1971 private RegionState forceRegionStateToOffline(
1972 final HRegionInfo region, final boolean forceNewPlan) {
1973 RegionState state = regionStates.getRegionState(region);
1974 if (state == null) {
1975 LOG.warn("Assigning but not in region states: " + region);
1976 state = regionStates.createRegionState(region);
1977 }
1978
1979 ServerName sn = state.getServerName();
1980 if (forceNewPlan && LOG.isDebugEnabled()) {
1981 LOG.debug("Force region state offline " + state);
1982 }
1983
1984 switch (state.getState()) {
1985 case OPEN:
1986 case OPENING:
1987 case PENDING_OPEN:
1988 case CLOSING:
1989 case PENDING_CLOSE:
1990 if (!forceNewPlan) {
1991 LOG.debug("Skip assigning " +
1992 region + ", it is already " + state);
1993 return null;
1994 }
1995 case FAILED_CLOSE:
1996 case FAILED_OPEN:
1997 unassign(region, state, -1, null, false, null);
1998 state = regionStates.getRegionState(region);
1999 if (state.isFailedClose()) {
2000
2001
2002 LOG.info("Skip assigning " +
2003 region + ", we couldn't close it: " + state);
2004 return null;
2005 }
2006 case OFFLINE:
2007
2008
2009
2010
2011
2012
2013
2014
2015 if (useZKForAssignment
2016 && regionStates.isServerDeadAndNotProcessed(sn)
2017 && wasRegionOnDeadServerByMeta(region, sn)) {
2018 if (!regionStates.isRegionInTransition(region)) {
2019 LOG.info("Updating the state to " + State.OFFLINE + " to allow to be reassigned by SSH");
2020 regionStates.updateRegionState(region, State.OFFLINE);
2021 }
2022 LOG.info("Skip assigning " + region.getRegionNameAsString()
2023 + ", it is on a dead but not processed yet server: " + sn);
2024 return null;
2025 }
2026 case CLOSED:
2027 break;
2028 default:
2029 LOG.error("Trying to assign region " + region
2030 + ", which is " + state);
2031 return null;
2032 }
2033 return state;
2034 }
2035
2036 @SuppressWarnings("deprecation")
2037 protected boolean wasRegionOnDeadServerByMeta(
2038 final HRegionInfo region, final ServerName sn) {
2039 try {
2040 if (region.isMetaRegion()) {
2041 ServerName server = this.server.getMetaTableLocator().
2042 getMetaRegionLocation(this.server.getZooKeeper());
2043 return regionStates.isServerDeadAndNotProcessed(server);
2044 }
2045 while (!server.isStopped()) {
2046 try {
2047 this.server.getMetaTableLocator().waitMetaRegionLocation(server.getZooKeeper());
2048 Result r = MetaTableAccessor.getRegionResult(server.getConnection(),
2049 region.getRegionName());
2050 if (r == null || r.isEmpty()) return false;
2051 ServerName server = HRegionInfo.getServerName(r);
2052 return regionStates.isServerDeadAndNotProcessed(server);
2053 } catch (IOException ioe) {
2054 LOG.info("Received exception accessing hbase:meta during force assign "
2055 + region.getRegionNameAsString() + ", retrying", ioe);
2056 }
2057 }
2058 } catch (InterruptedException e) {
2059 Thread.currentThread().interrupt();
2060 LOG.info("Interrupted accessing hbase:meta", e);
2061 }
2062
2063 return regionStates.isServerDeadAndNotProcessed(sn);
2064 }
2065
2066
2067
2068
2069
2070
2071
2072 public void assign(RegionState state,
2073 boolean setOfflineInZK, final boolean forceNewPlan) {
2074 long startTime = EnvironmentEdgeManager.currentTime();
2075 try {
2076 Configuration conf = server.getConfiguration();
2077 RegionState currentState = state;
2078 int versionOfOfflineNode = -1;
2079 RegionPlan plan = null;
2080 long maxWaitTime = -1;
2081 HRegionInfo region = state.getRegion();
2082 RegionOpeningState regionOpenState;
2083 Throwable previousException = null;
2084 for (int i = 1; i <= maximumAttempts; i++) {
2085 if (server.isStopped() || server.isAborted()) {
2086 LOG.info("Skip assigning " + region.getRegionNameAsString()
2087 + ", the server is stopped/aborted");
2088 return;
2089 }
2090
2091 if (plan == null) {
2092 try {
2093 plan = getRegionPlan(region, forceNewPlan);
2094 } catch (HBaseIOException e) {
2095 LOG.warn("Failed to get region plan", e);
2096 }
2097 }
2098
2099 if (plan == null) {
2100 LOG.warn("Unable to determine a plan to assign " + region);
2101
2102
2103 if (region.isMetaRegion()) {
2104 if (i == maximumAttempts) {
2105 i = 0;
2106
2107 LOG.warn("Unable to determine a plan to assign a hbase:meta region " + region +
2108 " after maximumAttempts (" + this.maximumAttempts +
2109 "). Reset attempts count and continue retrying.");
2110 }
2111 waitForRetryingMetaAssignment();
2112 continue;
2113 }
2114
2115 regionStates.updateRegionState(region, State.FAILED_OPEN);
2116 return;
2117 }
2118 if (setOfflineInZK && versionOfOfflineNode == -1) {
2119 LOG.info("Setting node as OFFLINED in ZooKeeper for region " + region);
2120
2121
2122 versionOfOfflineNode = setOfflineInZooKeeper(currentState, plan.getDestination());
2123 if (versionOfOfflineNode != -1) {
2124 if (isDisabledorDisablingRegionInRIT(region)) {
2125 return;
2126 }
2127
2128
2129
2130
2131
2132
2133 TableName tableName = region.getTable();
2134 if (!tableStateManager.isTableState(tableName,
2135 ZooKeeperProtos.Table.State.ENABLED, ZooKeeperProtos.Table.State.ENABLING)) {
2136 LOG.debug("Setting table " + tableName + " to ENABLED state.");
2137 setEnabledTable(tableName);
2138 }
2139 }
2140 }
2141 if (setOfflineInZK && versionOfOfflineNode == -1) {
2142 LOG.info("Unable to set offline in ZooKeeper to assign " + region);
2143
2144
2145
2146
2147 if (!server.isAborted()) {
2148 continue;
2149 }
2150 }
2151 LOG.info("Assigning " + region.getRegionNameAsString() +
2152 " to " + plan.getDestination().toString());
2153
2154 currentState = regionStates.updateRegionState(region,
2155 State.PENDING_OPEN, plan.getDestination());
2156
2157 boolean needNewPlan;
2158 final String assignMsg = "Failed assignment of " + region.getRegionNameAsString() +
2159 " to " + plan.getDestination();
2160 try {
2161 List<ServerName> favoredNodes = ServerName.EMPTY_SERVER_LIST;
2162 if (this.shouldAssignRegionsWithFavoredNodes) {
2163 favoredNodes = ((FavoredNodeLoadBalancer)this.balancer).getFavoredNodes(region);
2164 }
2165 regionOpenState = serverManager.sendRegionOpen(
2166 plan.getDestination(), region, versionOfOfflineNode, favoredNodes);
2167
2168 if (regionOpenState == RegionOpeningState.FAILED_OPENING) {
2169
2170 needNewPlan = true;
2171 LOG.warn(assignMsg + ", regionserver says 'FAILED_OPENING', " +
2172 " trying to assign elsewhere instead; " +
2173 "try=" + i + " of " + this.maximumAttempts);
2174 } else {
2175
2176 if (regionOpenState == RegionOpeningState.ALREADY_OPENED) {
2177 processAlreadyOpenedRegion(region, plan.getDestination());
2178 }
2179 return;
2180 }
2181
2182 } catch (Throwable t) {
2183 if (t instanceof RemoteException) {
2184 t = ((RemoteException) t).unwrapRemoteException();
2185 }
2186 previousException = t;
2187
2188
2189
2190
2191 boolean hold = (t instanceof RegionAlreadyInTransitionException ||
2192 t instanceof ServerNotRunningYetException);
2193
2194
2195
2196
2197
2198
2199 boolean retry = !hold && (t instanceof java.net.SocketTimeoutException
2200 && this.serverManager.isServerOnline(plan.getDestination()));
2201
2202
2203 if (hold) {
2204 LOG.warn(assignMsg + ", waiting a little before trying on the same region server " +
2205 "try=" + i + " of " + this.maximumAttempts, t);
2206
2207 if (maxWaitTime < 0) {
2208 if (t instanceof RegionAlreadyInTransitionException) {
2209 maxWaitTime = EnvironmentEdgeManager.currentTime()
2210 + this.server.getConfiguration().getLong(ALREADY_IN_TRANSITION_WAITTIME,
2211 DEFAULT_ALREADY_IN_TRANSITION_WAITTIME);
2212 } else {
2213 maxWaitTime = EnvironmentEdgeManager.currentTime()
2214 + this.server.getConfiguration().getLong(
2215 "hbase.regionserver.rpc.startup.waittime", 60000);
2216 }
2217 }
2218 try {
2219 needNewPlan = false;
2220 long now = EnvironmentEdgeManager.currentTime();
2221 if (now < maxWaitTime) {
2222 LOG.debug("Server is not yet up or region is already in transition; "
2223 + "waiting up to " + (maxWaitTime - now) + "ms", t);
2224 Thread.sleep(100);
2225 i--;
2226 } else if (!(t instanceof RegionAlreadyInTransitionException)) {
2227 LOG.debug("Server is not up for a while; try a new one", t);
2228 needNewPlan = true;
2229 }
2230 } catch (InterruptedException ie) {
2231 LOG.warn("Failed to assign "
2232 + region.getRegionNameAsString() + " since interrupted", ie);
2233 regionStates.updateRegionState(region, State.FAILED_OPEN);
2234 Thread.currentThread().interrupt();
2235 return;
2236 }
2237 } else if (retry) {
2238 needNewPlan = false;
2239 i--;
2240 LOG.warn(assignMsg + ", trying to assign to the same region server due ", t);
2241 } else {
2242 needNewPlan = true;
2243 LOG.warn(assignMsg + ", trying to assign elsewhere instead;" +
2244 " try=" + i + " of " + this.maximumAttempts, t);
2245 }
2246 }
2247
2248 if (i == this.maximumAttempts) {
2249
2250 if (region.isMetaRegion()) {
2251 i = 0;
2252 LOG.warn(assignMsg +
2253 ", trying to assign a hbase:meta region reached to maximumAttempts (" +
2254 this.maximumAttempts + "). Reset attempt counts and continue retrying.");
2255 waitForRetryingMetaAssignment();
2256 }
2257 else {
2258
2259
2260 continue;
2261 }
2262 }
2263
2264
2265
2266
2267 if (needNewPlan) {
2268
2269
2270
2271
2272 RegionPlan newPlan = null;
2273 try {
2274 newPlan = getRegionPlan(region, true);
2275 } catch (HBaseIOException e) {
2276 LOG.warn("Failed to get region plan", e);
2277 }
2278 if (newPlan == null) {
2279 regionStates.updateRegionState(region, State.FAILED_OPEN);
2280 LOG.warn("Unable to find a viable location to assign region " +
2281 region.getRegionNameAsString());
2282 return;
2283 }
2284
2285 if (plan != newPlan && !plan.getDestination().equals(newPlan.getDestination())) {
2286
2287
2288
2289 LOG.info("Region assignment plan changed from " + plan.getDestination() + " to "
2290 + newPlan.getDestination() + " server.");
2291 currentState = regionStates.updateRegionState(region, State.OFFLINE);
2292 versionOfOfflineNode = -1;
2293 if (useZKForAssignment) {
2294 setOfflineInZK = true;
2295 }
2296 plan = newPlan;
2297 } else if(plan.getDestination().equals(newPlan.getDestination()) &&
2298 previousException instanceof FailedServerException) {
2299 try {
2300 LOG.info("Trying to re-assign " + region.getRegionNameAsString() +
2301 " to the same failed server.");
2302 Thread.sleep(1 + conf.getInt(RpcClient.FAILED_SERVER_EXPIRY_KEY,
2303 RpcClient.FAILED_SERVER_EXPIRY_DEFAULT));
2304 } catch (InterruptedException ie) {
2305 LOG.warn("Failed to assign "
2306 + region.getRegionNameAsString() + " since interrupted", ie);
2307 regionStates.updateRegionState(region, State.FAILED_OPEN);
2308 Thread.currentThread().interrupt();
2309 return;
2310 }
2311 }
2312 }
2313 }
2314
2315 regionStates.updateRegionState(region, State.FAILED_OPEN);
2316 } finally {
2317 metricsAssignmentManager.updateAssignmentTime(EnvironmentEdgeManager.currentTime() - startTime);
2318 }
2319 }
2320
2321 private void processAlreadyOpenedRegion(HRegionInfo region, ServerName sn) {
2322
2323
2324
2325 LOG.debug("ALREADY_OPENED " + region.getRegionNameAsString()
2326 + " to " + sn);
2327 String encodedName = region.getEncodedName();
2328
2329
2330
2331 if(useZKForAssignment){
2332 String node = ZKAssign.getNodeName(watcher, encodedName);
2333 Stat stat = new Stat();
2334 try {
2335 byte[] existingBytes = ZKUtil.getDataNoWatch(watcher, node, stat);
2336 if(existingBytes!=null){
2337 RegionTransition rt= RegionTransition.parseFrom(existingBytes);
2338 EventType et = rt.getEventType();
2339 if (et.equals(EventType.RS_ZK_REGION_OPENED)) {
2340 LOG.debug("ALREADY_OPENED " + region.getRegionNameAsString()
2341 + " and node in "+et+" state");
2342 return;
2343 }
2344 }
2345 } catch (KeeperException ke) {
2346 LOG.warn("Unexpected ZK exception getData " + node
2347 + " node for the region " + encodedName, ke);
2348 } catch (DeserializationException e) {
2349 LOG.warn("Get RegionTransition from zk deserialization failed! ", e);
2350 }
2351
2352 deleteNodeInStates(encodedName, "offline", sn, EventType.M_ZK_REGION_OFFLINE);
2353 }
2354
2355 regionStates.regionOnline(region, sn);
2356 }
2357
2358 private boolean isDisabledorDisablingRegionInRIT(final HRegionInfo region) {
2359 if (this.tableStateManager.isTableState(region.getTable(),
2360 ZooKeeperProtos.Table.State.DISABLED,
2361 ZooKeeperProtos.Table.State.DISABLING) || replicasToClose.contains(region)) {
2362 LOG.info("Table " + region.getTable() + " is disabled or disabling;"
2363 + " skipping assign of " + region.getRegionNameAsString());
2364 offlineDisabledRegion(region);
2365 return true;
2366 }
2367 return false;
2368 }
2369
2370
2371
2372
2373
2374
2375
2376
2377 private int setOfflineInZooKeeper(final RegionState state, final ServerName destination) {
2378 if (!state.isClosed() && !state.isOffline()) {
2379 String msg = "Unexpected state : " + state + " .. Cannot transit it to OFFLINE.";
2380 this.server.abort(msg, new IllegalStateException(msg));
2381 return -1;
2382 }
2383 regionStates.updateRegionState(state.getRegion(), State.OFFLINE);
2384 int versionOfOfflineNode;
2385 try {
2386
2387 versionOfOfflineNode = ZKAssign.createOrForceNodeOffline(watcher,
2388 state.getRegion(), destination);
2389 if (versionOfOfflineNode == -1) {
2390 LOG.warn("Attempted to create/force node into OFFLINE state before "
2391 + "completing assignment but failed to do so for " + state);
2392 return -1;
2393 }
2394 } catch (KeeperException e) {
2395 server.abort("Unexpected ZK exception creating/setting node OFFLINE", e);
2396 return -1;
2397 }
2398 return versionOfOfflineNode;
2399 }
2400
2401
2402
2403
2404
2405
2406 private RegionPlan getRegionPlan(final HRegionInfo region,
2407 final boolean forceNewPlan) throws HBaseIOException {
2408 return getRegionPlan(region, null, forceNewPlan);
2409 }
2410
2411
2412
2413
2414
2415
2416
2417
2418
2419
2420 private RegionPlan getRegionPlan(final HRegionInfo region,
2421 final ServerName serverToExclude, final boolean forceNewPlan) throws HBaseIOException {
2422
2423 final String encodedName = region.getEncodedName();
2424 final List<ServerName> destServers =
2425 serverManager.createDestinationServersList(serverToExclude);
2426
2427 if (destServers.isEmpty()){
2428 LOG.warn("Can't move " + encodedName +
2429 ", there is no destination server available.");
2430 return null;
2431 }
2432
2433 RegionPlan randomPlan = null;
2434 boolean newPlan = false;
2435 RegionPlan existingPlan;
2436
2437 synchronized (this.regionPlans) {
2438 existingPlan = this.regionPlans.get(encodedName);
2439
2440 if (existingPlan != null && existingPlan.getDestination() != null) {
2441 LOG.debug("Found an existing plan for " + region.getRegionNameAsString()
2442 + " destination server is " + existingPlan.getDestination() +
2443 " accepted as a dest server = " + destServers.contains(existingPlan.getDestination()));
2444 }
2445
2446 if (forceNewPlan
2447 || existingPlan == null
2448 || existingPlan.getDestination() == null
2449 || !destServers.contains(existingPlan.getDestination())) {
2450 newPlan = true;
2451 }
2452 }
2453
2454 if (newPlan) {
2455 ServerName destination = balancer.randomAssignment(region, destServers);
2456 if (destination == null) {
2457 LOG.warn("Can't find a destination for " + encodedName);
2458 return null;
2459 }
2460 synchronized (this.regionPlans) {
2461 randomPlan = new RegionPlan(region, null, destination);
2462 if (!region.isMetaTable() && shouldAssignRegionsWithFavoredNodes) {
2463 List<HRegionInfo> regions = new ArrayList<HRegionInfo>(1);
2464 regions.add(region);
2465 try {
2466 processFavoredNodes(regions);
2467 } catch (IOException ie) {
2468 LOG.warn("Ignoring exception in processFavoredNodes " + ie);
2469 }
2470 }
2471 this.regionPlans.put(encodedName, randomPlan);
2472 }
2473 LOG.debug("No previous transition plan found (or ignoring " + "an existing plan) for "
2474 + region.getRegionNameAsString() + "; generated random plan=" + randomPlan + "; "
2475 + destServers.size() + " (online=" + serverManager.getOnlineServers().size()
2476 + ") available servers, forceNewPlan=" + forceNewPlan);
2477 return randomPlan;
2478 }
2479 LOG.debug("Using pre-existing plan for " +
2480 region.getRegionNameAsString() + "; plan=" + existingPlan);
2481 return existingPlan;
2482 }
2483
2484
2485
2486
2487 private void waitForRetryingMetaAssignment() {
2488 try {
2489 Thread.sleep(this.sleepTimeBeforeRetryingMetaAssignment);
2490 } catch (InterruptedException e) {
2491 LOG.error("Got exception while waiting for hbase:meta assignment");
2492 Thread.currentThread().interrupt();
2493 }
2494 }
2495
2496
2497
2498
2499
2500
2501
2502
2503
2504
2505
2506
2507
2508
2509 public void unassign(HRegionInfo region) {
2510 unassign(region, false);
2511 }
2512
2513
2514
2515
2516
2517
2518
2519
2520
2521
2522
2523
2524
2525
2526
2527
2528 public void unassign(HRegionInfo region, boolean force, ServerName dest) {
2529
2530 LOG.debug("Starting unassign of " + region.getRegionNameAsString()
2531 + " (offlining), current state: " + regionStates.getRegionState(region));
2532
2533 String encodedName = region.getEncodedName();
2534
2535 int versionOfClosingNode = -1;
2536
2537
2538 ReentrantLock lock = locker.acquireLock(encodedName);
2539 RegionState state = regionStates.getRegionTransitionState(encodedName);
2540 boolean reassign = true;
2541 try {
2542 if (state == null) {
2543
2544
2545 state = regionStates.getRegionState(encodedName);
2546 if (state != null && state.isUnassignable()) {
2547 LOG.info("Attempting to unassign " + state + ", ignored");
2548
2549 return;
2550 }
2551
2552 try {
2553 if (state == null || state.getServerName() == null) {
2554
2555
2556 LOG.warn("Attempting to unassign a region not in RegionStates "
2557 + region.getRegionNameAsString() + ", offlined");
2558 regionOffline(region);
2559 return;
2560 }
2561 if (useZKForAssignment) {
2562 versionOfClosingNode = ZKAssign.createNodeClosing(
2563 watcher, region, state.getServerName());
2564 if (versionOfClosingNode == -1) {
2565 LOG.info("Attempting to unassign " +
2566 region.getRegionNameAsString() + " but ZK closing node "
2567 + "can't be created.");
2568 reassign = false;
2569 return;
2570 }
2571 }
2572 } catch (KeeperException e) {
2573 if (e instanceof NodeExistsException) {
2574
2575
2576
2577
2578 NodeExistsException nee = (NodeExistsException)e;
2579 String path = nee.getPath();
2580 try {
2581 if (isSplitOrSplittingOrMergedOrMerging(path)) {
2582 LOG.debug(path + " is SPLIT or SPLITTING or MERGED or MERGING; " +
2583 "skipping unassign because region no longer exists -- its split or merge");
2584 reassign = false;
2585 return;
2586 }
2587 } catch (KeeperException.NoNodeException ke) {
2588 LOG.warn("Failed getData on SPLITTING/SPLIT at " + path +
2589 "; presuming split and that the region to unassign, " +
2590 encodedName + ", no longer exists -- confirm", ke);
2591 return;
2592 } catch (KeeperException ke) {
2593 LOG.error("Unexpected zk state", ke);
2594 } catch (DeserializationException de) {
2595 LOG.error("Failed parse", de);
2596 }
2597 }
2598
2599 server.abort("Unexpected ZK exception creating node CLOSING", e);
2600 reassign = false;
2601 return;
2602 }
2603 state = regionStates.updateRegionState(region, State.PENDING_CLOSE);
2604 } else if (state.isFailedOpen()) {
2605
2606 regionOffline(region);
2607 return;
2608 } else if (force && state.isPendingCloseOrClosing()) {
2609 LOG.debug("Attempting to unassign " + region.getRegionNameAsString() +
2610 " which is already " + state.getState() +
2611 " but forcing to send a CLOSE RPC again ");
2612 if (state.isFailedClose()) {
2613 state = regionStates.updateRegionState(region, State.PENDING_CLOSE);
2614 }
2615 state.updateTimestampToNow();
2616 } else {
2617 LOG.debug("Attempting to unassign " +
2618 region.getRegionNameAsString() + " but it is " +
2619 "already in transition (" + state.getState() + ", force=" + force + ")");
2620 return;
2621 }
2622
2623 unassign(region, state, versionOfClosingNode, dest, useZKForAssignment, null);
2624 } finally {
2625 lock.unlock();
2626
2627
2628 if (!replicasToClose.contains(region) && reassign && regionStates.isRegionOffline(region)) {
2629 assign(region, true);
2630 }
2631 }
2632 }
2633
2634 public void unassign(HRegionInfo region, boolean force){
2635 unassign(region, force, null);
2636 }
2637
2638
2639
2640
2641 public void deleteClosingOrClosedNode(HRegionInfo region, ServerName sn) {
2642 String encodedName = region.getEncodedName();
2643 deleteNodeInStates(encodedName, "closing", sn, EventType.M_ZK_REGION_CLOSING,
2644 EventType.RS_ZK_REGION_CLOSED);
2645 }
2646
2647
2648
2649
2650
2651
2652
2653 private boolean isSplitOrSplittingOrMergedOrMerging(final String path)
2654 throws KeeperException, DeserializationException {
2655 boolean result = false;
2656
2657
2658 byte [] data = ZKAssign.getData(watcher, path);
2659 if (data == null) {
2660 LOG.info("Node " + path + " is gone");
2661 return false;
2662 }
2663 RegionTransition rt = RegionTransition.parseFrom(data);
2664 switch (rt.getEventType()) {
2665 case RS_ZK_REQUEST_REGION_SPLIT:
2666 case RS_ZK_REGION_SPLIT:
2667 case RS_ZK_REGION_SPLITTING:
2668 case RS_ZK_REQUEST_REGION_MERGE:
2669 case RS_ZK_REGION_MERGED:
2670 case RS_ZK_REGION_MERGING:
2671 result = true;
2672 break;
2673 default:
2674 LOG.info("Node " + path + " is in " + rt.getEventType());
2675 break;
2676 }
2677 return result;
2678 }
2679
2680
2681
2682
2683
2684
2685 public int getNumRegionsOpened() {
2686 return numRegionsOpened.get();
2687 }
2688
2689
2690
2691
2692
2693
2694
2695
2696
2697
2698 public boolean waitForAssignment(HRegionInfo regionInfo)
2699 throws InterruptedException {
2700 ArrayList<HRegionInfo> regionSet = new ArrayList<HRegionInfo>(1);
2701 regionSet.add(regionInfo);
2702 return waitForAssignment(regionSet, true, Long.MAX_VALUE);
2703 }
2704
2705
2706
2707
2708 protected boolean waitForAssignment(final Collection<HRegionInfo> regionSet,
2709 final boolean waitTillAllAssigned, final int reassigningRegions,
2710 final long minEndTime) throws InterruptedException {
2711 long deadline = minEndTime + bulkPerRegionOpenTimeGuesstimate * (reassigningRegions + 1);
2712 if (deadline < 0) {
2713 deadline = Long.MAX_VALUE;
2714 }
2715 return waitForAssignment(regionSet, waitTillAllAssigned, deadline);
2716 }
2717
2718
2719
2720
2721
2722
2723
2724
2725
2726 protected boolean waitForAssignment(final Collection<HRegionInfo> regionSet,
2727 final boolean waitTillAllAssigned, final long deadline) throws InterruptedException {
2728
2729 while (!regionSet.isEmpty() && !server.isStopped() && deadline > System.currentTimeMillis()) {
2730 int failedOpenCount = 0;
2731 Iterator<HRegionInfo> regionInfoIterator = regionSet.iterator();
2732 while (regionInfoIterator.hasNext()) {
2733 HRegionInfo hri = regionInfoIterator.next();
2734 if (regionStates.isRegionOnline(hri) || regionStates.isRegionInState(hri,
2735 State.SPLITTING, State.SPLIT, State.MERGING, State.MERGED)) {
2736 regionInfoIterator.remove();
2737 } else if (regionStates.isRegionInState(hri, State.FAILED_OPEN)) {
2738 failedOpenCount++;
2739 }
2740 }
2741 if (!waitTillAllAssigned) {
2742
2743 break;
2744 }
2745 if (!regionSet.isEmpty()) {
2746 if (failedOpenCount == regionSet.size()) {
2747
2748 break;
2749 }
2750 regionStates.waitForUpdate(100);
2751 }
2752 }
2753 return regionSet.isEmpty();
2754 }
2755
2756
2757
2758
2759
2760
2761
2762
2763
2764
2765
2766
2767 public void assignMeta(HRegionInfo hri) throws KeeperException {
2768 this.server.getMetaTableLocator().deleteMetaLocation(this.watcher, hri.getReplicaId());
2769 assign(hri, true);
2770 }
2771
2772
2773
2774
2775
2776
2777
2778
2779
2780 public void assign(Map<HRegionInfo, ServerName> regions)
2781 throws IOException, InterruptedException {
2782 if (regions == null || regions.isEmpty()) {
2783 return;
2784 }
2785 List<ServerName> servers = serverManager.createDestinationServersList();
2786 if (servers == null || servers.isEmpty()) {
2787 throw new IOException("Found no destination server to assign region(s)");
2788 }
2789
2790
2791 Map<ServerName, List<HRegionInfo>> bulkPlan =
2792 balancer.retainAssignment(regions, servers);
2793 if (bulkPlan == null) {
2794 throw new IOException("Unable to determine a plan to assign region(s)");
2795 }
2796
2797 assign(regions.size(), servers.size(),
2798 "retainAssignment=true", bulkPlan);
2799 }
2800
2801
2802
2803
2804
2805
2806
2807
2808
2809 public void assign(List<HRegionInfo> regions)
2810 throws IOException, InterruptedException {
2811 if (regions == null || regions.isEmpty()) {
2812 return;
2813 }
2814
2815 List<ServerName> servers = serverManager.createDestinationServersList();
2816 if (servers == null || servers.isEmpty()) {
2817 throw new IOException("Found no destination server to assign region(s)");
2818 }
2819
2820
2821 Map<ServerName, List<HRegionInfo>> bulkPlan = balancer.roundRobinAssignment(regions, servers);
2822 if (bulkPlan == null) {
2823 throw new IOException("Unable to determine a plan to assign region(s)");
2824 }
2825
2826 processFavoredNodes(regions);
2827 assign(regions.size(), servers.size(), "round-robin=true", bulkPlan);
2828 }
2829
2830 private void assign(int regions, int totalServers,
2831 String message, Map<ServerName, List<HRegionInfo>> bulkPlan)
2832 throws InterruptedException, IOException {
2833
2834 int servers = bulkPlan.size();
2835 if (servers == 1 || (regions < bulkAssignThresholdRegions
2836 && servers < bulkAssignThresholdServers)) {
2837
2838
2839
2840 if (LOG.isTraceEnabled()) {
2841 LOG.trace("Not using bulk assignment since we are assigning only " + regions +
2842 " region(s) to " + servers + " server(s)");
2843 }
2844
2845
2846 ArrayList<HRegionInfo> userRegionSet = new ArrayList<HRegionInfo>(regions);
2847 for (Map.Entry<ServerName, List<HRegionInfo>> plan: bulkPlan.entrySet()) {
2848 if (!assign(plan.getKey(), plan.getValue())) {
2849 for (HRegionInfo region: plan.getValue()) {
2850 if (!regionStates.isRegionOnline(region)) {
2851 invokeAssign(region);
2852 if (!region.getTable().isSystemTable()) {
2853 userRegionSet.add(region);
2854 }
2855 }
2856 }
2857 }
2858 }
2859
2860
2861 if (!waitForAssignment(userRegionSet, true, userRegionSet.size(),
2862 System.currentTimeMillis())) {
2863 LOG.debug("some user regions are still in transition: " + userRegionSet);
2864 }
2865 } else {
2866 LOG.info("Bulk assigning " + regions + " region(s) across "
2867 + totalServers + " server(s), " + message);
2868
2869
2870 BulkAssigner ba = new GeneralBulkAssigner(
2871 this.server, bulkPlan, this, bulkAssignWaitTillAllAssigned);
2872 ba.bulkAssign();
2873 LOG.info("Bulk assigning done");
2874 }
2875 }
2876
2877
2878
2879
2880
2881
2882
2883
2884
2885
2886 private void assignAllUserRegions(Map<HRegionInfo, ServerName> allRegions)
2887 throws IOException, InterruptedException {
2888 if (allRegions == null || allRegions.isEmpty()) return;
2889
2890
2891 boolean retainAssignment = server.getConfiguration().
2892 getBoolean("hbase.master.startup.retainassign", true);
2893
2894 Set<HRegionInfo> regionsFromMetaScan = allRegions.keySet();
2895 if (retainAssignment) {
2896 assign(allRegions);
2897 } else {
2898 List<HRegionInfo> regions = new ArrayList<HRegionInfo>(regionsFromMetaScan);
2899 assign(regions);
2900 }
2901
2902 for (HRegionInfo hri : regionsFromMetaScan) {
2903 TableName tableName = hri.getTable();
2904 if (!tableStateManager.isTableState(tableName,
2905 ZooKeeperProtos.Table.State.ENABLED)) {
2906 setEnabledTable(tableName);
2907 }
2908 }
2909
2910 assign(replicaRegionsNotRecordedInMeta(regionsFromMetaScan, server));
2911 }
2912
2913
2914
2915
2916
2917
2918
2919
2920
2921
2922
2923
2924 public static List<HRegionInfo> replicaRegionsNotRecordedInMeta(
2925 Set<HRegionInfo> regionsRecordedInMeta, MasterServices master)throws IOException {
2926 List<HRegionInfo> regionsNotRecordedInMeta = new ArrayList<HRegionInfo>();
2927 for (HRegionInfo hri : regionsRecordedInMeta) {
2928 TableName table = hri.getTable();
2929 HTableDescriptor htd = master.getTableDescriptors().get(table);
2930
2931 int desiredRegionReplication = htd.getRegionReplication();
2932 for (int i = 0; i < desiredRegionReplication; i++) {
2933 HRegionInfo replica = RegionReplicaUtil.getRegionInfoForReplica(hri, i);
2934 if (regionsRecordedInMeta.contains(replica)) continue;
2935 regionsNotRecordedInMeta.add(replica);
2936 }
2937 }
2938 return regionsNotRecordedInMeta;
2939 }
2940
2941
2942
2943
2944
2945
2946
2947 boolean waitUntilNoRegionsInTransition(final long timeout)
2948 throws InterruptedException {
2949
2950
2951
2952
2953
2954
2955 final long endTime = System.currentTimeMillis() + timeout;
2956
2957 while (!this.server.isStopped() && regionStates.isRegionsInTransition()
2958 && endTime > System.currentTimeMillis()) {
2959 regionStates.waitForUpdate(100);
2960 }
2961
2962 return !regionStates.isRegionsInTransition();
2963 }
2964
2965
2966
2967
2968
2969
2970
2971 Set<ServerName> rebuildUserRegions() throws
2972 IOException, KeeperException, CoordinatedStateException {
2973 Set<TableName> disabledOrEnablingTables = tableStateManager.getTablesInStates(
2974 ZooKeeperProtos.Table.State.DISABLED, ZooKeeperProtos.Table.State.ENABLING);
2975
2976 Set<TableName> disabledOrDisablingOrEnabling = tableStateManager.getTablesInStates(
2977 ZooKeeperProtos.Table.State.DISABLED,
2978 ZooKeeperProtos.Table.State.DISABLING,
2979 ZooKeeperProtos.Table.State.ENABLING);
2980
2981
2982 List<Result> results = MetaTableAccessor.fullScanOfMeta(server.getConnection());
2983
2984 Set<ServerName> onlineServers = serverManager.getOnlineServers().keySet();
2985
2986 Set<ServerName> offlineServers = new HashSet<ServerName>();
2987
2988 for (Result result : results) {
2989 if (result == null && LOG.isDebugEnabled()){
2990 LOG.debug("null result from meta - ignoring but this is strange.");
2991 continue;
2992 }
2993
2994
2995
2996 PairOfSameType<HRegionInfo> p = MetaTableAccessor.getMergeRegions(result);
2997 if (p.getFirst() != null && p.getSecond() != null) {
2998 int numReplicas = server.getTableDescriptors().get(p.getFirst().
2999 getTable()).getRegionReplication();
3000 for (HRegionInfo merge : p) {
3001 for (int i = 1; i < numReplicas; i++) {
3002 replicasToClose.add(RegionReplicaUtil.getRegionInfoForReplica(merge, i));
3003 }
3004 }
3005 }
3006 RegionLocations rl = MetaTableAccessor.getRegionLocations(result);
3007 if (rl == null) continue;
3008 HRegionLocation[] locations = rl.getRegionLocations();
3009 if (locations == null) continue;
3010 for (HRegionLocation hrl : locations) {
3011 if (hrl == null) continue;
3012 HRegionInfo regionInfo = hrl.getRegionInfo();
3013 if (regionInfo == null) continue;
3014 int replicaId = regionInfo.getReplicaId();
3015 State state = RegionStateStore.getRegionState(result, replicaId);
3016
3017
3018
3019 if (replicaId == 0 && state.equals(State.SPLIT)) {
3020 for (HRegionLocation h : locations) {
3021 replicasToClose.add(h.getRegionInfo());
3022 }
3023 }
3024 ServerName lastHost = hrl.getServerName();
3025 ServerName regionLocation = RegionStateStore.getRegionServer(result, replicaId);
3026 if (tableStateManager.isTableState(regionInfo.getTable(),
3027 ZooKeeperProtos.Table.State.DISABLED)) {
3028
3029
3030 lastHost = null;
3031 regionLocation = null;
3032 }
3033 regionStates.createRegionState(regionInfo, state, regionLocation, lastHost);
3034 if (!regionStates.isRegionInState(regionInfo, State.OPEN)) {
3035
3036 continue;
3037 }
3038 TableName tableName = regionInfo.getTable();
3039 if (!onlineServers.contains(regionLocation)) {
3040
3041 offlineServers.add(regionLocation);
3042 if (useZKForAssignment) {
3043 regionStates.regionOffline(regionInfo);
3044 }
3045 } else if (!disabledOrEnablingTables.contains(tableName)) {
3046
3047
3048 regionStates.regionOnline(regionInfo, regionLocation);
3049 balancer.regionOnline(regionInfo, regionLocation);
3050 } else if (useZKForAssignment) {
3051 regionStates.regionOffline(regionInfo);
3052 }
3053
3054
3055 if (!disabledOrDisablingOrEnabling.contains(tableName)
3056 && !getTableStateManager().isTableState(tableName,
3057 ZooKeeperProtos.Table.State.ENABLED)) {
3058 setEnabledTable(tableName);
3059 }
3060 }
3061 }
3062 return offlineServers;
3063 }
3064
3065
3066
3067
3068
3069
3070
3071
3072
3073 private void recoverTableInDisablingState()
3074 throws KeeperException, IOException, CoordinatedStateException {
3075 Set<TableName> disablingTables =
3076 tableStateManager.getTablesInStates(ZooKeeperProtos.Table.State.DISABLING);
3077 if (disablingTables.size() != 0) {
3078 for (TableName tableName : disablingTables) {
3079
3080 LOG.info("The table " + tableName
3081 + " is in DISABLING state. Hence recovering by moving the table"
3082 + " to DISABLED state.");
3083 new DisableTableHandler(this.server, tableName,
3084 this, tableLockManager, true).prepare().process();
3085 }
3086 }
3087 }
3088
3089
3090
3091
3092
3093
3094
3095
3096
3097 private void recoverTableInEnablingState()
3098 throws KeeperException, IOException, CoordinatedStateException {
3099 Set<TableName> enablingTables = tableStateManager.
3100 getTablesInStates(ZooKeeperProtos.Table.State.ENABLING);
3101 if (enablingTables.size() != 0) {
3102 for (TableName tableName : enablingTables) {
3103
3104 LOG.info("The table " + tableName
3105 + " is in ENABLING state. Hence recovering by moving the table"
3106 + " to ENABLED state.");
3107
3108
3109 EnableTableHandler eth = new EnableTableHandler(this.server, tableName,
3110 this, tableLockManager, true);
3111 try {
3112 eth.prepare();
3113 } catch (TableNotFoundException e) {
3114 LOG.warn("Table " + tableName + " not found in hbase:meta to recover.");
3115 continue;
3116 }
3117 eth.process();
3118 }
3119 }
3120 }
3121
3122
3123
3124
3125
3126
3127
3128
3129
3130
3131
3132
3133 private void processDeadServersAndRecoverLostRegions(Set<ServerName> deadServers)
3134 throws IOException, KeeperException {
3135 if (deadServers != null && !deadServers.isEmpty()) {
3136 for (ServerName serverName: deadServers) {
3137 if (!serverManager.isServerDead(serverName)) {
3138 serverManager.expireServer(serverName);
3139 }
3140 }
3141 }
3142
3143 List<String> nodes = useZKForAssignment ?
3144 ZKUtil.listChildrenAndWatchForNewChildren(watcher, watcher.assignmentZNode)
3145 : ZKUtil.listChildrenNoWatch(watcher, watcher.assignmentZNode);
3146 if (nodes != null && !nodes.isEmpty()) {
3147 for (String encodedRegionName : nodes) {
3148 processRegionInTransition(encodedRegionName, null);
3149 }
3150 } else if (!useZKForAssignment) {
3151 processRegionInTransitionZkLess();
3152 }
3153 }
3154
3155 void processRegionInTransitionZkLess() {
3156
3157
3158
3159
3160
3161 Map<String, RegionState> rits = regionStates.getRegionsInTransition();
3162 for (RegionState regionState : rits.values()) {
3163 LOG.info("Processing " + regionState);
3164 ServerName serverName = regionState.getServerName();
3165
3166
3167 if (serverName != null
3168 && !serverManager.getOnlineServers().containsKey(serverName)) {
3169 LOG.info("Server " + serverName + " isn't online. SSH will handle this");
3170 continue;
3171 }
3172 HRegionInfo regionInfo = regionState.getRegion();
3173 State state = regionState.getState();
3174
3175 switch (state) {
3176 case CLOSED:
3177 invokeAssign(regionInfo);
3178 break;
3179 case PENDING_OPEN:
3180 retrySendRegionOpen(regionState);
3181 break;
3182 case PENDING_CLOSE:
3183 retrySendRegionClose(regionState);
3184 break;
3185 case FAILED_CLOSE:
3186 case FAILED_OPEN:
3187 invokeUnAssign(regionInfo);
3188 break;
3189 default:
3190
3191 }
3192 }
3193 }
3194
3195
3196
3197
3198
3199 private void retrySendRegionOpen(final RegionState regionState) {
3200 this.executorService.submit(
3201 new EventHandler(server, EventType.M_MASTER_RECOVERY) {
3202 @Override
3203 public void process() throws IOException {
3204 HRegionInfo hri = regionState.getRegion();
3205 ServerName serverName = regionState.getServerName();
3206 ReentrantLock lock = locker.acquireLock(hri.getEncodedName());
3207 try {
3208 for (int i = 1; i <= maximumAttempts; i++) {
3209 if (!serverManager.isServerOnline(serverName)
3210 || server.isStopped() || server.isAborted()) {
3211 return;
3212 }
3213 try {
3214 if (!regionState.equals(regionStates.getRegionState(hri))) {
3215 return;
3216 }
3217 List<ServerName> favoredNodes = ServerName.EMPTY_SERVER_LIST;
3218 if (shouldAssignRegionsWithFavoredNodes) {
3219 favoredNodes = ((FavoredNodeLoadBalancer)balancer).getFavoredNodes(hri);
3220 }
3221 RegionOpeningState regionOpenState = serverManager.sendRegionOpen(
3222 serverName, hri, -1, favoredNodes);
3223
3224 if (regionOpenState == RegionOpeningState.FAILED_OPENING) {
3225
3226
3227 LOG.debug("Got failed_opening in retry sendRegionOpen for "
3228 + regionState + ", re-assign it");
3229 invokeAssign(hri, true);
3230 }
3231 return;
3232 } catch (Throwable t) {
3233 if (t instanceof RemoteException) {
3234 t = ((RemoteException) t).unwrapRemoteException();
3235 }
3236
3237 if (t instanceof java.net.SocketTimeoutException
3238 || t instanceof FailedServerException) {
3239 Threads.sleep(100);
3240 continue;
3241 }
3242
3243 LOG.debug("Got exception in retry sendRegionOpen for "
3244 + regionState + ", re-assign it", t);
3245 invokeAssign(hri);
3246 return;
3247 }
3248 }
3249 } finally {
3250 lock.unlock();
3251 }
3252 }
3253 });
3254 }
3255
3256
3257
3258
3259
3260 private void retrySendRegionClose(final RegionState regionState) {
3261 this.executorService.submit(
3262 new EventHandler(server, EventType.M_MASTER_RECOVERY) {
3263 @Override
3264 public void process() throws IOException {
3265 HRegionInfo hri = regionState.getRegion();
3266 ServerName serverName = regionState.getServerName();
3267 ReentrantLock lock = locker.acquireLock(hri.getEncodedName());
3268 try {
3269 for (int i = 1; i <= maximumAttempts; i++) {
3270 if (!serverManager.isServerOnline(serverName)
3271 || server.isStopped() || server.isAborted()) {
3272 return;
3273 }
3274 try {
3275 if (!regionState.equals(regionStates.getRegionState(hri))) {
3276 return;
3277 }
3278 if (!serverManager.sendRegionClose(serverName, hri, -1, null, false)) {
3279
3280 LOG.debug("Got false in retry sendRegionClose for "
3281 + regionState + ", re-close it");
3282 invokeUnAssign(hri);
3283 }
3284 return;
3285 } catch (Throwable t) {
3286 if (t instanceof RemoteException) {
3287 t = ((RemoteException) t).unwrapRemoteException();
3288 }
3289
3290 if (t instanceof java.net.SocketTimeoutException
3291 || t instanceof FailedServerException) {
3292 Threads.sleep(100);
3293 continue;
3294 }
3295 if (!(t instanceof NotServingRegionException
3296 || t instanceof RegionAlreadyInTransitionException)) {
3297
3298
3299
3300 LOG.debug("Got exception in retry sendRegionClose for "
3301 + regionState + ", re-close it", t);
3302 invokeUnAssign(hri);
3303 }
3304 return;
3305 }
3306 }
3307 } finally {
3308 lock.unlock();
3309 }
3310 }
3311 });
3312 }
3313
3314
3315
3316
3317
3318
3319
3320
3321 public void updateRegionsInTransitionMetrics() {
3322 long currentTime = System.currentTimeMillis();
3323 int totalRITs = 0;
3324 int totalRITsOverThreshold = 0;
3325 long oldestRITTime = 0;
3326 int ritThreshold = this.server.getConfiguration().
3327 getInt(HConstants.METRICS_RIT_STUCK_WARNING_THRESHOLD, 60000);
3328 for (RegionState state: regionStates.getRegionsInTransition().values()) {
3329 totalRITs++;
3330 long ritTime = currentTime - state.getStamp();
3331 if (ritTime > ritThreshold) {
3332 totalRITsOverThreshold++;
3333 }
3334 if (oldestRITTime < ritTime) {
3335 oldestRITTime = ritTime;
3336 }
3337 }
3338 if (this.metricsAssignmentManager != null) {
3339 this.metricsAssignmentManager.updateRITOldestAge(oldestRITTime);
3340 this.metricsAssignmentManager.updateRITCount(totalRITs);
3341 this.metricsAssignmentManager.updateRITCountOverThreshold(totalRITsOverThreshold);
3342 }
3343 }
3344
3345
3346
3347
3348 void clearRegionPlan(final HRegionInfo region) {
3349 synchronized (this.regionPlans) {
3350 this.regionPlans.remove(region.getEncodedName());
3351 }
3352 }
3353
3354
3355
3356
3357
3358
3359 public void waitOnRegionToClearRegionsInTransition(final HRegionInfo hri)
3360 throws IOException, InterruptedException {
3361 waitOnRegionToClearRegionsInTransition(hri, -1L);
3362 }
3363
3364
3365
3366
3367
3368
3369
3370
3371 public boolean waitOnRegionToClearRegionsInTransition(final HRegionInfo hri, long timeOut)
3372 throws InterruptedException {
3373 if (!regionStates.isRegionInTransition(hri)) return true;
3374 long end = (timeOut <= 0) ? Long.MAX_VALUE : EnvironmentEdgeManager.currentTime()
3375 + timeOut;
3376
3377
3378 LOG.info("Waiting for " + hri.getEncodedName() +
3379 " to leave regions-in-transition, timeOut=" + timeOut + " ms.");
3380 while (!this.server.isStopped() && regionStates.isRegionInTransition(hri)) {
3381 regionStates.waitForUpdate(100);
3382 if (EnvironmentEdgeManager.currentTime() > end) {
3383 LOG.info("Timed out on waiting for " + hri.getEncodedName() + " to be assigned.");
3384 return false;
3385 }
3386 }
3387 if (this.server.isStopped()) {
3388 LOG.info("Giving up wait on regions in transition because stoppable.isStopped is set");
3389 return false;
3390 }
3391 return true;
3392 }
3393
3394 void invokeAssign(HRegionInfo regionInfo) {
3395 invokeAssign(regionInfo, true);
3396 }
3397
3398 void invokeAssign(HRegionInfo regionInfo, boolean newPlan) {
3399 threadPoolExecutorService.submit(new AssignCallable(this, regionInfo, newPlan));
3400 }
3401
3402 void invokeUnAssign(HRegionInfo regionInfo) {
3403 threadPoolExecutorService.submit(new UnAssignCallable(this, regionInfo));
3404 }
3405
3406 public ServerHostRegion isCarryingMeta(ServerName serverName) {
3407 return isCarryingRegion(serverName, HRegionInfo.FIRST_META_REGIONINFO);
3408 }
3409
3410 public ServerHostRegion isCarryingMetaReplica(ServerName serverName, int replicaId) {
3411 return isCarryingRegion(serverName,
3412 RegionReplicaUtil.getRegionInfoForReplica(HRegionInfo.FIRST_META_REGIONINFO, replicaId));
3413 }
3414
3415 public ServerHostRegion isCarryingMetaReplica(ServerName serverName, HRegionInfo metaHri) {
3416 return isCarryingRegion(serverName, metaHri);
3417 }
3418
3419
3420
3421
3422
3423
3424
3425
3426
3427
3428
3429 private ServerHostRegion isCarryingRegion(ServerName serverName, HRegionInfo hri) {
3430 RegionTransition rt = null;
3431 try {
3432 byte [] data = ZKAssign.getData(watcher, hri.getEncodedName());
3433
3434 rt = data == null? null: RegionTransition.parseFrom(data);
3435 } catch (KeeperException e) {
3436 server.abort("Exception reading unassigned node for region=" + hri.getEncodedName(), e);
3437 } catch (DeserializationException e) {
3438 server.abort("Exception parsing unassigned node for region=" + hri.getEncodedName(), e);
3439 }
3440
3441 ServerName addressFromZK = rt != null? rt.getServerName(): null;
3442 if (addressFromZK != null) {
3443
3444 boolean matchZK = addressFromZK.equals(serverName);
3445 LOG.debug("Checking region=" + hri.getRegionNameAsString() + ", zk server=" + addressFromZK +
3446 " current=" + serverName + ", matches=" + matchZK);
3447 return matchZK ? ServerHostRegion.HOSTING_REGION : ServerHostRegion.NOT_HOSTING_REGION;
3448 }
3449
3450 ServerName addressFromAM = regionStates.getRegionServerOfRegion(hri);
3451 if (LOG.isDebugEnabled()) {
3452 LOG.debug("based on AM, current region=" + hri.getRegionNameAsString() +
3453 " is on server=" + (addressFromAM != null ? addressFromAM : "null") +
3454 " server being checked: " + serverName);
3455 }
3456 if (addressFromAM != null) {
3457 return addressFromAM.equals(serverName) ?
3458 ServerHostRegion.HOSTING_REGION : ServerHostRegion.NOT_HOSTING_REGION;
3459 }
3460
3461 if (hri.isMetaRegion() && RegionReplicaUtil.isDefaultReplica(hri)) {
3462
3463 final ServerName serverNameInZK =
3464 server.getMetaTableLocator().getMetaRegionLocation(this.server.getZooKeeper());
3465 if (LOG.isDebugEnabled()) {
3466 LOG.debug("Based on MetaTableLocator, the META region is on server=" +
3467 (serverNameInZK == null ? "null" : serverNameInZK) +
3468 " server being checked: " + serverName);
3469 }
3470 if (serverNameInZK != null) {
3471 return serverNameInZK.equals(serverName) ?
3472 ServerHostRegion.HOSTING_REGION : ServerHostRegion.NOT_HOSTING_REGION;
3473 }
3474 }
3475
3476
3477 return ServerHostRegion.UNKNOWN;
3478 }
3479
3480
3481
3482
3483
3484
3485 public List<HRegionInfo> cleanOutCrashedServerReferences(final ServerName sn) {
3486
3487 synchronized (this.regionPlans) {
3488 for (Iterator <Map.Entry<String, RegionPlan>> i = this.regionPlans.entrySet().iterator();
3489 i.hasNext();) {
3490 Map.Entry<String, RegionPlan> e = i.next();
3491 ServerName otherSn = e.getValue().getDestination();
3492
3493 if (otherSn != null && otherSn.equals(sn)) {
3494
3495 i.remove();
3496 }
3497 }
3498 }
3499 List<HRegionInfo> regions = regionStates.serverOffline(watcher, sn);
3500 for (Iterator<HRegionInfo> it = regions.iterator(); it.hasNext(); ) {
3501 HRegionInfo hri = it.next();
3502 String encodedName = hri.getEncodedName();
3503
3504
3505 Lock lock = locker.acquireLock(encodedName);
3506 try {
3507 RegionState regionState = regionStates.getRegionTransitionState(encodedName);
3508 if (regionState == null
3509 || (regionState.getServerName() != null && !regionState.isOnServer(sn))
3510 || !(regionState.isFailedClose() || regionState.isOffline()
3511 || regionState.isPendingOpenOrOpening())) {
3512 LOG.info("Skip " + regionState + " since it is not opening/failed_close"
3513 + " on the dead server any more: " + sn);
3514 it.remove();
3515 } else {
3516 try {
3517
3518 ZKAssign.deleteNodeFailSilent(watcher, hri);
3519 } catch (KeeperException ke) {
3520 server.abort("Unexpected ZK exception deleting node " + hri, ke);
3521 }
3522 if (tableStateManager.isTableState(hri.getTable(),
3523 ZooKeeperProtos.Table.State.DISABLED, ZooKeeperProtos.Table.State.DISABLING)) {
3524 regionStates.regionOffline(hri);
3525 it.remove();
3526 continue;
3527 }
3528
3529 regionStates.updateRegionState(hri, State.OFFLINE);
3530 }
3531 } finally {
3532 lock.unlock();
3533 }
3534 }
3535 return regions;
3536 }
3537
3538
3539
3540
3541 public void balance(final RegionPlan plan) {
3542
3543 HRegionInfo hri = plan.getRegionInfo();
3544 TableName tableName = hri.getTable();
3545 if (tableStateManager.isTableState(tableName,
3546 ZooKeeperProtos.Table.State.DISABLED, ZooKeeperProtos.Table.State.DISABLING)) {
3547 LOG.info("Ignored moving region of disabling/disabled table "
3548 + tableName);
3549 return;
3550 }
3551
3552
3553 String encodedName = hri.getEncodedName();
3554 ReentrantLock lock = locker.acquireLock(encodedName);
3555 try {
3556 if (!regionStates.isRegionOnline(hri)) {
3557 RegionState state = regionStates.getRegionState(encodedName);
3558 LOG.info("Ignored moving region not assigned: " + hri + ", "
3559 + (state == null ? "not in region states" : state));
3560 return;
3561 }
3562 synchronized (this.regionPlans) {
3563 this.regionPlans.put(plan.getRegionName(), plan);
3564 }
3565 unassign(hri, false, plan.getDestination());
3566 } finally {
3567 lock.unlock();
3568 }
3569 }
3570
3571 public void stop() {
3572 shutdown();
3573 }
3574
3575
3576
3577
3578 public void shutdown() {
3579
3580 synchronized (zkEventWorkerWaitingList){
3581 zkEventWorkerWaitingList.clear();
3582 }
3583
3584
3585 threadPoolExecutorService.shutdownNow();
3586 zkEventWorkers.shutdownNow();
3587 regionStateStore.stop();
3588 }
3589
3590 protected void setEnabledTable(TableName tableName) {
3591 try {
3592 this.tableStateManager.setTableState(tableName,
3593 ZooKeeperProtos.Table.State.ENABLED);
3594 } catch (CoordinatedStateException e) {
3595
3596 String errorMsg = "Unable to ensure that the table " + tableName
3597 + " will be" + " enabled because of a ZooKeeper issue";
3598 LOG.error(errorMsg);
3599 this.server.abort(errorMsg, e);
3600 }
3601 }
3602
3603
3604
3605
3606
3607
3608
3609 private boolean asyncSetOfflineInZooKeeper(final RegionState state,
3610 final AsyncCallback.StringCallback cb, final ServerName destination) {
3611 if (!state.isClosed() && !state.isOffline()) {
3612 this.server.abort("Unexpected state trying to OFFLINE; " + state,
3613 new IllegalStateException());
3614 return false;
3615 }
3616 regionStates.updateRegionState(state.getRegion(), State.OFFLINE);
3617 try {
3618 ZKAssign.asyncCreateNodeOffline(watcher, state.getRegion(),
3619 destination, cb, state);
3620 } catch (KeeperException e) {
3621 if (e instanceof NodeExistsException) {
3622 LOG.warn("Node for " + state.getRegion() + " already exists");
3623 } else {
3624 server.abort("Unexpected ZK exception creating/setting node OFFLINE", e);
3625 }
3626 return false;
3627 }
3628 return true;
3629 }
3630
3631 private boolean deleteNodeInStates(String encodedName,
3632 String desc, ServerName sn, EventType... types) {
3633 try {
3634 for (EventType et: types) {
3635 if (ZKAssign.deleteNode(watcher, encodedName, et, sn)) {
3636 return true;
3637 }
3638 }
3639 LOG.info("Failed to delete the " + desc + " node for "
3640 + encodedName + ". The node type may not match");
3641 } catch (NoNodeException e) {
3642 if (LOG.isDebugEnabled()) {
3643 LOG.debug("The " + desc + " node for " + encodedName + " already deleted");
3644 }
3645 } catch (KeeperException ke) {
3646 server.abort("Unexpected ZK exception deleting " + desc
3647 + " node for the region " + encodedName, ke);
3648 }
3649 return false;
3650 }
3651
3652 private void deleteMergingNode(String encodedName, ServerName sn) {
3653 deleteNodeInStates(encodedName, "merging", sn, EventType.RS_ZK_REGION_MERGING,
3654 EventType.RS_ZK_REQUEST_REGION_MERGE, EventType.RS_ZK_REGION_MERGED);
3655 }
3656
3657 private void deleteSplittingNode(String encodedName, ServerName sn) {
3658 deleteNodeInStates(encodedName, "splitting", sn, EventType.RS_ZK_REGION_SPLITTING,
3659 EventType.RS_ZK_REQUEST_REGION_SPLIT, EventType.RS_ZK_REGION_SPLIT);
3660 }
3661
3662 @edu.umd.cs.findbugs.annotations.SuppressWarnings(
3663 value="AT_OPERATION_SEQUENCE_ON_CONCURRENT_ABSTRACTION",
3664 justification="Modification of Maps not ATOMIC!!!! FIX!!!")
3665 private void onRegionFailedOpen(
3666 final HRegionInfo hri, final ServerName sn) {
3667 String encodedName = hri.getEncodedName();
3668
3669 AtomicInteger failedOpenCount = failedOpenTracker.get(encodedName);
3670 if (failedOpenCount == null) {
3671 failedOpenCount = new AtomicInteger();
3672
3673
3674
3675
3676 failedOpenTracker.put(encodedName, failedOpenCount);
3677 }
3678 if (failedOpenCount.incrementAndGet() >= maximumAttempts && !hri.isMetaRegion()) {
3679
3680 regionStates.updateRegionState(hri, State.FAILED_OPEN);
3681
3682
3683 failedOpenTracker.remove(encodedName);
3684 } else {
3685 if (hri.isMetaRegion() && failedOpenCount.get() >= maximumAttempts) {
3686
3687
3688 LOG.warn("Failed to open the hbase:meta region " +
3689 hri.getRegionNameAsString() + " after" +
3690 failedOpenCount.get() + " retries. Continue retrying.");
3691 }
3692
3693
3694 RegionState regionState = regionStates.updateRegionState(hri, State.CLOSED);
3695 if (regionState != null) {
3696
3697
3698 if (getTableStateManager().isTableState(hri.getTable(),
3699 ZooKeeperProtos.Table.State.DISABLED, ZooKeeperProtos.Table.State.DISABLING) ||
3700 replicasToClose.contains(hri)) {
3701 offlineDisabledRegion(hri);
3702 return;
3703 }
3704
3705 regionStates.updateRegionState(hri, RegionState.State.CLOSED);
3706
3707 removeClosedRegion(hri);
3708 try {
3709 getRegionPlan(hri, sn, true);
3710 } catch (HBaseIOException e) {
3711 LOG.warn("Failed to get region plan", e);
3712 }
3713 invokeAssign(hri, false);
3714 }
3715 }
3716 }
3717
3718 private void onRegionOpen(final HRegionInfo hri, final ServerName sn, long openSeqNum) {
3719 regionOnline(hri, sn, openSeqNum);
3720 if (useZKForAssignment) {
3721 try {
3722
3723 ZKAssign.deleteNodeFailSilent(watcher, hri);
3724 } catch (KeeperException ke) {
3725 server.abort("Unexpected ZK exception deleting node " + hri, ke);
3726 }
3727 }
3728
3729
3730 failedOpenTracker.remove(hri.getEncodedName());
3731 if (getTableStateManager().isTableState(hri.getTable(),
3732 ZooKeeperProtos.Table.State.DISABLED, ZooKeeperProtos.Table.State.DISABLING)) {
3733 invokeUnAssign(hri);
3734 }
3735 }
3736
3737 private void onRegionClosed(final HRegionInfo hri) {
3738 if (getTableStateManager().isTableState(hri.getTable(),
3739 ZooKeeperProtos.Table.State.DISABLED, ZooKeeperProtos.Table.State.DISABLING) ||
3740 replicasToClose.contains(hri)) {
3741 offlineDisabledRegion(hri);
3742 return;
3743 }
3744 regionStates.updateRegionState(hri, RegionState.State.CLOSED);
3745 sendRegionClosedNotification(hri);
3746
3747 removeClosedRegion(hri);
3748 invokeAssign(hri, false);
3749 }
3750
3751 private String checkInStateForSplit(ServerName sn,
3752 final HRegionInfo p, final HRegionInfo a, final HRegionInfo b) {
3753 final RegionState rs_p = regionStates.getRegionState(p);
3754 RegionState rs_a = regionStates.getRegionState(a);
3755 RegionState rs_b = regionStates.getRegionState(b);
3756 if (!(rs_p.isOpenOrSplittingOnServer(sn)
3757 && (rs_a == null || rs_a.isOpenOrSplittingNewOnServer(sn))
3758 && (rs_b == null || rs_b.isOpenOrSplittingNewOnServer(sn)))) {
3759 return "Not in state good for split";
3760 }
3761 return "";
3762 }
3763
3764 private String onRegionSplitReverted(ServerName sn,
3765 final HRegionInfo p, final HRegionInfo a, final HRegionInfo b) {
3766 String s = checkInStateForSplit(sn, p, a, b);
3767 if (!org.apache.commons.lang.StringUtils.isEmpty(s)) {
3768 return s;
3769 }
3770
3771
3772
3773 regionOnline(p, sn);
3774
3775
3776 RegionState regionStateA = regionStates.getRegionState(a);
3777 RegionState regionStateB = regionStates.getRegionState(b);
3778 if (regionStateA != null) {
3779 regionOffline(a);
3780 }
3781 if (regionStateB != null) {
3782 regionOffline(b);
3783 }
3784
3785 if (getTableStateManager().isTableState(p.getTable(),
3786 ZooKeeperProtos.Table.State.DISABLED, ZooKeeperProtos.Table.State.DISABLING)) {
3787 invokeUnAssign(p);
3788 }
3789 return null;
3790 }
3791
3792 private String onRegionSplit(ServerName sn, TransitionCode code,
3793 final HRegionInfo p, final HRegionInfo a, final HRegionInfo b) {
3794 String s = checkInStateForSplit(sn, p, a, b);
3795 if (!org.apache.commons.lang.StringUtils.isEmpty(s)) {
3796 return s;
3797 }
3798 regionStates.updateRegionState(a, State.SPLITTING_NEW, sn);
3799 regionStates.updateRegionState(b, State.SPLITTING_NEW, sn);
3800 regionStates.updateRegionState(p, State.SPLITTING);
3801
3802 if (code == TransitionCode.SPLIT) {
3803 if (TEST_SKIP_SPLIT_HANDLING) {
3804 return "Skipping split message, TEST_SKIP_SPLIT_HANDLING is set";
3805 }
3806 regionOffline(p, State.SPLIT);
3807 regionOnline(a, sn, 1);
3808 regionOnline(b, sn, 1);
3809
3810
3811 if (getTableStateManager().isTableState(p.getTable(),
3812 ZooKeeperProtos.Table.State.DISABLED, ZooKeeperProtos.Table.State.DISABLING)) {
3813 invokeUnAssign(a);
3814 invokeUnAssign(b);
3815 } else {
3816 Callable<Object> splitReplicasCallable = new Callable<Object>() {
3817 @Override
3818 public Object call() {
3819 doSplittingOfReplicas(p, a, b);
3820 return null;
3821 }
3822 };
3823 threadPoolExecutorService.submit(splitReplicasCallable);
3824 }
3825 } else if (code == TransitionCode.SPLIT_PONR) {
3826 try {
3827 regionStates.splitRegion(p, a, b, sn);
3828 } catch (IOException ioe) {
3829 LOG.info("Failed to record split region " + p.getShortNameToLog());
3830 return "Failed to record the splitting in meta";
3831 }
3832 }
3833 return null;
3834 }
3835
3836 private String onRegionMerge(ServerName sn, TransitionCode code,
3837 final HRegionInfo p, final HRegionInfo a, final HRegionInfo b) {
3838 RegionState rs_p = regionStates.getRegionState(p);
3839 RegionState rs_a = regionStates.getRegionState(a);
3840 RegionState rs_b = regionStates.getRegionState(b);
3841 if (!(rs_a.isOpenOrMergingOnServer(sn) && rs_b.isOpenOrMergingOnServer(sn)
3842 && (rs_p == null || rs_p.isOpenOrMergingNewOnServer(sn)))) {
3843 return "Not in state good for merge";
3844 }
3845 regionStates.updateRegionState(a, State.MERGING);
3846 regionStates.updateRegionState(b, State.MERGING);
3847 regionStates.updateRegionState(p, State.MERGING_NEW, sn);
3848
3849 String encodedName = p.getEncodedName();
3850 if (code == TransitionCode.READY_TO_MERGE) {
3851 mergingRegions.put(encodedName,
3852 new PairOfSameType<HRegionInfo>(a, b));
3853 } else if (code == TransitionCode.MERGED) {
3854 mergingRegions.remove(encodedName);
3855 regionOffline(a, State.MERGED);
3856 regionOffline(b, State.MERGED);
3857 regionOnline(p, sn, 1);
3858
3859
3860 if (getTableStateManager().isTableState(p.getTable(),
3861 ZooKeeperProtos.Table.State.DISABLED, ZooKeeperProtos.Table.State.DISABLING)) {
3862 invokeUnAssign(p);
3863 } else {
3864 Callable<Object> mergeReplicasCallable = new Callable<Object>() {
3865 @Override
3866 public Object call() {
3867 doMergingOfReplicas(p, a, b);
3868 return null;
3869 }
3870 };
3871 threadPoolExecutorService.submit(mergeReplicasCallable);
3872 }
3873 } else if (code == TransitionCode.MERGE_PONR) {
3874 try {
3875 regionStates.mergeRegions(p, a, b, sn);
3876 } catch (IOException ioe) {
3877 LOG.info("Failed to record merged region " + p.getShortNameToLog());
3878 return "Failed to record the merging in meta";
3879 }
3880 }
3881 return null;
3882 }
3883
3884 private String onRegionMergeReverted(ServerName sn, TransitionCode code,
3885 final HRegionInfo p, final HRegionInfo a, final HRegionInfo b) {
3886 RegionState rs_p = regionStates.getRegionState(p);
3887 String encodedName = p.getEncodedName();
3888 mergingRegions.remove(encodedName);
3889
3890
3891
3892 regionOnline(a, sn);
3893 regionOnline(b, sn);
3894
3895
3896 if (rs_p != null) {
3897 regionOffline(p);
3898 }
3899
3900 if (getTableStateManager().isTableState(p.getTable(),
3901 ZooKeeperProtos.Table.State.DISABLED, ZooKeeperProtos.Table.State.DISABLING)) {
3902 invokeUnAssign(a);
3903 invokeUnAssign(b);
3904 }
3905
3906 return null;
3907 }
3908
3909
3910
3911
3912
3913 private boolean handleRegionMerging(final RegionTransition rt, final String encodedName,
3914 final String prettyPrintedRegionName, final ServerName sn) {
3915 if (!serverManager.isServerOnline(sn)) {
3916 LOG.warn("Dropped merging! ServerName=" + sn + " unknown.");
3917 return false;
3918 }
3919 byte [] payloadOfMerging = rt.getPayload();
3920 List<HRegionInfo> mergingRegions;
3921 try {
3922 mergingRegions = HRegionInfo.parseDelimitedFrom(
3923 payloadOfMerging, 0, payloadOfMerging.length);
3924 } catch (IOException e) {
3925 LOG.error("Dropped merging! Failed reading " + rt.getEventType()
3926 + " payload for " + prettyPrintedRegionName);
3927 return false;
3928 }
3929 assert mergingRegions.size() == 3;
3930 HRegionInfo p = mergingRegions.get(0);
3931 HRegionInfo hri_a = mergingRegions.get(1);
3932 HRegionInfo hri_b = mergingRegions.get(2);
3933
3934 RegionState rs_p = regionStates.getRegionState(p);
3935 RegionState rs_a = regionStates.getRegionState(hri_a);
3936 RegionState rs_b = regionStates.getRegionState(hri_b);
3937
3938 if (!((rs_a == null || rs_a.isOpenOrMergingOnServer(sn))
3939 && (rs_b == null || rs_b.isOpenOrMergingOnServer(sn))
3940 && (rs_p == null || rs_p.isOpenOrMergingNewOnServer(sn)))) {
3941 LOG.warn("Dropped merging! Not in state good for MERGING; rs_p="
3942 + rs_p + ", rs_a=" + rs_a + ", rs_b=" + rs_b);
3943 return false;
3944 }
3945
3946 EventType et = rt.getEventType();
3947 if (et == EventType.RS_ZK_REQUEST_REGION_MERGE) {
3948 try {
3949 RegionMergeCoordination.RegionMergeDetails std =
3950 ((BaseCoordinatedStateManager) server.getCoordinatedStateManager())
3951 .getRegionMergeCoordination().getDefaultDetails();
3952 ((BaseCoordinatedStateManager) server.getCoordinatedStateManager())
3953 .getRegionMergeCoordination().processRegionMergeRequest(p, hri_a, hri_b, sn, std);
3954 if (((ZkRegionMergeCoordination.ZkRegionMergeDetails) std).getZnodeVersion() == -1) {
3955 byte[] data = ZKAssign.getData(watcher, encodedName);
3956 EventType currentType = null;
3957 if (data != null) {
3958 RegionTransition newRt = RegionTransition.parseFrom(data);
3959 currentType = newRt.getEventType();
3960 }
3961 if (currentType == null || (currentType != EventType.RS_ZK_REGION_MERGED
3962 && currentType != EventType.RS_ZK_REGION_MERGING)) {
3963 LOG.warn("Failed to transition pending_merge node "
3964 + encodedName + " to merging, it's now " + currentType);
3965 return false;
3966 }
3967 }
3968 } catch (Exception e) {
3969 LOG.warn("Failed to transition pending_merge node "
3970 + encodedName + " to merging", e);
3971 return false;
3972 }
3973 }
3974
3975 synchronized (regionStates) {
3976 regionStates.updateRegionState(hri_a, State.MERGING);
3977 regionStates.updateRegionState(hri_b, State.MERGING);
3978 regionStates.updateRegionState(p, State.MERGING_NEW, sn);
3979
3980 if (et != EventType.RS_ZK_REGION_MERGED) {
3981 this.mergingRegions.put(encodedName,
3982 new PairOfSameType<HRegionInfo>(hri_a, hri_b));
3983 } else {
3984 this.mergingRegions.remove(encodedName);
3985 regionOffline(hri_a, State.MERGED);
3986 regionOffline(hri_b, State.MERGED);
3987 regionOnline(p, sn);
3988 }
3989 }
3990
3991 if (et == EventType.RS_ZK_REGION_MERGED) {
3992 doMergingOfReplicas(p, hri_a, hri_b);
3993 LOG.debug("Handling MERGED event for " + encodedName + "; deleting node");
3994
3995 try {
3996 boolean successful = false;
3997 while (!successful) {
3998
3999
4000 successful = ZKAssign.deleteNode(watcher, encodedName,
4001 EventType.RS_ZK_REGION_MERGED, sn);
4002 }
4003 } catch (KeeperException e) {
4004 if (e instanceof NoNodeException) {
4005 String znodePath = ZKUtil.joinZNode(watcher.splitLogZNode, encodedName);
4006 LOG.debug("The znode " + znodePath + " does not exist. May be deleted already.");
4007 } else {
4008 server.abort("Error deleting MERGED node " + encodedName, e);
4009 }
4010 }
4011 LOG.info("Handled MERGED event; merged=" + p.getRegionNameAsString()
4012 + ", region_a=" + hri_a.getRegionNameAsString() + ", region_b="
4013 + hri_b.getRegionNameAsString() + ", on " + sn);
4014
4015
4016 if (tableStateManager.isTableState(p.getTable(),
4017 ZooKeeperProtos.Table.State.DISABLED, ZooKeeperProtos.Table.State.DISABLING)) {
4018 unassign(p);
4019 }
4020 }
4021 return true;
4022 }
4023
4024
4025
4026
4027 private boolean handleRegionSplitting(final RegionTransition rt, final String encodedName,
4028 final String prettyPrintedRegionName, final ServerName sn) {
4029 if (!serverManager.isServerOnline(sn)) {
4030 LOG.warn("Dropped splitting! ServerName=" + sn + " unknown.");
4031 return false;
4032 }
4033 byte [] payloadOfSplitting = rt.getPayload();
4034 List<HRegionInfo> splittingRegions;
4035 try {
4036 splittingRegions = HRegionInfo.parseDelimitedFrom(
4037 payloadOfSplitting, 0, payloadOfSplitting.length);
4038 } catch (IOException e) {
4039 LOG.error("Dropped splitting! Failed reading " + rt.getEventType()
4040 + " payload for " + prettyPrintedRegionName);
4041 return false;
4042 }
4043 assert splittingRegions.size() == 2;
4044 HRegionInfo hri_a = splittingRegions.get(0);
4045 HRegionInfo hri_b = splittingRegions.get(1);
4046
4047 RegionState rs_p = regionStates.getRegionState(encodedName);
4048 RegionState rs_a = regionStates.getRegionState(hri_a);
4049 RegionState rs_b = regionStates.getRegionState(hri_b);
4050
4051 if (!((rs_p == null || rs_p.isOpenOrSplittingOnServer(sn))
4052 && (rs_a == null || rs_a.isOpenOrSplittingNewOnServer(sn))
4053 && (rs_b == null || rs_b.isOpenOrSplittingNewOnServer(sn)))) {
4054 LOG.warn("Dropped splitting! Not in state good for SPLITTING; rs_p="
4055 + rs_p + ", rs_a=" + rs_a + ", rs_b=" + rs_b);
4056 return false;
4057 }
4058
4059 if (rs_p == null) {
4060
4061 rs_p = regionStates.updateRegionState(rt, State.OPEN);
4062 if (rs_p == null) {
4063 LOG.warn("Received splitting for region " + prettyPrintedRegionName
4064 + " from server " + sn + " but it doesn't exist anymore,"
4065 + " probably already processed its split");
4066 return false;
4067 }
4068 regionStates.regionOnline(rs_p.getRegion(), sn);
4069 }
4070
4071 HRegionInfo p = rs_p.getRegion();
4072 EventType et = rt.getEventType();
4073 if (et == EventType.RS_ZK_REQUEST_REGION_SPLIT) {
4074 try {
4075 SplitTransactionDetails std =
4076 ((BaseCoordinatedStateManager) server.getCoordinatedStateManager())
4077 .getSplitTransactionCoordination().getDefaultDetails();
4078 if (((BaseCoordinatedStateManager) server.getCoordinatedStateManager())
4079 .getSplitTransactionCoordination().processTransition(p, hri_a, hri_b, sn, std) == -1) {
4080 byte[] data = ZKAssign.getData(watcher, encodedName);
4081 EventType currentType = null;
4082 if (data != null) {
4083 RegionTransition newRt = RegionTransition.parseFrom(data);
4084 currentType = newRt.getEventType();
4085 }
4086 if (currentType == null
4087 || (currentType != EventType.RS_ZK_REGION_SPLIT && currentType != EventType.RS_ZK_REGION_SPLITTING)) {
4088 LOG.warn("Failed to transition pending_split node " + encodedName
4089 + " to splitting, it's now " + currentType);
4090 return false;
4091 }
4092 }
4093 } catch (Exception e) {
4094 LOG.warn("Failed to transition pending_split node " + encodedName + " to splitting", e);
4095 return false;
4096 }
4097 }
4098
4099 synchronized (regionStates) {
4100 splitRegions.put(p, new PairOfSameType<HRegionInfo>(hri_a, hri_b));
4101 regionStates.updateRegionState(hri_a, State.SPLITTING_NEW, sn);
4102 regionStates.updateRegionState(hri_b, State.SPLITTING_NEW, sn);
4103 regionStates.updateRegionState(rt, State.SPLITTING);
4104
4105
4106
4107 if (TEST_SKIP_SPLIT_HANDLING) {
4108 LOG.warn("Skipping split message, TEST_SKIP_SPLIT_HANDLING is set");
4109 return true;
4110 }
4111
4112 if (et == EventType.RS_ZK_REGION_SPLIT) {
4113 regionOffline(p, State.SPLIT);
4114 regionOnline(hri_a, sn);
4115 regionOnline(hri_b, sn);
4116 splitRegions.remove(p);
4117 }
4118 }
4119
4120 if (et == EventType.RS_ZK_REGION_SPLIT) {
4121
4122 doSplittingOfReplicas(rs_p.getRegion(), hri_a, hri_b);
4123 LOG.debug("Handling SPLIT event for " + encodedName + "; deleting node");
4124
4125 try {
4126 boolean successful = false;
4127 while (!successful) {
4128
4129
4130 successful = ZKAssign.deleteNode(watcher, encodedName,
4131 EventType.RS_ZK_REGION_SPLIT, sn);
4132 }
4133 } catch (KeeperException e) {
4134 if (e instanceof NoNodeException) {
4135 String znodePath = ZKUtil.joinZNode(watcher.splitLogZNode, encodedName);
4136 LOG.debug("The znode " + znodePath + " does not exist. May be deleted already.");
4137 } else {
4138 server.abort("Error deleting SPLIT node " + encodedName, e);
4139 }
4140 }
4141 LOG.info("Handled SPLIT event; parent=" + p.getRegionNameAsString()
4142 + ", daughter a=" + hri_a.getRegionNameAsString() + ", daughter b="
4143 + hri_b.getRegionNameAsString() + ", on " + sn);
4144
4145
4146 if (tableStateManager.isTableState(p.getTable(),
4147 ZooKeeperProtos.Table.State.DISABLED, ZooKeeperProtos.Table.State.DISABLING)) {
4148 unassign(hri_a);
4149 unassign(hri_b);
4150 }
4151 }
4152 return true;
4153 }
4154
4155 private void doMergingOfReplicas(HRegionInfo mergedHri, final HRegionInfo hri_a,
4156 final HRegionInfo hri_b) {
4157
4158
4159 List<HRegionInfo> unmergedRegions = new ArrayList<HRegionInfo>();
4160 unmergedRegions.add(hri_a);
4161 unmergedRegions.add(hri_b);
4162 Map<ServerName, List<HRegionInfo>> map = regionStates.getRegionAssignments(unmergedRegions);
4163 Collection<List<HRegionInfo>> c = map.values();
4164 for (List<HRegionInfo> l : c) {
4165 for (HRegionInfo h : l) {
4166 if (!RegionReplicaUtil.isDefaultReplica(h)) {
4167 LOG.debug("Unassigning un-merged replica " + h);
4168 unassign(h);
4169 }
4170 }
4171 }
4172 int numReplicas = 1;
4173 try {
4174 numReplicas = server.getTableDescriptors().get(mergedHri.getTable()).
4175 getRegionReplication();
4176 } catch (IOException e) {
4177 LOG.warn("Couldn't get the replication attribute of the table " + mergedHri.getTable() +
4178 " due to " + e.getMessage() + ". The assignment of replicas for the merged region " +
4179 "will not be done");
4180 }
4181 List<HRegionInfo> regions = new ArrayList<HRegionInfo>();
4182 for (int i = 1; i < numReplicas; i++) {
4183 regions.add(RegionReplicaUtil.getRegionInfoForReplica(mergedHri, i));
4184 }
4185 try {
4186 assign(regions);
4187 } catch (IOException ioe) {
4188 LOG.warn("Couldn't assign all replica(s) of region " + mergedHri + " because of " +
4189 ioe.getMessage());
4190 } catch (InterruptedException ie) {
4191 LOG.warn("Couldn't assign all replica(s) of region " + mergedHri+ " because of " +
4192 ie.getMessage());
4193 }
4194 }
4195
4196 private void doSplittingOfReplicas(final HRegionInfo parentHri, final HRegionInfo hri_a,
4197 final HRegionInfo hri_b) {
4198
4199
4200
4201 int numReplicas = 1;
4202 try {
4203 numReplicas = server.getTableDescriptors().get(parentHri.getTable()).
4204 getRegionReplication();
4205 } catch (IOException e) {
4206 LOG.warn("Couldn't get the replication attribute of the table " + parentHri.getTable() +
4207 " due to " + e.getMessage() + ". The assignment of daughter replicas " +
4208 "replicas will not be done");
4209 }
4210
4211 List<HRegionInfo> parentRegion = new ArrayList<HRegionInfo>();
4212 parentRegion.add(parentHri);
4213 Map<ServerName, List<HRegionInfo>> currentAssign =
4214 regionStates.getRegionAssignments(parentRegion);
4215 Collection<List<HRegionInfo>> c = currentAssign.values();
4216 for (List<HRegionInfo> l : c) {
4217 for (HRegionInfo h : l) {
4218 if (!RegionReplicaUtil.isDefaultReplica(h)) {
4219 LOG.debug("Unassigning parent's replica " + h);
4220 unassign(h);
4221 }
4222 }
4223 }
4224
4225 Map<HRegionInfo, ServerName> map = new HashMap<HRegionInfo, ServerName>();
4226 for (int i = 1; i < numReplicas; i++) {
4227 prepareDaughterReplicaForAssignment(hri_a, parentHri, i, map);
4228 prepareDaughterReplicaForAssignment(hri_b, parentHri, i, map);
4229 }
4230 try {
4231 assign(map);
4232 } catch (IOException e) {
4233 LOG.warn("Caught exception " + e + " while trying to assign replica(s) of daughter(s)");
4234 } catch (InterruptedException e) {
4235 LOG.warn("Caught exception " + e + " while trying to assign replica(s) of daughter(s)");
4236 }
4237 }
4238
4239 private void prepareDaughterReplicaForAssignment(HRegionInfo daughterHri, HRegionInfo parentHri,
4240 int replicaId, Map<HRegionInfo, ServerName> map) {
4241 HRegionInfo parentReplica = RegionReplicaUtil.getRegionInfoForReplica(parentHri, replicaId);
4242 HRegionInfo daughterReplica = RegionReplicaUtil.getRegionInfoForReplica(daughterHri,
4243 replicaId);
4244 LOG.debug("Created replica region for daughter " + daughterReplica);
4245 ServerName sn;
4246 if ((sn = regionStates.getRegionServerOfRegion(parentReplica)) != null) {
4247 map.put(daughterReplica, sn);
4248 } else {
4249 List<ServerName> servers = serverManager.getOnlineServersList();
4250 sn = servers.get((new Random(System.currentTimeMillis())).nextInt(servers.size()));
4251 map.put(daughterReplica, sn);
4252 }
4253 }
4254
4255 public Set<HRegionInfo> getReplicasToClose() {
4256 return replicasToClose;
4257 }
4258
4259
4260
4261
4262
4263
4264 private void regionOffline(final HRegionInfo regionInfo, final State state) {
4265 regionStates.regionOffline(regionInfo, state);
4266 removeClosedRegion(regionInfo);
4267
4268 clearRegionPlan(regionInfo);
4269 balancer.regionOffline(regionInfo);
4270
4271
4272 sendRegionClosedNotification(regionInfo);
4273
4274 if (state != null && state.equals(State.SPLIT)) {
4275 Collection<HRegionInfo> c = new ArrayList<HRegionInfo>(1);
4276 c.add(regionInfo);
4277 Map<ServerName, List<HRegionInfo>> map = regionStates.getRegionAssignments(c);
4278 Collection<List<HRegionInfo>> allReplicas = map.values();
4279 for (List<HRegionInfo> list : allReplicas) {
4280 replicasToClose.addAll(list);
4281 }
4282 }
4283 else if (state != null && state.equals(State.MERGED)) {
4284 Collection<HRegionInfo> c = new ArrayList<HRegionInfo>(1);
4285 c.add(regionInfo);
4286 Map<ServerName, List<HRegionInfo>> map = regionStates.getRegionAssignments(c);
4287 Collection<List<HRegionInfo>> allReplicas = map.values();
4288 for (List<HRegionInfo> list : allReplicas) {
4289 replicasToClose.addAll(list);
4290 }
4291 }
4292 }
4293
4294 private void sendRegionOpenedNotification(final HRegionInfo regionInfo,
4295 final ServerName serverName) {
4296 if (!this.listeners.isEmpty()) {
4297 for (AssignmentListener listener : this.listeners) {
4298 listener.regionOpened(regionInfo, serverName);
4299 }
4300 }
4301 }
4302
4303 private void sendRegionClosedNotification(final HRegionInfo regionInfo) {
4304 if (!this.listeners.isEmpty()) {
4305 for (AssignmentListener listener : this.listeners) {
4306 listener.regionClosed(regionInfo);
4307 }
4308 }
4309 }
4310
4311
4312
4313
4314
4315
4316
4317
4318
4319
4320
4321
4322
4323
4324
4325
4326
4327
4328
4329
4330
4331
4332
4333
4334
4335
4336
4337
4338
4339
4340
4341
4342
4343
4344
4345
4346
4347
4348
4349
4350
4351
4352
4353
4354
4355 protected String onRegionTransition(final ServerName serverName,
4356 final RegionStateTransition transition) {
4357 TransitionCode code = transition.getTransitionCode();
4358 HRegionInfo hri = HRegionInfo.convert(transition.getRegionInfo(0));
4359 RegionState current = regionStates.getRegionState(hri);
4360 if (LOG.isDebugEnabled()) {
4361 LOG.debug("Got transition " + code + " for "
4362 + (current != null ? current.toString() : hri.getShortNameToLog())
4363 + " from " + serverName);
4364 }
4365 String errorMsg = null;
4366 switch (code) {
4367 case OPENED:
4368 if (current != null && current.isOpened() && current.isOnServer(serverName)) {
4369 LOG.info("Region " + hri.getShortNameToLog() + " is already " + current.getState() + " on "
4370 + serverName);
4371 break;
4372 }
4373 case FAILED_OPEN:
4374 if (current == null
4375 || !current.isPendingOpenOrOpeningOnServer(serverName)) {
4376 errorMsg = hri.getShortNameToLog()
4377 + " is not pending open on " + serverName;
4378 } else if (code == TransitionCode.FAILED_OPEN) {
4379 onRegionFailedOpen(hri, serverName);
4380 } else {
4381 long openSeqNum = HConstants.NO_SEQNUM;
4382 if (transition.hasOpenSeqNum()) {
4383 openSeqNum = transition.getOpenSeqNum();
4384 }
4385 if (openSeqNum < 0) {
4386 errorMsg = "Newly opened region has invalid open seq num " + openSeqNum;
4387 } else {
4388 onRegionOpen(hri, serverName, openSeqNum);
4389 }
4390 }
4391 break;
4392
4393 case CLOSED:
4394 if (current == null
4395 || !current.isPendingCloseOrClosingOnServer(serverName)) {
4396 errorMsg = hri.getShortNameToLog()
4397 + " is not pending close on " + serverName;
4398 } else {
4399 onRegionClosed(hri);
4400 }
4401 break;
4402
4403 case READY_TO_SPLIT:
4404 try {
4405 regionStateListener.onRegionSplit(hri);
4406 if (!((HMaster)server).getSplitOrMergeTracker().isSplitOrMergeEnabled(
4407 Admin.MasterSwitchType.SPLIT)) {
4408 errorMsg = "split switch is off!";
4409 }
4410 } catch (IOException exp) {
4411 errorMsg = StringUtils.stringifyException(exp);
4412 }
4413 break;
4414 case SPLIT_PONR:
4415 case SPLIT:
4416 errorMsg =
4417 onRegionSplit(serverName, code, hri, HRegionInfo.convert(transition.getRegionInfo(1)),
4418 HRegionInfo.convert(transition.getRegionInfo(2)));
4419 break;
4420
4421 case SPLIT_REVERTED:
4422 errorMsg =
4423 onRegionSplitReverted(serverName, hri,
4424 HRegionInfo.convert(transition.getRegionInfo(1)),
4425 HRegionInfo.convert(transition.getRegionInfo(2)));
4426 if (org.apache.commons.lang.StringUtils.isEmpty(errorMsg)) {
4427 try {
4428 regionStateListener.onRegionSplitReverted(hri);
4429 } catch (IOException exp) {
4430 LOG.warn(StringUtils.stringifyException(exp));
4431 }
4432 }
4433 break;
4434 case READY_TO_MERGE:
4435 if (!((HMaster)server).getSplitOrMergeTracker().isSplitOrMergeEnabled(
4436 Admin.MasterSwitchType.MERGE)) {
4437 errorMsg = "merge switch is off!";
4438 }
4439 break;
4440 case MERGE_PONR:
4441 case MERGED:
4442 errorMsg = onRegionMerge(serverName, code, hri,
4443 HRegionInfo.convert(transition.getRegionInfo(1)),
4444 HRegionInfo.convert(transition.getRegionInfo(2)));
4445 if (code == TransitionCode.MERGED && org.apache.commons.lang.StringUtils.isEmpty(errorMsg)) {
4446 try {
4447 regionStateListener.onRegionMerged(hri);
4448 } catch (IOException exp) {
4449 errorMsg = StringUtils.stringifyException(exp);
4450 }
4451 }
4452 break;
4453 case MERGE_REVERTED:
4454 errorMsg = onRegionMergeReverted(serverName, code, hri,
4455 HRegionInfo.convert(transition.getRegionInfo(1)),
4456 HRegionInfo.convert(transition.getRegionInfo(2)));
4457 break;
4458
4459 default:
4460 errorMsg = "Unexpected transition code " + code;
4461 }
4462 if (errorMsg != null) {
4463 LOG.error("Failed to transtion region from " + current + " to "
4464 + code + " by " + serverName + ": " + errorMsg);
4465 }
4466 return errorMsg;
4467 }
4468
4469
4470
4471
4472 public LoadBalancer getBalancer() {
4473 return this.balancer;
4474 }
4475
4476 public Map<ServerName, List<HRegionInfo>>
4477 getSnapShotOfAssignment(Collection<HRegionInfo> infos) {
4478 return getRegionStates().getRegionAssignments(infos);
4479 }
4480
4481 void setRegionStateListener(RegionStateListener listener) {
4482 this.regionStateListener = listener;
4483 }
4484 }