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