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 static org.apache.hadoop.hbase.SplitLogCounters.tot_mgr_wait_for_zk_delete;
22 import static org.apache.hadoop.hbase.SplitLogCounters.tot_wkr_final_transition_failed;
23 import static org.apache.hadoop.hbase.SplitLogCounters.tot_wkr_preempt_task;
24 import static org.apache.hadoop.hbase.SplitLogCounters.tot_wkr_task_acquired;
25 import static org.apache.hadoop.hbase.SplitLogCounters.tot_wkr_task_done;
26 import static org.apache.hadoop.hbase.SplitLogCounters.tot_wkr_task_err;
27 import static org.apache.hadoop.hbase.SplitLogCounters.tot_wkr_task_resigned;
28 import static org.junit.Assert.assertEquals;
29 import static org.junit.Assert.assertFalse;
30 import static org.junit.Assert.assertTrue;
31 import static org.junit.Assert.fail;
32
33 import java.io.IOException;
34 import java.util.ArrayList;
35 import java.util.Arrays;
36 import java.util.HashSet;
37 import java.util.Iterator;
38 import java.util.LinkedList;
39 import java.util.List;
40 import java.util.NavigableSet;
41 import java.util.Set;
42 import java.util.concurrent.ExecutorService;
43 import java.util.concurrent.Executors;
44 import java.util.concurrent.Future;
45 import java.util.concurrent.TimeUnit;
46 import java.util.concurrent.TimeoutException;
47 import java.util.concurrent.atomic.AtomicLong;
48
49 import org.apache.commons.logging.Log;
50 import org.apache.commons.logging.LogFactory;
51 import org.apache.hadoop.conf.Configuration;
52 import org.apache.hadoop.fs.FSDataOutputStream;
53 import org.apache.hadoop.fs.FileStatus;
54 import org.apache.hadoop.fs.FileSystem;
55 import org.apache.hadoop.fs.Path;
56 import org.apache.hadoop.fs.PathFilter;
57 import org.apache.hadoop.hbase.HBaseConfiguration;
58 import org.apache.hadoop.hbase.HBaseTestingUtility;
59 import org.apache.hadoop.hbase.HColumnDescriptor;
60 import org.apache.hadoop.hbase.HConstants;
61 import org.apache.hadoop.hbase.HRegionInfo;
62 import org.apache.hadoop.hbase.HTableDescriptor;
63 import org.apache.hadoop.hbase.KeyValue;
64 import org.apache.hadoop.hbase.MiniHBaseCluster;
65 import org.apache.hadoop.hbase.NamespaceDescriptor;
66 import org.apache.hadoop.hbase.ServerName;
67 import org.apache.hadoop.hbase.SplitLogCounters;
68 import org.apache.hadoop.hbase.TableName;
69 import org.apache.hadoop.hbase.Waiter;
70 import org.apache.hadoop.hbase.client.ClusterConnection;
71 import org.apache.hadoop.hbase.client.ConnectionUtils;
72 import org.apache.hadoop.hbase.client.Delete;
73 import org.apache.hadoop.hbase.client.Get;
74 import org.apache.hadoop.hbase.client.Increment;
75 import org.apache.hadoop.hbase.client.NonceGenerator;
76 import org.apache.hadoop.hbase.client.PerClientRandomNonceGenerator;
77 import org.apache.hadoop.hbase.client.Put;
78 import org.apache.hadoop.hbase.client.RegionLocator;
79 import org.apache.hadoop.hbase.client.Result;
80 import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException;
81 import org.apache.hadoop.hbase.client.Table;
82 import org.apache.hadoop.hbase.coordination.BaseCoordinatedStateManager;
83 import org.apache.hadoop.hbase.coordination.ZKSplitLogManagerCoordination;
84 import org.apache.hadoop.hbase.exceptions.OperationConflictException;
85 import org.apache.hadoop.hbase.exceptions.RegionInRecoveryException;
86 import org.apache.hadoop.hbase.ipc.ServerNotRunningYetException;
87 import org.apache.hadoop.hbase.master.SplitLogManager.TaskBatch;
88 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
89 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.GetRegionInfoResponse.CompactionState;
90 import org.apache.hadoop.hbase.regionserver.HRegion;
91 import org.apache.hadoop.hbase.regionserver.HRegionServer;
92 import org.apache.hadoop.hbase.regionserver.Region;
93 import org.apache.hadoop.hbase.regionserver.wal.HLogKey;
94 import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
95 import org.apache.hadoop.hbase.testclassification.LargeTests;
96 import org.apache.hadoop.hbase.util.Bytes;
97 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
98 import org.apache.hadoop.hbase.util.FSUtils;
99 import org.apache.hadoop.hbase.util.JVMClusterUtil.MasterThread;
100 import org.apache.hadoop.hbase.util.JVMClusterUtil.RegionServerThread;
101 import org.apache.hadoop.hbase.util.Threads;
102 import org.apache.hadoop.hbase.wal.DefaultWALProvider;
103 import org.apache.hadoop.hbase.wal.WAL;
104 import org.apache.hadoop.hbase.wal.WALFactory;
105 import org.apache.hadoop.hbase.wal.WALSplitter;
106 import org.apache.hadoop.hbase.zookeeper.MiniZooKeeperCluster;
107 import org.apache.hadoop.hbase.zookeeper.ZKUtil;
108 import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
109 import org.apache.hadoop.hdfs.MiniDFSCluster;
110 import org.junit.After;
111 import org.junit.AfterClass;
112 import org.junit.Assert;
113 import org.junit.Before;
114 import org.junit.BeforeClass;
115 import org.junit.Ignore;
116 import org.junit.Test;
117 import org.junit.experimental.categories.Category;
118
119 @Category({LargeTests.class})
120 @SuppressWarnings("deprecation")
121 public class TestDistributedLogSplitting {
122 private static final Log LOG = LogFactory.getLog(TestSplitLogManager.class);
123 static {
124
125
126
127
128
129
130 System.setProperty("hbase.tests.use.shortcircuit.reads", "false");
131
132 }
133
134
135 static final int NUM_MASTERS = 2;
136 static final int NUM_RS = 6;
137
138 MiniHBaseCluster cluster;
139 HMaster master;
140 Configuration conf;
141 static Configuration originalConf;
142 static HBaseTestingUtility TEST_UTIL;
143 static MiniDFSCluster dfsCluster;
144 static MiniZooKeeperCluster zkCluster;
145
146 @BeforeClass
147 public static void setup() throws Exception {
148 TEST_UTIL = new HBaseTestingUtility(HBaseConfiguration.create());
149 dfsCluster = TEST_UTIL.startMiniDFSCluster(1);
150 zkCluster = TEST_UTIL.startMiniZKCluster();
151 originalConf = TEST_UTIL.getConfiguration();
152 }
153
154 @AfterClass
155 public static void tearDown() throws IOException {
156 TEST_UTIL.shutdownMiniZKCluster();
157 TEST_UTIL.shutdownMiniDFSCluster();
158 TEST_UTIL.shutdownMiniHBaseCluster();
159 }
160
161 private void startCluster(int num_rs) throws Exception {
162 SplitLogCounters.resetCounters();
163 LOG.info("Starting cluster");
164 conf.getLong("hbase.splitlog.max.resubmit", 0);
165
166 conf.setInt("zookeeper.recovery.retry", 0);
167 conf.setInt(HConstants.REGIONSERVER_INFO_PORT, -1);
168 conf.setFloat(HConstants.LOAD_BALANCER_SLOP_KEY, (float) 100.0);
169 conf.setInt("hbase.regionserver.wal.max.splitters", 3);
170 conf.setInt(HConstants.REGION_SERVER_HIGH_PRIORITY_HANDLER_COUNT, 40);
171 TEST_UTIL.shutdownMiniHBaseCluster();
172 TEST_UTIL = new HBaseTestingUtility(conf);
173 TEST_UTIL.setDFSCluster(dfsCluster);
174 TEST_UTIL.setZkCluster(zkCluster);
175 TEST_UTIL.startMiniHBaseCluster(NUM_MASTERS, num_rs);
176 cluster = TEST_UTIL.getHBaseCluster();
177 LOG.info("Waiting for active/ready master");
178 cluster.waitForActiveAndReadyMaster();
179 master = cluster.getMaster();
180 while (cluster.getLiveRegionServerThreads().size() < num_rs) {
181 Threads.sleep(10);
182 }
183 }
184
185 @Before
186 public void before() throws Exception {
187
188 conf = HBaseConfiguration.create(originalConf);
189 }
190
191 @After
192 public void after() throws Exception {
193 try {
194 if (TEST_UTIL.getHBaseCluster() != null) {
195 for (MasterThread mt : TEST_UTIL.getHBaseCluster().getLiveMasterThreads()) {
196 mt.getMaster().abort("closing...", null);
197 }
198 }
199 TEST_UTIL.shutdownMiniHBaseCluster();
200 } finally {
201 TEST_UTIL.getTestFileSystem().delete(FSUtils.getRootDir(TEST_UTIL.getConfiguration()), true);
202 ZKUtil.deleteNodeRecursively(TEST_UTIL.getZooKeeperWatcher(), "/hbase");
203 }
204 }
205
206 @Test (timeout=300000)
207 public void testRecoveredEdits() throws Exception {
208 LOG.info("testRecoveredEdits");
209 conf.setLong("hbase.regionserver.hlog.blocksize", 30 * 1024);
210 conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, false);
211 startCluster(NUM_RS);
212
213 final int NUM_LOG_LINES = 1000;
214 final SplitLogManager slm = master.getMasterFileSystem().splitLogManager;
215
216
217 master.balanceSwitch(false);
218 FileSystem fs = master.getMasterFileSystem().getFileSystem();
219
220 List<RegionServerThread> rsts = cluster.getLiveRegionServerThreads();
221
222 Path rootdir = FSUtils.getRootDir(conf);
223
224 Table t = installTable(new ZooKeeperWatcher(conf, "table-creation", null),
225 "table", "family", 40);
226 try {
227 TableName table = t.getName();
228 List<HRegionInfo> regions = null;
229 HRegionServer hrs = null;
230 for (int i = 0; i < NUM_RS; i++) {
231 boolean foundRs = false;
232 hrs = rsts.get(i).getRegionServer();
233 regions = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
234 for (HRegionInfo region : regions) {
235 if (region.getTable().getNameAsString().equalsIgnoreCase("table")) {
236 foundRs = true;
237 break;
238 }
239 }
240 if (foundRs) break;
241 }
242 final Path logDir = new Path(rootdir, DefaultWALProvider.getWALDirectoryName(hrs
243 .getServerName().toString()));
244
245 LOG.info("#regions = " + regions.size());
246 Iterator<HRegionInfo> it = regions.iterator();
247 while (it.hasNext()) {
248 HRegionInfo region = it.next();
249 if (region.getTable().getNamespaceAsString()
250 .equals(NamespaceDescriptor.SYSTEM_NAMESPACE_NAME_STR)) {
251 it.remove();
252 }
253 }
254
255 makeWAL(hrs, regions, "table", "family", NUM_LOG_LINES, 100);
256
257 slm.splitLogDistributed(logDir);
258
259 int count = 0;
260 for (HRegionInfo hri : regions) {
261
262 Path tdir = FSUtils.getTableDir(rootdir, table);
263 Path editsdir =
264 WALSplitter.getRegionDirRecoveredEditsDir(
265 HRegion.getRegionDir(tdir, hri.getEncodedName()));
266 LOG.debug("checking edits dir " + editsdir);
267 FileStatus[] files = fs.listStatus(editsdir, new PathFilter() {
268 @Override
269 public boolean accept(Path p) {
270 if (WALSplitter.isSequenceIdFile(p)) {
271 return false;
272 }
273 return true;
274 }
275 });
276 assertTrue(
277 "edits dir should have more than a single file in it. instead has " + files.length,
278 files.length > 1);
279 for (int i = 0; i < files.length; i++) {
280 int c = countWAL(files[i].getPath(), fs, conf);
281 count += c;
282 }
283 LOG.info(count + " edits in " + files.length + " recovered edits files.");
284 }
285
286
287 assertFalse(fs.exists(logDir));
288
289 assertEquals(NUM_LOG_LINES, count);
290 } finally {
291 if (t != null) t.close();
292 }
293 }
294
295 @Test(timeout = 300000)
296 public void testLogReplayWithNonMetaRSDown() throws Exception {
297 LOG.info("testLogReplayWithNonMetaRSDown");
298 conf.setLong("hbase.regionserver.hlog.blocksize", 30 * 1024);
299 conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, true);
300 startCluster(NUM_RS);
301 final int NUM_REGIONS_TO_CREATE = 40;
302 final int NUM_LOG_LINES = 1000;
303
304
305 master.balanceSwitch(false);
306
307 final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "table-creation", null);
308 Table ht = installTable(zkw, "table", "family", NUM_REGIONS_TO_CREATE);
309 try {
310 HRegionServer hrs = findRSToKill(false, "table");
311 List<HRegionInfo> regions = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
312 makeWAL(hrs, regions, "table", "family", NUM_LOG_LINES, 100);
313
314
315 this.abortRSAndVerifyRecovery(hrs, ht, zkw, NUM_REGIONS_TO_CREATE, NUM_LOG_LINES);
316 } finally {
317 if (ht != null) ht.close();
318 if (zkw != null) zkw.close();
319 }
320 }
321
322 private static class NonceGeneratorWithDups extends PerClientRandomNonceGenerator {
323 private boolean isDups = false;
324 private LinkedList<Long> nonces = new LinkedList<Long>();
325
326 public void startDups() {
327 isDups = true;
328 }
329
330 @Override
331 public long newNonce() {
332 long nonce = isDups ? nonces.removeFirst() : super.newNonce();
333 if (!isDups) {
334 nonces.add(nonce);
335 }
336 return nonce;
337 }
338 }
339
340 @Test(timeout = 300000)
341 public void testNonceRecovery() throws Exception {
342 LOG.info("testNonceRecovery");
343 final String TABLE_NAME = "table";
344 final String FAMILY_NAME = "family";
345 final int NUM_REGIONS_TO_CREATE = 40;
346
347 conf.setLong("hbase.regionserver.hlog.blocksize", 100*1024);
348 conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, true);
349 startCluster(NUM_RS);
350 master.balanceSwitch(false);
351
352 final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "table-creation", null);
353 Table ht = installTable(zkw, TABLE_NAME, FAMILY_NAME, NUM_REGIONS_TO_CREATE);
354 NonceGeneratorWithDups ng = new NonceGeneratorWithDups();
355 NonceGenerator oldNg =
356 ConnectionUtils.injectNonceGeneratorForTesting(
357 (ClusterConnection)TEST_UTIL.getConnection(), ng);
358
359 try {
360 List<Increment> reqs = new ArrayList<Increment>();
361 for (RegionServerThread rst : cluster.getLiveRegionServerThreads()) {
362 HRegionServer hrs = rst.getRegionServer();
363 List<HRegionInfo> hris = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
364 for (HRegionInfo hri : hris) {
365 if (TABLE_NAME.equalsIgnoreCase(hri.getTable().getNameAsString())) {
366 byte[] key = hri.getStartKey();
367 if (key == null || key.length == 0) {
368 key = Bytes.copy(hri.getEndKey());
369 --(key[key.length - 1]);
370 }
371 Increment incr = new Increment(key);
372 incr.addColumn(Bytes.toBytes(FAMILY_NAME), Bytes.toBytes("q"), 1);
373 ht.increment(incr);
374 reqs.add(incr);
375 }
376 }
377 }
378
379 HRegionServer hrs = findRSToKill(false, "table");
380 abortRSAndWaitForRecovery(hrs, zkw, NUM_REGIONS_TO_CREATE);
381 ng.startDups();
382 for (Increment incr : reqs) {
383 try {
384 ht.increment(incr);
385 fail("should have thrown");
386 } catch (OperationConflictException ope) {
387 LOG.debug("Caught as expected: " + ope.getMessage());
388 }
389 }
390 } finally {
391 ConnectionUtils.injectNonceGeneratorForTesting((ClusterConnection)
392 TEST_UTIL.getConnection(), oldNg);
393 if (ht != null) ht.close();
394 if (zkw != null) zkw.close();
395 }
396 }
397
398 @Test(timeout = 300000)
399 public void testLogReplayWithMetaRSDown() throws Exception {
400 LOG.info("testRecoveredEditsReplayWithMetaRSDown");
401 conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, true);
402 startCluster(NUM_RS);
403 final int NUM_REGIONS_TO_CREATE = 40;
404 final int NUM_LOG_LINES = 1000;
405
406
407 master.balanceSwitch(false);
408
409 final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "table-creation", null);
410 Table ht = installTable(zkw, "table", "family", NUM_REGIONS_TO_CREATE);
411 try {
412 HRegionServer hrs = findRSToKill(true, "table");
413 List<HRegionInfo> regions = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
414 makeWAL(hrs, regions, "table", "family", NUM_LOG_LINES, 100);
415
416 this.abortRSAndVerifyRecovery(hrs, ht, zkw, NUM_REGIONS_TO_CREATE, NUM_LOG_LINES);
417 } finally {
418 if (ht != null) ht.close();
419 if (zkw != null) zkw.close();
420 }
421 }
422
423 private void abortRSAndVerifyRecovery(HRegionServer hrs, Table ht, final ZooKeeperWatcher zkw,
424 final int numRegions, final int numofLines) throws Exception {
425
426 abortRSAndWaitForRecovery(hrs, zkw, numRegions);
427 assertEquals(numofLines, TEST_UTIL.countRows(ht));
428 }
429
430 private void abortRSAndWaitForRecovery(HRegionServer hrs, final ZooKeeperWatcher zkw,
431 final int numRegions) throws Exception {
432 final MiniHBaseCluster tmpCluster = this.cluster;
433
434
435 LOG.info("Aborting region server: " + hrs.getServerName());
436 hrs.abort("testing");
437
438
439 TEST_UTIL.waitFor(120000, 200, new Waiter.Predicate<Exception>() {
440 @Override
441 public boolean evaluate() throws Exception {
442 return (tmpCluster.getLiveRegionServerThreads().size() <= (NUM_RS - 1));
443 }
444 });
445
446
447 TEST_UTIL.waitFor(180000, 200, new Waiter.Predicate<Exception>() {
448 @Override
449 public boolean evaluate() throws Exception {
450 return (HBaseTestingUtility.getAllOnlineRegions(tmpCluster).size()
451 >= (numRegions + 1));
452 }
453 });
454
455
456 TEST_UTIL.waitFor(180000, 200, new Waiter.Predicate<Exception>() {
457 @Override
458 public boolean evaluate() throws Exception {
459 List<String> recoveringRegions = zkw.getRecoverableZooKeeper().getChildren(
460 zkw.recoveringRegionsZNode, false);
461 return (recoveringRegions != null && recoveringRegions.size() == 0);
462 }
463 });
464 }
465
466 @Test(timeout = 300000)
467 public void testMasterStartsUpWithLogSplittingWork() throws Exception {
468 LOG.info("testMasterStartsUpWithLogSplittingWork");
469 conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, false);
470 conf.setInt(ServerManager.WAIT_ON_REGIONSERVERS_MINTOSTART, NUM_RS - 1);
471 startCluster(NUM_RS);
472
473 final int NUM_REGIONS_TO_CREATE = 40;
474 final int NUM_LOG_LINES = 1000;
475
476
477 master.balanceSwitch(false);
478
479 final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "table-creation", null);
480 Table ht = installTable(zkw, "table", "family", NUM_REGIONS_TO_CREATE);
481 try {
482 HRegionServer hrs = findRSToKill(false, "table");
483 List<HRegionInfo> regions = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
484 makeWAL(hrs, regions, "table", "family", NUM_LOG_LINES, 100);
485
486
487 abortMaster(cluster);
488
489
490 LOG.info("Aborting region server: " + hrs.getServerName());
491 hrs.abort("testing");
492
493
494 TEST_UTIL.waitFor(120000, 200, new Waiter.Predicate<Exception>() {
495 @Override
496 public boolean evaluate() throws Exception {
497 return (cluster.getLiveRegionServerThreads().size() <= (NUM_RS - 1));
498 }
499 });
500
501 Thread.sleep(2000);
502 LOG.info("Current Open Regions:"
503 + HBaseTestingUtility.getAllOnlineRegions(cluster).size());
504
505
506 TEST_UTIL.waitFor(120000, 200, new Waiter.Predicate<Exception>() {
507 @Override
508 public boolean evaluate() throws Exception {
509 return (HBaseTestingUtility.getAllOnlineRegions(cluster).size()
510 >= (NUM_REGIONS_TO_CREATE + 1));
511 }
512 });
513
514 LOG.info("Current Open Regions After Master Node Starts Up:"
515 + HBaseTestingUtility.getAllOnlineRegions(cluster).size());
516
517 assertEquals(NUM_LOG_LINES, TEST_UTIL.countRows(ht));
518 } finally {
519 if (ht != null) ht.close();
520 if (zkw != null) zkw.close();
521 }
522 }
523
524 @Test(timeout = 300000)
525 public void testMasterStartsUpWithLogReplayWork() throws Exception {
526 LOG.info("testMasterStartsUpWithLogReplayWork");
527 conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, true);
528 conf.setInt(ServerManager.WAIT_ON_REGIONSERVERS_MINTOSTART, NUM_RS - 1);
529 startCluster(NUM_RS);
530
531 final int NUM_REGIONS_TO_CREATE = 40;
532 final int NUM_LOG_LINES = 1000;
533
534
535 master.balanceSwitch(false);
536
537 final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "table-creation", null);
538 Table ht = installTable(zkw, "table", "family", NUM_REGIONS_TO_CREATE);
539 try {
540 HRegionServer hrs = findRSToKill(false, "table");
541 List<HRegionInfo> regions = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
542 makeWAL(hrs, regions, "table", "family", NUM_LOG_LINES, 100);
543
544
545 abortMaster(cluster);
546
547
548 LOG.info("Aborting region server: " + hrs.getServerName());
549 hrs.abort("testing");
550
551
552 TEST_UTIL.waitFor(120000, 200, new Waiter.Predicate<Exception>() {
553 @Override
554 public boolean evaluate() throws Exception {
555 return (cluster.getLiveRegionServerThreads().size() <= (NUM_RS - 1));
556 }
557 });
558
559 Thread.sleep(2000);
560 LOG.info("Current Open Regions:" + HBaseTestingUtility.getAllOnlineRegions(cluster).size());
561
562
563 TEST_UTIL.waitFor(180000, 200, new Waiter.Predicate<Exception>() {
564 @Override
565 public boolean evaluate() throws Exception {
566 List<String> recoveringRegions = zkw.getRecoverableZooKeeper().getChildren(
567 zkw.recoveringRegionsZNode, false);
568 boolean done = recoveringRegions != null && recoveringRegions.size() == 0;
569 if (!done) {
570 LOG.info("Recovering regions: " + recoveringRegions);
571 }
572 return done;
573 }
574 });
575
576 LOG.info("Current Open Regions After Master Node Starts Up:"
577 + HBaseTestingUtility.getAllOnlineRegions(cluster).size());
578
579 assertEquals(NUM_LOG_LINES, TEST_UTIL.countRows(ht));
580 } finally {
581 if (ht != null) ht.close();
582 if (zkw != null) zkw.close();
583 }
584 }
585
586
587 @Test(timeout = 300000)
588 public void testLogReplayTwoSequentialRSDown() throws Exception {
589 LOG.info("testRecoveredEditsReplayTwoSequentialRSDown");
590 conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, true);
591 startCluster(NUM_RS);
592 final int NUM_REGIONS_TO_CREATE = 40;
593 final int NUM_LOG_LINES = 1000;
594
595
596 master.balanceSwitch(false);
597
598 List<RegionServerThread> rsts = cluster.getLiveRegionServerThreads();
599 final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "table-creation", null);
600 Table ht = installTable(zkw, "table", "family", NUM_REGIONS_TO_CREATE);
601 try {
602 List<HRegionInfo> regions = null;
603 HRegionServer hrs1 = findRSToKill(false, "table");
604 regions = ProtobufUtil.getOnlineRegions(hrs1.getRSRpcServices());
605
606 makeWAL(hrs1, regions, "table", "family", NUM_LOG_LINES, 100);
607
608
609 LOG.info("Aborting region server: " + hrs1.getServerName());
610 hrs1.abort("testing");
611
612
613 TEST_UTIL.waitFor(120000, 200, new Waiter.Predicate<Exception>() {
614 @Override
615 public boolean evaluate() throws Exception {
616 return (cluster.getLiveRegionServerThreads().size() <= (NUM_RS - 1));
617 }
618 });
619
620
621 TEST_UTIL.waitFor(180000, 200, new Waiter.Predicate<Exception>() {
622 @Override
623 public boolean evaluate() throws Exception {
624 return (HBaseTestingUtility.getAllOnlineRegions(cluster).size()
625 >= (NUM_REGIONS_TO_CREATE + 1));
626 }
627 });
628
629
630 Thread.sleep(300);
631
632 rsts = cluster.getLiveRegionServerThreads();
633 HRegionServer hrs2 = rsts.get(0).getRegionServer();
634 LOG.info("Aborting one more region server: " + hrs2.getServerName());
635 hrs2.abort("testing");
636
637
638 TEST_UTIL.waitFor(120000, 200, new Waiter.Predicate<Exception>() {
639 @Override
640 public boolean evaluate() throws Exception {
641 return (cluster.getLiveRegionServerThreads().size() <= (NUM_RS - 2));
642 }
643 });
644
645
646 TEST_UTIL.waitFor(180000, 200, new Waiter.Predicate<Exception>() {
647 @Override
648 public boolean evaluate() throws Exception {
649 return (HBaseTestingUtility.getAllOnlineRegions(cluster).size()
650 >= (NUM_REGIONS_TO_CREATE + 1));
651 }
652 });
653
654
655 TEST_UTIL.waitFor(180000, 200, new Waiter.Predicate<Exception>() {
656 @Override
657 public boolean evaluate() throws Exception {
658 List<String> recoveringRegions = zkw.getRecoverableZooKeeper().getChildren(
659 zkw.recoveringRegionsZNode, false);
660 return (recoveringRegions != null && recoveringRegions.size() == 0);
661 }
662 });
663
664 assertEquals(NUM_LOG_LINES, TEST_UTIL.countRows(ht));
665 } finally {
666 if (ht != null) ht.close();
667 if (zkw != null) zkw.close();
668 }
669 }
670
671 @Test(timeout = 300000)
672 public void testMarkRegionsRecoveringInZK() throws Exception {
673 LOG.info("testMarkRegionsRecoveringInZK");
674 conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, true);
675 startCluster(NUM_RS);
676 master.balanceSwitch(false);
677 List<RegionServerThread> rsts = cluster.getLiveRegionServerThreads();
678 final ZooKeeperWatcher zkw = master.getZooKeeper();
679 Table ht = installTable(zkw, "table", "family", 40);
680 try {
681 final SplitLogManager slm = master.getMasterFileSystem().splitLogManager;
682
683 Set<HRegionInfo> regionSet = new HashSet<HRegionInfo>();
684 HRegionInfo region = null;
685 HRegionServer hrs = null;
686 ServerName firstFailedServer = null;
687 ServerName secondFailedServer = null;
688 for (int i = 0; i < NUM_RS; i++) {
689 hrs = rsts.get(i).getRegionServer();
690 List<HRegionInfo> regions = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
691 if (regions.isEmpty()) continue;
692 region = regions.get(0);
693 regionSet.add(region);
694 firstFailedServer = hrs.getServerName();
695 secondFailedServer = rsts.get((i + 1) % NUM_RS).getRegionServer().getServerName();
696 break;
697 }
698
699 slm.markRegionsRecovering(firstFailedServer, regionSet);
700 slm.markRegionsRecovering(secondFailedServer, regionSet);
701
702 List<String> recoveringRegions = ZKUtil.listChildrenNoWatch(zkw,
703 ZKUtil.joinZNode(zkw.recoveringRegionsZNode, region.getEncodedName()));
704
705 assertEquals(recoveringRegions.size(), 2);
706
707
708 final HRegionServer tmphrs = hrs;
709 TEST_UTIL.waitFor(60000, 1000, new Waiter.Predicate<Exception>() {
710 @Override
711 public boolean evaluate() throws Exception {
712 return (tmphrs.getRecoveringRegions().size() == 0);
713 }
714 });
715 } finally {
716 if (ht != null) ht.close();
717 if (zkw != null) zkw.close();
718 }
719 }
720
721 @Test(timeout = 300000)
722 public void testReplayCmd() throws Exception {
723 LOG.info("testReplayCmd");
724 conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, true);
725 startCluster(NUM_RS);
726 final int NUM_REGIONS_TO_CREATE = 40;
727
728
729 master.balanceSwitch(false);
730
731 List<RegionServerThread> rsts = cluster.getLiveRegionServerThreads();
732 final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "table-creation", null);
733 Table ht = installTable(zkw, "table", "family", NUM_REGIONS_TO_CREATE);
734 try {
735 List<HRegionInfo> regions = null;
736 HRegionServer hrs = null;
737 for (int i = 0; i < NUM_RS; i++) {
738 boolean isCarryingMeta = false;
739 hrs = rsts.get(i).getRegionServer();
740 regions = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
741 for (HRegionInfo region : regions) {
742 if (region.isMetaRegion()) {
743 isCarryingMeta = true;
744 break;
745 }
746 }
747 if (isCarryingMeta) {
748 continue;
749 }
750 if (regions.size() > 0) break;
751 }
752
753 this.prepareData(ht, Bytes.toBytes("family"), Bytes.toBytes("c1"));
754 String originalCheckSum = TEST_UTIL.checksumRows(ht);
755
756
757 abortRSAndWaitForRecovery(hrs, zkw, NUM_REGIONS_TO_CREATE);
758
759 assertEquals("Data should remain after reopening of regions", originalCheckSum,
760 TEST_UTIL.checksumRows(ht));
761 } finally {
762 if (ht != null) ht.close();
763 if (zkw != null) zkw.close();
764 }
765 }
766
767 @Test(timeout = 300000)
768 public void testLogReplayForDisablingTable() throws Exception {
769 LOG.info("testLogReplayForDisablingTable");
770 conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, true);
771 startCluster(NUM_RS);
772 final int NUM_REGIONS_TO_CREATE = 40;
773 final int NUM_LOG_LINES = 1000;
774
775 List<RegionServerThread> rsts = cluster.getLiveRegionServerThreads();
776 final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "table-creation", null);
777 Table disablingHT = installTable(zkw, "disableTable", "family", NUM_REGIONS_TO_CREATE);
778 Table ht = installTable(zkw, "table", "family", NUM_REGIONS_TO_CREATE, NUM_REGIONS_TO_CREATE);
779 try {
780
781
782 master.balanceSwitch(false);
783
784 List<HRegionInfo> regions = null;
785 HRegionServer hrs = null;
786 boolean hasRegionsForBothTables = false;
787 String tableName = null;
788 for (int i = 0; i < NUM_RS; i++) {
789 tableName = null;
790 hasRegionsForBothTables = false;
791 boolean isCarryingSystem = false;
792 hrs = rsts.get(i).getRegionServer();
793 regions = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
794 for (HRegionInfo region : regions) {
795 if (region.getTable().isSystemTable()) {
796 isCarryingSystem = true;
797 break;
798 }
799 if (tableName != null &&
800 !tableName.equalsIgnoreCase(region.getTable().getNameAsString())) {
801
802 hasRegionsForBothTables = true;
803 break;
804 } else if (tableName == null) {
805 tableName = region.getTable().getNameAsString();
806 }
807 }
808 if (isCarryingSystem) {
809 continue;
810 }
811 if (hasRegionsForBothTables) {
812 break;
813 }
814 }
815
816
817 Assert.assertTrue(hasRegionsForBothTables);
818
819 LOG.info("#regions = " + regions.size());
820 Iterator<HRegionInfo> it = regions.iterator();
821 while (it.hasNext()) {
822 HRegionInfo region = it.next();
823 if (region.isMetaTable()) {
824 it.remove();
825 }
826 }
827 makeWAL(hrs, regions, "disableTable", "family", NUM_LOG_LINES, 100, false);
828 makeWAL(hrs, regions, "table", "family", NUM_LOG_LINES, 100);
829
830 LOG.info("Disabling table\n");
831 TEST_UTIL.getHBaseAdmin().disableTable(TableName.valueOf("disableTable"));
832 TEST_UTIL.waitTableDisabled(TableName.valueOf("disableTable").getName());
833
834
835 LOG.info("Aborting region server: " + hrs.getServerName());
836 hrs.abort("testing");
837
838
839 TEST_UTIL.waitFor(120000, 200, new Waiter.Predicate<Exception>() {
840 @Override
841 public boolean evaluate() throws Exception {
842 return (cluster.getLiveRegionServerThreads().size() <= (NUM_RS - 1));
843 }
844 });
845
846
847 TEST_UTIL.waitFor(180000, 200, new Waiter.Predicate<Exception>() {
848 @Override
849 public boolean evaluate() throws Exception {
850 return (HBaseTestingUtility.getAllOnlineRegions(cluster).size()
851 >= (NUM_REGIONS_TO_CREATE + 1));
852 }
853 });
854
855
856 TEST_UTIL.waitFor(180000, 200, new Waiter.Predicate<Exception>() {
857 @Override
858 public boolean evaluate() throws Exception {
859 List<String> recoveringRegions = zkw.getRecoverableZooKeeper().getChildren(
860 zkw.recoveringRegionsZNode, false);
861 ServerManager serverManager = master.getServerManager();
862 return (!serverManager.areDeadServersInProgress() &&
863 recoveringRegions != null && recoveringRegions.size() == 0);
864 }
865 });
866
867 int count = 0;
868 FileSystem fs = master.getMasterFileSystem().getFileSystem();
869 Path rootdir = FSUtils.getRootDir(conf);
870 Path tdir = FSUtils.getTableDir(rootdir, TableName.valueOf("disableTable"));
871 for (HRegionInfo hri : regions) {
872 Path editsdir =
873 WALSplitter.getRegionDirRecoveredEditsDir(
874 HRegion.getRegionDir(tdir, hri.getEncodedName()));
875 LOG.debug("checking edits dir " + editsdir);
876 if(!fs.exists(editsdir)) continue;
877 FileStatus[] files = fs.listStatus(editsdir, new PathFilter() {
878 @Override
879 public boolean accept(Path p) {
880 if (WALSplitter.isSequenceIdFile(p)) {
881 return false;
882 }
883 return true;
884 }
885 });
886 if(files != null) {
887 for(FileStatus file : files) {
888 int c = countWAL(file.getPath(), fs, conf);
889 count += c;
890 LOG.info(c + " edits in " + file.getPath());
891 }
892 }
893 }
894
895 LOG.info("Verify edits in recovered.edits files");
896 assertEquals(NUM_LOG_LINES, count);
897 LOG.info("Verify replayed edits");
898 assertEquals(NUM_LOG_LINES, TEST_UTIL.countRows(ht));
899
900
901 for (HRegionInfo hri : regions) {
902 Path editsdir =
903 WALSplitter.getRegionDirRecoveredEditsDir(
904 HRegion.getRegionDir(tdir, hri.getEncodedName()));
905 fs.delete(editsdir, true);
906 }
907 disablingHT.close();
908 } finally {
909 if (ht != null) ht.close();
910 if (zkw != null) zkw.close();
911 }
912 }
913
914 @Ignore ("We don't support DLR anymore") @Test(timeout = 300000)
915 public void testDisallowWritesInRecovering() throws Exception {
916 LOG.info("testDisallowWritesInRecovering");
917 conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, true);
918 conf.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 3);
919 conf.setBoolean(HConstants.DISALLOW_WRITES_IN_RECOVERING, true);
920 startCluster(NUM_RS);
921 final int NUM_REGIONS_TO_CREATE = 40;
922
923
924 master.balanceSwitch(false);
925
926 List<RegionServerThread> rsts = cluster.getLiveRegionServerThreads();
927 final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "table-creation", null);
928 Table ht = installTable(zkw, "table", "family", NUM_REGIONS_TO_CREATE);
929 try {
930 final SplitLogManager slm = master.getMasterFileSystem().splitLogManager;
931
932 Set<HRegionInfo> regionSet = new HashSet<HRegionInfo>();
933 HRegionInfo region = null;
934 HRegionServer hrs = null;
935 HRegionServer dstRS = null;
936 for (int i = 0; i < NUM_RS; i++) {
937 hrs = rsts.get(i).getRegionServer();
938 List<HRegionInfo> regions = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
939 if (regions.isEmpty()) continue;
940 region = regions.get(0);
941 regionSet.add(region);
942 dstRS = rsts.get((i+1) % NUM_RS).getRegionServer();
943 break;
944 }
945
946 slm.markRegionsRecovering(hrs.getServerName(), regionSet);
947
948 final HRegionInfo hri = region;
949 final HRegionServer tmpRS = dstRS;
950 TEST_UTIL.getHBaseAdmin().move(region.getEncodedNameAsBytes(),
951 Bytes.toBytes(dstRS.getServerName().getServerName()));
952
953 final RegionStates regionStates =
954 TEST_UTIL.getHBaseCluster().getMaster().getAssignmentManager().getRegionStates();
955 TEST_UTIL.waitFor(45000, 200, new Waiter.Predicate<Exception>() {
956 @Override
957 public boolean evaluate() throws Exception {
958 ServerName sn = regionStates.getRegionServerOfRegion(hri);
959 return (sn != null && sn.equals(tmpRS.getServerName()));
960 }
961 });
962
963 try {
964 byte[] key = region.getStartKey();
965 if (key == null || key.length == 0) {
966 key = new byte[] { 0, 0, 0, 0, 1 };
967 }
968 Put put = new Put(key);
969 put.add(Bytes.toBytes("family"), Bytes.toBytes("c1"), new byte[]{'b'});
970 ht.put(put);
971 } catch (IOException ioe) {
972 Assert.assertTrue(ioe instanceof RetriesExhaustedWithDetailsException);
973 RetriesExhaustedWithDetailsException re = (RetriesExhaustedWithDetailsException) ioe;
974 boolean foundRegionInRecoveryException = false;
975 for (Throwable t : re.getCauses()) {
976 if (t instanceof RegionInRecoveryException) {
977 foundRegionInRecoveryException = true;
978 break;
979 }
980 }
981 Assert.assertTrue(
982 "No RegionInRecoveryException. Following exceptions returned=" + re.getCauses(),
983 foundRegionInRecoveryException);
984 }
985 } finally {
986 if (ht != null) ht.close();
987 if (ht != null) zkw.close();
988 }
989 }
990
991
992
993
994
995
996
997
998
999
1000 @Test (timeout=300000)
1001 public void testWorkerAbort() throws Exception {
1002 LOG.info("testWorkerAbort");
1003 startCluster(3);
1004 final int NUM_LOG_LINES = 10000;
1005 final SplitLogManager slm = master.getMasterFileSystem().splitLogManager;
1006 FileSystem fs = master.getMasterFileSystem().getFileSystem();
1007
1008 final List<RegionServerThread> rsts = cluster.getLiveRegionServerThreads();
1009 HRegionServer hrs = findRSToKill(false, "table");
1010 Path rootdir = FSUtils.getRootDir(conf);
1011 final Path logDir = new Path(rootdir,
1012 DefaultWALProvider.getWALDirectoryName(hrs.getServerName().toString()));
1013
1014 Table t = installTable(new ZooKeeperWatcher(conf, "table-creation", null),
1015 "table", "family", 40);
1016 try {
1017 makeWAL(hrs, ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices()),
1018 "table", "family", NUM_LOG_LINES, 100);
1019
1020 new Thread() {
1021 @Override
1022 public void run() {
1023 waitForCounter(tot_wkr_task_acquired, 0, 1, 1000);
1024 for (RegionServerThread rst : rsts) {
1025 rst.getRegionServer().abort("testing");
1026 break;
1027 }
1028 }
1029 }.start();
1030
1031 FileStatus[] logfiles = fs.listStatus(logDir);
1032 TaskBatch batch = new TaskBatch();
1033 slm.enqueueSplitTask(logfiles[0].getPath().toString(), batch);
1034
1035 long curt = System.currentTimeMillis();
1036 long waitTime = 80000;
1037 long endt = curt + waitTime;
1038 while (curt < endt) {
1039 if ((tot_wkr_task_resigned.get() + tot_wkr_task_err.get() +
1040 tot_wkr_final_transition_failed.get() + tot_wkr_task_done.get() +
1041 tot_wkr_preempt_task.get()) == 0) {
1042 Thread.yield();
1043 curt = System.currentTimeMillis();
1044 } else {
1045 assertTrue(1 <= (tot_wkr_task_resigned.get() + tot_wkr_task_err.get() +
1046 tot_wkr_final_transition_failed.get() + tot_wkr_task_done.get() +
1047 tot_wkr_preempt_task.get()));
1048 return;
1049 }
1050 }
1051 fail("none of the following counters went up in " + waitTime +
1052 " milliseconds - " +
1053 "tot_wkr_task_resigned, tot_wkr_task_err, " +
1054 "tot_wkr_final_transition_failed, tot_wkr_task_done, " +
1055 "tot_wkr_preempt_task");
1056 } finally {
1057 if (t != null) t.close();
1058 }
1059 }
1060
1061 @Test (timeout=300000)
1062 public void testThreeRSAbort() throws Exception {
1063 LOG.info("testThreeRSAbort");
1064 final int NUM_REGIONS_TO_CREATE = 40;
1065 final int NUM_ROWS_PER_REGION = 100;
1066
1067 startCluster(NUM_RS);
1068
1069 final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf,
1070 "distributed log splitting test", null);
1071
1072 Table ht = installTable(zkw, "table", "family", NUM_REGIONS_TO_CREATE);
1073 try {
1074 populateDataInTable(NUM_ROWS_PER_REGION, "family");
1075
1076
1077 List<RegionServerThread> rsts = cluster.getLiveRegionServerThreads();
1078 assertEquals(NUM_RS, rsts.size());
1079 rsts.get(0).getRegionServer().abort("testing");
1080 rsts.get(1).getRegionServer().abort("testing");
1081 rsts.get(2).getRegionServer().abort("testing");
1082
1083 long start = EnvironmentEdgeManager.currentTime();
1084 while (cluster.getLiveRegionServerThreads().size() > (NUM_RS - 3)) {
1085 if (EnvironmentEdgeManager.currentTime() - start > 60000) {
1086 assertTrue(false);
1087 }
1088 Thread.sleep(200);
1089 }
1090
1091 start = EnvironmentEdgeManager.currentTime();
1092 while (HBaseTestingUtility.getAllOnlineRegions(cluster).size()
1093 < (NUM_REGIONS_TO_CREATE + 1)) {
1094 if (EnvironmentEdgeManager.currentTime() - start > 60000) {
1095 assertTrue("Timedout", false);
1096 }
1097 Thread.sleep(200);
1098 }
1099
1100
1101 TEST_UTIL.waitFor(180000, 200, new Waiter.Predicate<Exception>() {
1102 @Override
1103 public boolean evaluate() throws Exception {
1104 List<String> recoveringRegions = zkw.getRecoverableZooKeeper().getChildren(
1105 zkw.recoveringRegionsZNode, false);
1106 return (recoveringRegions != null && recoveringRegions.size() == 0);
1107 }
1108 });
1109
1110 assertEquals(NUM_REGIONS_TO_CREATE * NUM_ROWS_PER_REGION,
1111 TEST_UTIL.countRows(ht));
1112 } finally {
1113 if (ht != null) ht.close();
1114 if (zkw != null) zkw.close();
1115 }
1116 }
1117
1118
1119
1120 @Test(timeout=30000)
1121 public void testDelayedDeleteOnFailure() throws Exception {
1122 LOG.info("testDelayedDeleteOnFailure");
1123 startCluster(1);
1124 final SplitLogManager slm = master.getMasterFileSystem().splitLogManager;
1125 final FileSystem fs = master.getMasterFileSystem().getFileSystem();
1126 final Path logDir = new Path(FSUtils.getRootDir(conf), "x");
1127 fs.mkdirs(logDir);
1128 ExecutorService executor = null;
1129 try {
1130 final Path corruptedLogFile = new Path(logDir, "x");
1131 FSDataOutputStream out;
1132 out = fs.create(corruptedLogFile);
1133 out.write(0);
1134 out.write(Bytes.toBytes("corrupted bytes"));
1135 out.close();
1136 ZKSplitLogManagerCoordination coordination =
1137 (ZKSplitLogManagerCoordination) ((BaseCoordinatedStateManager) master
1138 .getCoordinatedStateManager()).getSplitLogManagerCoordination();
1139 coordination.setIgnoreDeleteForTesting(true);
1140 executor = Executors.newSingleThreadExecutor();
1141 Runnable runnable = new Runnable() {
1142 @Override
1143 public void run() {
1144 try {
1145
1146
1147
1148 slm.splitLogDistributed(logDir);
1149 } catch (IOException ioe) {
1150 try {
1151 assertTrue(fs.exists(corruptedLogFile));
1152
1153
1154
1155 slm.splitLogDistributed(logDir);
1156 } catch (IOException e) {
1157 assertTrue(Thread.currentThread().isInterrupted());
1158 return;
1159 }
1160 fail("did not get the expected IOException from the 2nd call");
1161 }
1162 fail("did not get the expected IOException from the 1st call");
1163 }
1164 };
1165 Future<?> result = executor.submit(runnable);
1166 try {
1167 result.get(2000, TimeUnit.MILLISECONDS);
1168 } catch (TimeoutException te) {
1169
1170 }
1171 waitForCounter(tot_mgr_wait_for_zk_delete, 0, 1, 10000);
1172 executor.shutdownNow();
1173 executor = null;
1174
1175
1176 result.get();
1177 } finally {
1178 if (executor != null) {
1179
1180
1181 executor.shutdownNow();
1182 }
1183 fs.delete(logDir, true);
1184 }
1185 }
1186
1187 @Test(timeout = 300000)
1188 public void testMetaRecoveryInZK() throws Exception {
1189 LOG.info("testMetaRecoveryInZK");
1190 conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, true);
1191 startCluster(NUM_RS);
1192
1193
1194
1195 master.balanceSwitch(false);
1196 final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "table-creation", null);
1197
1198
1199 HRegionServer hrs = findRSToKill(true, null);
1200 List<HRegionInfo> regions = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
1201
1202 LOG.info("#regions = " + regions.size());
1203 Set<HRegionInfo> tmpRegions = new HashSet<HRegionInfo>();
1204 tmpRegions.add(HRegionInfo.FIRST_META_REGIONINFO);
1205 master.getMasterFileSystem().prepareLogReplay(hrs.getServerName(), tmpRegions);
1206 Set<HRegionInfo> userRegionSet = new HashSet<HRegionInfo>();
1207 userRegionSet.addAll(regions);
1208 master.getMasterFileSystem().prepareLogReplay(hrs.getServerName(), userRegionSet);
1209 boolean isMetaRegionInRecovery = false;
1210 List<String> recoveringRegions =
1211 zkw.getRecoverableZooKeeper().getChildren(zkw.recoveringRegionsZNode, false);
1212 for (String curEncodedRegionName : recoveringRegions) {
1213 if (curEncodedRegionName.equals(HRegionInfo.FIRST_META_REGIONINFO.getEncodedName())) {
1214 isMetaRegionInRecovery = true;
1215 break;
1216 }
1217 }
1218 assertTrue(isMetaRegionInRecovery);
1219
1220 master.getMasterFileSystem().splitMetaLog(hrs.getServerName());
1221
1222 isMetaRegionInRecovery = false;
1223 recoveringRegions =
1224 zkw.getRecoverableZooKeeper().getChildren(zkw.recoveringRegionsZNode, false);
1225 for (String curEncodedRegionName : recoveringRegions) {
1226 if (curEncodedRegionName.equals(HRegionInfo.FIRST_META_REGIONINFO.getEncodedName())) {
1227 isMetaRegionInRecovery = true;
1228 break;
1229 }
1230 }
1231
1232 assertFalse(isMetaRegionInRecovery);
1233 zkw.close();
1234 }
1235
1236 @Test(timeout = 300000)
1237 public void testSameVersionUpdatesRecovery() throws Exception {
1238 LOG.info("testSameVersionUpdatesRecovery");
1239 conf.setLong("hbase.regionserver.hlog.blocksize", 15 * 1024);
1240 conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, true);
1241 startCluster(NUM_RS);
1242 final AtomicLong sequenceId = new AtomicLong(100);
1243 final int NUM_REGIONS_TO_CREATE = 40;
1244 final int NUM_LOG_LINES = 1000;
1245
1246
1247 master.balanceSwitch(false);
1248
1249 List<RegionServerThread> rsts = cluster.getLiveRegionServerThreads();
1250 final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "table-creation", null);
1251 Table ht = installTable(zkw, "table", "family", NUM_REGIONS_TO_CREATE);
1252 try {
1253 List<HRegionInfo> regions = null;
1254 HRegionServer hrs = null;
1255 for (int i = 0; i < NUM_RS; i++) {
1256 boolean isCarryingMeta = false;
1257 hrs = rsts.get(i).getRegionServer();
1258 regions = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
1259 for (HRegionInfo region : regions) {
1260 if (region.isMetaRegion()) {
1261 isCarryingMeta = true;
1262 break;
1263 }
1264 }
1265 if (isCarryingMeta) {
1266 continue;
1267 }
1268 break;
1269 }
1270
1271 LOG.info("#regions = " + regions.size());
1272 Iterator<HRegionInfo> it = regions.iterator();
1273 while (it.hasNext()) {
1274 HRegionInfo region = it.next();
1275 if (region.isMetaTable()
1276 || region.getEncodedName().equals(
1277 HRegionInfo.FIRST_META_REGIONINFO.getEncodedName())) {
1278 it.remove();
1279 }
1280 }
1281 if (regions.size() == 0) return;
1282 HRegionInfo curRegionInfo = regions.get(0);
1283 byte[] startRow = curRegionInfo.getStartKey();
1284 if (startRow == null || startRow.length == 0) {
1285 startRow = new byte[] { 0, 0, 0, 0, 1 };
1286 }
1287 byte[] row = Bytes.incrementBytes(startRow, 1);
1288
1289 row = Arrays.copyOfRange(row, 3, 8);
1290 long value = 0;
1291 TableName tableName = TableName.valueOf("table");
1292 byte[] family = Bytes.toBytes("family");
1293 byte[] qualifier = Bytes.toBytes("c1");
1294 long timeStamp = System.currentTimeMillis();
1295 HTableDescriptor htd = new HTableDescriptor(tableName);
1296 htd.addFamily(new HColumnDescriptor(family));
1297 final WAL wal = hrs.getWAL(curRegionInfo);
1298 for (int i = 0; i < NUM_LOG_LINES; i += 1) {
1299 WALEdit e = new WALEdit();
1300 value++;
1301 e.add(new KeyValue(row, family, qualifier, timeStamp, Bytes.toBytes(value)));
1302 wal.append(htd, curRegionInfo,
1303 new HLogKey(curRegionInfo.getEncodedNameAsBytes(), tableName, System.currentTimeMillis()),
1304 e, sequenceId, true, null);
1305 }
1306 wal.sync();
1307 wal.shutdown();
1308
1309
1310 this.abortRSAndWaitForRecovery(hrs, zkw, NUM_REGIONS_TO_CREATE);
1311
1312
1313 LOG.info("Verification Starts...");
1314 Get g = new Get(row);
1315 Result r = ht.get(g);
1316 long theStoredVal = Bytes.toLong(r.getValue(family, qualifier));
1317 assertEquals(value, theStoredVal);
1318
1319
1320 LOG.info("Verification after flush...");
1321 TEST_UTIL.getHBaseAdmin().flush(tableName);
1322 r = ht.get(g);
1323 theStoredVal = Bytes.toLong(r.getValue(family, qualifier));
1324 assertEquals(value, theStoredVal);
1325 } finally {
1326 if (ht != null) ht.close();
1327 if (zkw != null) zkw.close();
1328 }
1329 }
1330
1331 @Test(timeout = 300000)
1332 public void testSameVersionUpdatesRecoveryWithCompaction() throws Exception {
1333 LOG.info("testSameVersionUpdatesRecoveryWithWrites");
1334 conf.setLong("hbase.regionserver.hlog.blocksize", 15 * 1024);
1335 conf.setBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, true);
1336 conf.setInt(HConstants.HREGION_MEMSTORE_FLUSH_SIZE, 30 * 1024);
1337 conf.setInt("hbase.hstore.compactionThreshold", 3);
1338 startCluster(NUM_RS);
1339 final AtomicLong sequenceId = new AtomicLong(100);
1340 final int NUM_REGIONS_TO_CREATE = 40;
1341 final int NUM_LOG_LINES = 2000;
1342
1343
1344 master.balanceSwitch(false);
1345
1346 List<RegionServerThread> rsts = cluster.getLiveRegionServerThreads();
1347 final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "table-creation", null);
1348 Table ht = installTable(zkw, "table", "family", NUM_REGIONS_TO_CREATE);
1349 try {
1350 List<HRegionInfo> regions = null;
1351 HRegionServer hrs = null;
1352 for (int i = 0; i < NUM_RS; i++) {
1353 boolean isCarryingMeta = false;
1354 hrs = rsts.get(i).getRegionServer();
1355 regions = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
1356 for (HRegionInfo region : regions) {
1357 if (region.isMetaRegion()) {
1358 isCarryingMeta = true;
1359 break;
1360 }
1361 }
1362 if (isCarryingMeta) {
1363 continue;
1364 }
1365 break;
1366 }
1367
1368 LOG.info("#regions = " + regions.size());
1369 Iterator<HRegionInfo> it = regions.iterator();
1370 while (it.hasNext()) {
1371 HRegionInfo region = it.next();
1372 if (region.isMetaTable()
1373 || region.getEncodedName().equals(HRegionInfo.FIRST_META_REGIONINFO.getEncodedName())) {
1374 it.remove();
1375 }
1376 }
1377 if (regions.size() == 0) return;
1378 HRegionInfo curRegionInfo = regions.get(0);
1379 byte[] startRow = curRegionInfo.getStartKey();
1380 if (startRow == null || startRow.length == 0) {
1381 startRow = new byte[] { 0, 0, 0, 0, 1 };
1382 }
1383 byte[] row = Bytes.incrementBytes(startRow, 1);
1384
1385 row = Arrays.copyOfRange(row, 3, 8);
1386 long value = 0;
1387 final TableName tableName = TableName.valueOf("table");
1388 byte[] family = Bytes.toBytes("family");
1389 byte[] qualifier = Bytes.toBytes("c1");
1390 long timeStamp = System.currentTimeMillis();
1391 HTableDescriptor htd = new HTableDescriptor(tableName);
1392 htd.addFamily(new HColumnDescriptor(family));
1393 final WAL wal = hrs.getWAL(curRegionInfo);
1394 for (int i = 0; i < NUM_LOG_LINES; i += 1) {
1395 WALEdit e = new WALEdit();
1396 value++;
1397 e.add(new KeyValue(row, family, qualifier, timeStamp, Bytes.toBytes(value)));
1398 wal.append(htd, curRegionInfo, new HLogKey(curRegionInfo.getEncodedNameAsBytes(),
1399 tableName, System.currentTimeMillis()), e, sequenceId, true, null);
1400 }
1401 wal.sync();
1402 wal.shutdown();
1403
1404
1405 this.abortRSAndWaitForRecovery(hrs, zkw, NUM_REGIONS_TO_CREATE);
1406
1407
1408 LOG.info("Verification Starts...");
1409 Get g = new Get(row);
1410 Result r = ht.get(g);
1411 long theStoredVal = Bytes.toLong(r.getValue(family, qualifier));
1412 assertEquals(value, theStoredVal);
1413
1414
1415 LOG.info("Verification after flush...");
1416 TEST_UTIL.getHBaseAdmin().flush(tableName);
1417 TEST_UTIL.getHBaseAdmin().compact(tableName);
1418
1419
1420 TEST_UTIL.waitFor(30000, 200, new Waiter.Predicate<Exception>() {
1421 @Override
1422 public boolean evaluate() throws Exception {
1423 return (TEST_UTIL.getHBaseAdmin().getCompactionState(tableName) == CompactionState.NONE);
1424 }
1425 });
1426
1427 r = ht.get(g);
1428 theStoredVal = Bytes.toLong(r.getValue(family, qualifier));
1429 assertEquals(value, theStoredVal);
1430 } finally {
1431 if (ht != null) ht.close();
1432 if (zkw != null) zkw.close();
1433 }
1434 }
1435
1436 @Test(timeout = 300000)
1437 public void testReadWriteSeqIdFiles() throws Exception {
1438 LOG.info("testReadWriteSeqIdFiles");
1439 startCluster(2);
1440 final ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "table-creation", null);
1441 Table ht = installTable(zkw, "table", "family", 10);
1442 try {
1443 FileSystem fs = master.getMasterFileSystem().getFileSystem();
1444 Path tableDir = FSUtils.getTableDir(FSUtils.getRootDir(conf), TableName.valueOf("table"));
1445 List<Path> regionDirs = FSUtils.getRegionDirs(fs, tableDir);
1446 long newSeqId = WALSplitter.writeRegionSequenceIdFile(fs, regionDirs.get(0), 1L, 1000L);
1447 WALSplitter.writeRegionSequenceIdFile(fs, regionDirs.get(0) , 1L, 1000L);
1448 assertEquals(newSeqId + 2000,
1449 WALSplitter.writeRegionSequenceIdFile(fs, regionDirs.get(0), 3L, 1000L));
1450
1451 Path editsdir = WALSplitter.getRegionDirRecoveredEditsDir(regionDirs.get(0));
1452 FileStatus[] files = FSUtils.listStatus(fs, editsdir, new PathFilter() {
1453 @Override
1454 public boolean accept(Path p) {
1455 return WALSplitter.isSequenceIdFile(p);
1456 }
1457 });
1458
1459 assertEquals(1, files.length);
1460
1461
1462 NavigableSet<Path> recoveredEdits =
1463 WALSplitter.getSplitEditFilesSorted(fs, regionDirs.get(0));
1464 assertEquals(0, recoveredEdits.size());
1465 } finally {
1466 if (ht != null) ht.close();
1467 if (zkw != null) zkw.close();
1468 }
1469 }
1470
1471 Table installTable(ZooKeeperWatcher zkw, String tname, String fname, int nrs) throws Exception {
1472 return installTable(zkw, tname, fname, nrs, 0);
1473 }
1474
1475 Table installTable(ZooKeeperWatcher zkw, String tname, String fname, int nrs,
1476 int existingRegions) throws Exception {
1477
1478 TableName table = TableName.valueOf(tname);
1479 byte [] family = Bytes.toBytes(fname);
1480 LOG.info("Creating table with " + nrs + " regions");
1481 Table ht = TEST_UTIL.createMultiRegionTable(table, family, nrs);
1482 int numRegions = -1;
1483 try (RegionLocator r = TEST_UTIL.getConnection().getRegionLocator(table)) {
1484 numRegions = r.getStartKeys().length;
1485 }
1486 assertEquals(nrs, numRegions);
1487 LOG.info("Waiting for no more RIT\n");
1488 blockUntilNoRIT(zkw, master);
1489
1490
1491 LOG.debug("Disabling table\n");
1492 TEST_UTIL.getHBaseAdmin().disableTable(table);
1493 LOG.debug("Waiting for no more RIT\n");
1494 blockUntilNoRIT(zkw, master);
1495 NavigableSet<String> regions = HBaseTestingUtility.getAllOnlineRegions(cluster);
1496 LOG.debug("Verifying only catalog and namespace regions are assigned\n");
1497 if (regions.size() != 2) {
1498 for (String oregion : regions)
1499 LOG.debug("Region still online: " + oregion);
1500 }
1501 assertEquals(2 + existingRegions, regions.size());
1502 LOG.debug("Enabling table\n");
1503 TEST_UTIL.getHBaseAdmin().enableTable(table);
1504 LOG.debug("Waiting for no more RIT\n");
1505 blockUntilNoRIT(zkw, master);
1506 LOG.debug("Verifying there are " + numRegions + " assigned on cluster\n");
1507 regions = HBaseTestingUtility.getAllOnlineRegions(cluster);
1508 assertEquals(numRegions + 2 + existingRegions, regions.size());
1509 return ht;
1510 }
1511
1512 void populateDataInTable(int nrows, String fname) throws Exception {
1513 byte [] family = Bytes.toBytes(fname);
1514
1515 List<RegionServerThread> rsts = cluster.getLiveRegionServerThreads();
1516 assertEquals(NUM_RS, rsts.size());
1517
1518 for (RegionServerThread rst : rsts) {
1519 HRegionServer hrs = rst.getRegionServer();
1520 List<HRegionInfo> hris = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
1521 for (HRegionInfo hri : hris) {
1522 if (hri.getTable().isSystemTable()) {
1523 continue;
1524 }
1525 LOG.debug("adding data to rs = " + rst.getName() +
1526 " region = "+ hri.getRegionNameAsString());
1527 Region region = hrs.getOnlineRegion(hri.getRegionName());
1528 assertTrue(region != null);
1529 putData(region, hri.getStartKey(), nrows, Bytes.toBytes("q"), family);
1530 }
1531 }
1532
1533 for (MasterThread mt : cluster.getLiveMasterThreads()) {
1534 HRegionServer hrs = mt.getMaster();
1535 List<HRegionInfo> hris;
1536 try {
1537 hris = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
1538 } catch (ServerNotRunningYetException e) {
1539
1540 continue;
1541 }
1542 for (HRegionInfo hri : hris) {
1543 if (hri.getTable().isSystemTable()) {
1544 continue;
1545 }
1546 LOG.debug("adding data to rs = " + mt.getName() +
1547 " region = "+ hri.getRegionNameAsString());
1548 Region region = hrs.getOnlineRegion(hri.getRegionName());
1549 assertTrue(region != null);
1550 putData(region, hri.getStartKey(), nrows, Bytes.toBytes("q"), family);
1551 }
1552 }
1553 }
1554
1555 public void makeWAL(HRegionServer hrs, List<HRegionInfo> regions, String tname, String fname,
1556 int num_edits, int edit_size) throws IOException {
1557 makeWAL(hrs, regions, tname, fname, num_edits, edit_size, true);
1558 }
1559
1560 public void makeWAL(HRegionServer hrs, List<HRegionInfo> regions, String tname, String fname,
1561 int num_edits, int edit_size, boolean cleanShutdown) throws IOException {
1562 TableName fullTName = TableName.valueOf(tname);
1563
1564 regions.remove(HRegionInfo.FIRST_META_REGIONINFO);
1565
1566 final AtomicLong sequenceId = new AtomicLong(10);
1567
1568
1569 for(Iterator<HRegionInfo> iter = regions.iterator(); iter.hasNext(); ) {
1570 HRegionInfo regionInfo = iter.next();
1571 if(regionInfo.getTable().isSystemTable()) {
1572 iter.remove();
1573 }
1574 }
1575 HTableDescriptor htd = new HTableDescriptor(fullTName);
1576 byte[] family = Bytes.toBytes(fname);
1577 htd.addFamily(new HColumnDescriptor(family));
1578 byte[] value = new byte[edit_size];
1579
1580 List<HRegionInfo> hris = new ArrayList<HRegionInfo>();
1581 for (HRegionInfo region : regions) {
1582 if (!region.getTable().getNameAsString().equalsIgnoreCase(tname)) {
1583 continue;
1584 }
1585 hris.add(region);
1586 }
1587 LOG.info("Creating wal edits across " + hris.size() + " regions.");
1588 for (int i = 0; i < edit_size; i++) {
1589 value[i] = (byte) ('a' + (i % 26));
1590 }
1591 int n = hris.size();
1592 int[] counts = new int[n];
1593
1594 final int syncEvery = 30 * 1024 / edit_size;
1595 if (n > 0) {
1596 for (int i = 0; i < num_edits; i += 1) {
1597 WALEdit e = new WALEdit();
1598 HRegionInfo curRegionInfo = hris.get(i % n);
1599 final WAL log = hrs.getWAL(curRegionInfo);
1600 byte[] startRow = curRegionInfo.getStartKey();
1601 if (startRow == null || startRow.length == 0) {
1602 startRow = new byte[] { 0, 0, 0, 0, 1 };
1603 }
1604 byte[] row = Bytes.incrementBytes(startRow, counts[i % n]);
1605 row = Arrays.copyOfRange(row, 3, 8);
1606
1607
1608 byte[] qualifier = Bytes.toBytes("c" + Integer.toString(i));
1609 e.add(new KeyValue(row, family, qualifier, System.currentTimeMillis(), value));
1610 log.append(htd, curRegionInfo,
1611 new HLogKey(curRegionInfo.getEncodedNameAsBytes(), fullTName,
1612 System.currentTimeMillis()), e, sequenceId, true, null);
1613 if (0 == i % syncEvery) {
1614 log.sync();
1615 }
1616 counts[i % n] += 1;
1617 }
1618 }
1619
1620
1621 for (HRegionInfo info : hris) {
1622 final WAL log = hrs.getWAL(info);
1623 log.sync();
1624 }
1625 if (cleanShutdown) {
1626 for (HRegionInfo info : hris) {
1627 final WAL log = hrs.getWAL(info);
1628 log.shutdown();
1629 }
1630 }
1631 for (int i = 0; i < n; i++) {
1632 LOG.info("region " + hris.get(i).getRegionNameAsString() + " has " + counts[i] + " edits");
1633 }
1634 return;
1635 }
1636
1637 private int countWAL(Path log, FileSystem fs, Configuration conf)
1638 throws IOException {
1639 int count = 0;
1640 WAL.Reader in = WALFactory.createReader(fs, log, conf);
1641 try {
1642 WAL.Entry e;
1643 while ((e = in.next()) != null) {
1644 if (!WALEdit.isMetaEditFamily(e.getEdit().getCells().get(0))) {
1645 count++;
1646 }
1647 }
1648 } finally {
1649 try {
1650 in.close();
1651 } catch (IOException exception) {
1652 LOG.warn("Problem closing wal: " + exception.getMessage());
1653 LOG.debug("exception details.", exception);
1654 }
1655 }
1656 return count;
1657 }
1658
1659 private void blockUntilNoRIT(ZooKeeperWatcher zkw, HMaster master) throws Exception {
1660 TEST_UTIL.waitUntilNoRegionsInTransition(60000);
1661 }
1662
1663 private void putData(Region region, byte[] startRow, int numRows, byte [] qf,
1664 byte [] ...families)
1665 throws IOException {
1666 for(int i = 0; i < numRows; i++) {
1667 Put put = new Put(Bytes.add(startRow, Bytes.toBytes(i)));
1668 for(byte [] family : families) {
1669 put.add(family, qf, null);
1670 }
1671 region.put(put);
1672 }
1673 }
1674
1675
1676
1677
1678 private void prepareData(final Table t, final byte[] f, final byte[] column) throws IOException {
1679 byte[] k = new byte[3];
1680
1681
1682 List<Put> puts = new ArrayList<>();
1683 for (byte b1 = 'a'; b1 <= 'z'; b1++) {
1684 for (byte b2 = 'a'; b2 <= 'z'; b2++) {
1685 for (byte b3 = 'a'; b3 <= 'z'; b3++) {
1686 k[0] = b1;
1687 k[1] = b2;
1688 k[2] = b3;
1689 Put put = new Put(k);
1690 put.add(f, column, k);
1691 puts.add(put);
1692 }
1693 }
1694 }
1695 t.put(puts);
1696
1697 for (byte b3 = 'a'; b3 <= 'z'; b3++) {
1698 k[0] = 'a';
1699 k[1] = 'a';
1700 k[2] = b3;
1701 Delete del = new Delete(k);
1702 t.delete(del);
1703 }
1704 }
1705
1706 private void waitForCounter(AtomicLong ctr, long oldval, long newval,
1707 long timems) {
1708 long curt = System.currentTimeMillis();
1709 long endt = curt + timems;
1710 while (curt < endt) {
1711 if (ctr.get() == oldval) {
1712 Thread.yield();
1713 curt = System.currentTimeMillis();
1714 } else {
1715 assertEquals(newval, ctr.get());
1716 return;
1717 }
1718 }
1719 assertTrue(false);
1720 }
1721
1722 private void abortMaster(MiniHBaseCluster cluster) throws InterruptedException {
1723 for (MasterThread mt : cluster.getLiveMasterThreads()) {
1724 if (mt.getMaster().isActiveMaster()) {
1725 mt.getMaster().abort("Aborting for tests", new Exception("Trace info"));
1726 mt.join();
1727 break;
1728 }
1729 }
1730 LOG.debug("Master is aborted");
1731 }
1732
1733
1734
1735
1736
1737
1738
1739
1740 private HRegionServer findRSToKill(boolean hasMetaRegion, String tableName) throws Exception {
1741 List<RegionServerThread> rsts = cluster.getLiveRegionServerThreads();
1742 List<HRegionInfo> regions = null;
1743 HRegionServer hrs = null;
1744
1745 for (RegionServerThread rst: rsts) {
1746 hrs = rst.getRegionServer();
1747 while (rst.isAlive() && !hrs.isOnline()) {
1748 Thread.sleep(100);
1749 }
1750 if (!rst.isAlive()) {
1751 continue;
1752 }
1753 boolean isCarryingMeta = false;
1754 boolean foundTableRegion = false;
1755 regions = ProtobufUtil.getOnlineRegions(hrs.getRSRpcServices());
1756 for (HRegionInfo region : regions) {
1757 if (region.isMetaRegion()) {
1758 isCarryingMeta = true;
1759 }
1760 if (tableName == null || region.getTable().getNameAsString().equals(tableName)) {
1761 foundTableRegion = true;
1762 }
1763 if (foundTableRegion && (isCarryingMeta || !hasMetaRegion)) {
1764 break;
1765 }
1766 }
1767 if (isCarryingMeta && hasMetaRegion) {
1768
1769 if (!foundTableRegion) {
1770 final HRegionServer destRS = hrs;
1771
1772 List<HRegionInfo> tableRegions =
1773 TEST_UTIL.getHBaseAdmin().getTableRegions(TableName.valueOf(tableName));
1774 final HRegionInfo hri = tableRegions.get(0);
1775 TEST_UTIL.getHBaseAdmin().move(hri.getEncodedNameAsBytes(),
1776 Bytes.toBytes(destRS.getServerName().getServerName()));
1777
1778 final RegionStates regionStates =
1779 TEST_UTIL.getHBaseCluster().getMaster().getAssignmentManager().getRegionStates();
1780 TEST_UTIL.waitFor(45000, 200, new Waiter.Predicate<Exception>() {
1781 @Override
1782 public boolean evaluate() throws Exception {
1783 ServerName sn = regionStates.getRegionServerOfRegion(hri);
1784 return (sn != null && sn.equals(destRS.getServerName()));
1785 }
1786 });
1787 }
1788 return hrs;
1789 } else if (hasMetaRegion || isCarryingMeta) {
1790 continue;
1791 }
1792 if (foundTableRegion) break;
1793 }
1794
1795 return hrs;
1796 }
1797
1798 }