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