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