1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.replication;
19
20 import static org.junit.Assert.assertArrayEquals;
21 import static org.junit.Assert.assertEquals;
22 import static org.junit.Assert.assertNotNull;
23 import static org.junit.Assert.fail;
24
25 import java.io.Closeable;
26 import java.io.IOException;
27 import java.util.Arrays;
28 import java.util.List;
29 import java.util.Random;
30 import java.util.concurrent.CountDownLatch;
31
32 import org.apache.commons.logging.Log;
33 import org.apache.commons.logging.LogFactory;
34 import org.apache.hadoop.conf.Configuration;
35 import org.apache.hadoop.fs.FileSystem;
36 import org.apache.hadoop.fs.Path;
37 import org.apache.hadoop.hbase.Cell;
38 import org.apache.hadoop.hbase.ClusterStatus;
39 import org.apache.hadoop.hbase.HBaseConfiguration;
40 import org.apache.hadoop.hbase.HBaseTestingUtility;
41 import org.apache.hadoop.hbase.HColumnDescriptor;
42 import org.apache.hadoop.hbase.HConstants;
43 import org.apache.hadoop.hbase.HTableDescriptor;
44 import org.apache.hadoop.hbase.KeyValue;
45 import org.apache.hadoop.hbase.MiniHBaseCluster;
46 import org.apache.hadoop.hbase.ServerLoad;
47 import org.apache.hadoop.hbase.ServerName;
48 import org.apache.hadoop.hbase.TableName;
49 import org.apache.hadoop.hbase.Waiter;
50 import org.apache.hadoop.hbase.client.Admin;
51 import org.apache.hadoop.hbase.client.Delete;
52 import org.apache.hadoop.hbase.client.Durability;
53 import org.apache.hadoop.hbase.client.Get;
54 import org.apache.hadoop.hbase.client.HBaseAdmin;
55 import org.apache.hadoop.hbase.client.HTable;
56 import org.apache.hadoop.hbase.client.Put;
57 import org.apache.hadoop.hbase.client.Result;
58 import org.apache.hadoop.hbase.client.Table;
59 import org.apache.hadoop.hbase.client.replication.ReplicationAdmin;
60 import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
61 import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
62 import org.apache.hadoop.hbase.coprocessor.ObserverContext;
63 import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
64 import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles;
65 import org.apache.hadoop.hbase.regionserver.HRegion;
66 import org.apache.hadoop.hbase.regionserver.HRegionServer;
67 import org.apache.hadoop.hbase.regionserver.RSRpcServices;
68 import org.apache.hadoop.hbase.regionserver.wal.WALActionsListener;
69 import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
70 import org.apache.hadoop.hbase.replication.regionserver.TestSourceFSConfigurationProvider;
71 import org.apache.hadoop.hbase.testclassification.LargeTests;
72 import org.apache.hadoop.hbase.util.Bytes;
73 import org.apache.hadoop.hbase.util.HFileTestUtil;
74 import org.apache.hadoop.hbase.zookeeper.MiniZooKeeperCluster;
75 import org.apache.hadoop.hbase.zookeeper.ZKUtil;
76 import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
77 import org.junit.After;
78 import org.junit.Before;
79 import org.junit.Test;
80 import org.junit.experimental.categories.Category;
81
82 import com.google.protobuf.ServiceException;
83
84 @Category(LargeTests.class)
85 public class TestMasterReplication {
86
87 private static final Log LOG = LogFactory.getLog(TestReplicationBase.class);
88
89 private Configuration baseConfiguration;
90
91 private HBaseTestingUtility[] utilities;
92 private Configuration[] configurations;
93 private MiniZooKeeperCluster miniZK;
94
95 private static final long SLEEP_TIME = 1000;
96 private static final int NB_RETRIES = 120;
97
98 private static final TableName tableName = TableName.valueOf("test");
99 private static final byte[] famName = Bytes.toBytes("f");
100 private static final byte[] famName1 = Bytes.toBytes("f1");
101 private static final byte[] row = Bytes.toBytes("row");
102 private static final byte[] row1 = Bytes.toBytes("row1");
103 private static final byte[] row2 = Bytes.toBytes("row2");
104 private static final byte[] row3 = Bytes.toBytes("row3");
105 private static final byte[] row4 = Bytes.toBytes("row4");
106 private static final byte[] noRepfamName = Bytes.toBytes("norep");
107
108 private static final byte[] count = Bytes.toBytes("count");
109 private static final byte[] put = Bytes.toBytes("put");
110 private static final byte[] delete = Bytes.toBytes("delete");
111
112 private HTableDescriptor table;
113
114 @Before
115 public void setUp() throws Exception {
116 baseConfiguration = HBaseConfiguration.create();
117
118
119 baseConfiguration.setInt("hbase.regionserver.hlog.blocksize", 1024 * 20);
120 baseConfiguration.setInt("replication.source.size.capacity", 1024);
121 baseConfiguration.setLong("replication.source.sleepforretries", 100);
122 baseConfiguration.setInt("hbase.regionserver.maxlogs", 10);
123 baseConfiguration.setLong("hbase.master.logcleaner.ttl", 10);
124 baseConfiguration.setBoolean(HConstants.REPLICATION_ENABLE_KEY,
125 HConstants.REPLICATION_ENABLE_DEFAULT);
126 baseConfiguration.setBoolean(HConstants.REPLICATION_BULKLOAD_ENABLE_KEY, true);
127 baseConfiguration.set("hbase.replication.source.fs.conf.provider",
128 TestSourceFSConfigurationProvider.class.getCanonicalName());
129 baseConfiguration.set(HConstants.REPLICATION_CLUSTER_ID, "12345");
130 baseConfiguration.setBoolean("dfs.support.append", true);
131 baseConfiguration.setLong(HConstants.THREAD_WAKE_FREQUENCY, 100);
132 baseConfiguration.setStrings(
133 CoprocessorHost.USER_REGION_COPROCESSOR_CONF_KEY,
134 CoprocessorCounter.class.getName());
135
136 table = new HTableDescriptor(tableName);
137 HColumnDescriptor fam = new HColumnDescriptor(famName);
138 fam.setScope(HConstants.REPLICATION_SCOPE_GLOBAL);
139 table.addFamily(fam);
140 fam = new HColumnDescriptor(famName1);
141 fam.setScope(HConstants.REPLICATION_SCOPE_GLOBAL);
142 table.addFamily(fam);
143 fam = new HColumnDescriptor(noRepfamName);
144 table.addFamily(fam);
145 }
146
147
148
149
150
151
152
153 @Test(timeout = 300000)
154 public void testCyclicReplication1() throws Exception {
155 LOG.info("testSimplePutDelete");
156 int numClusters = 2;
157 Table[] htables = null;
158 try {
159 htables = setUpClusterTablesAndPeers(numClusters);
160
161 int[] expectedCounts = new int[] { 2, 2 };
162
163
164
165 putAndWait(row, famName, htables[0], htables[1]);
166 putAndWait(row1, famName, htables[1], htables[0]);
167 validateCounts(htables, put, expectedCounts);
168
169 deleteAndWait(row, htables[0], htables[1]);
170 deleteAndWait(row1, htables[1], htables[0]);
171 validateCounts(htables, delete, expectedCounts);
172 } finally {
173 close(htables);
174 shutDownMiniClusters();
175 }
176 }
177
178
179
180
181
182
183 @Test(timeout = 300000)
184 public void testLoopedReplication() throws Exception {
185 LOG.info("testLoopedReplication");
186 startMiniClusters(1);
187 createTableOnClusters(table);
188 addPeer("1", 0, 0);
189 Thread.sleep(SLEEP_TIME);
190
191
192 final ServerName rsName = utilities[0].getHBaseCluster().getRegionServer(0).getServerName();
193 Waiter.waitFor(baseConfiguration, 10000, new Waiter.Predicate<Exception>() {
194 @Override
195 public boolean evaluate() throws Exception {
196 ClusterStatus clusterStatus = utilities[0].getHBaseAdmin().getClusterStatus();
197 ServerLoad serverLoad = clusterStatus.getLoad(rsName);
198 List<ReplicationLoadSource> replicationLoadSourceList =
199 serverLoad.getReplicationLoadSourceList();
200 return replicationLoadSourceList.size() == 0;
201 }
202 });
203
204 Table[] htables = getHTablesOnClusters(tableName);
205 putAndWait(row, famName, htables[0], htables[0]);
206 rollWALAndWait(utilities[0], table.getTableName(), row);
207 ZooKeeperWatcher zkw = utilities[0].getZooKeeperWatcher();
208 String queuesZnode =
209 ZKUtil.joinZNode(zkw.baseZNode, ZKUtil.joinZNode("replication", "rs"));
210 List<String> listChildrenNoWatch =
211 ZKUtil.listChildrenNoWatch(zkw, ZKUtil.joinZNode(queuesZnode, rsName.toString()));
212 assertEquals(0, listChildrenNoWatch.size());
213 }
214
215
216
217
218
219 @Test(timeout = 300000)
220 public void testHFileCyclicReplication() throws Exception {
221 LOG.info("testHFileCyclicReplication");
222 int numClusters = 2;
223 Table[] htables = null;
224 try {
225 htables = setUpClusterTablesAndPeers(numClusters);
226
227
228
229 byte[][][] hfileRanges =
230 new byte[][][] { new byte[][] { Bytes.toBytes("aaaa"), Bytes.toBytes("cccc") },
231 new byte[][] { Bytes.toBytes("ddd"), Bytes.toBytes("fff") }, };
232 int numOfRows = 100;
233 int[] expectedCounts =
234 new int[] { hfileRanges.length * numOfRows, hfileRanges.length * numOfRows };
235
236 loadAndValidateHFileReplication("testHFileCyclicReplication_01", 0, new int[] { 1 }, row,
237 famName, htables, hfileRanges, numOfRows, expectedCounts, true);
238
239
240
241 hfileRanges = new byte[][][] { new byte[][] { Bytes.toBytes("gggg"), Bytes.toBytes("iiii") },
242 new byte[][] { Bytes.toBytes("jjj"), Bytes.toBytes("lll") }, };
243 numOfRows = 200;
244 int[] newExpectedCounts = new int[] { hfileRanges.length * numOfRows + expectedCounts[0],
245 hfileRanges.length * numOfRows + expectedCounts[1] };
246
247 loadAndValidateHFileReplication("testHFileCyclicReplication_10", 1, new int[] { 0 }, row,
248 famName, htables, hfileRanges, numOfRows, newExpectedCounts, true);
249
250 } finally {
251 close(htables);
252 shutDownMiniClusters();
253 }
254 }
255
256 private Table[] setUpClusterTablesAndPeers(int numClusters) throws Exception {
257 Table[] htables;
258 startMiniClusters(numClusters);
259 createTableOnClusters(table);
260
261 htables = getHTablesOnClusters(tableName);
262
263 addPeer("1", 0, 1);
264 addPeer("1", 1, 0);
265 return htables;
266 }
267
268
269
270
271
272
273
274
275 @Test(timeout = 300000)
276 public void testCyclicReplication2() throws Exception {
277 LOG.info("testCyclicReplication1");
278 int numClusters = 3;
279 Table[] htables = null;
280 try {
281 startMiniClusters(numClusters);
282 createTableOnClusters(table);
283
284
285 addPeer("1", 0, 1);
286 addPeer("1", 1, 2);
287 addPeer("1", 2, 0);
288
289 htables = getHTablesOnClusters(tableName);
290
291
292 putAndWait(row, famName, htables[0], htables[2]);
293 putAndWait(row1, famName, htables[1], htables[0]);
294 putAndWait(row2, famName, htables[2], htables[1]);
295
296 deleteAndWait(row, htables[0], htables[2]);
297 deleteAndWait(row1, htables[1], htables[0]);
298 deleteAndWait(row2, htables[2], htables[1]);
299
300 int[] expectedCounts = new int[] { 3, 3, 3 };
301 validateCounts(htables, put, expectedCounts);
302 validateCounts(htables, delete, expectedCounts);
303
304
305 disablePeer("1", 2);
306
307
308 putAndWait(row3, famName, htables[0], htables[1]);
309
310 htables[1].put(new Put(row4).add(famName, row4, row4));
311
312 enablePeer("1", 2);
313
314
315
316 wait(row4, htables[0], true);
317 } finally {
318 close(htables);
319 shutDownMiniClusters();
320 }
321 }
322
323
324
325
326
327 @Test(timeout = 300000)
328 public void testHFileMultiSlaveReplication() throws Exception {
329 LOG.info("testHFileMultiSlaveReplication");
330 int numClusters = 3;
331 Table[] htables = null;
332 try {
333 startMiniClusters(numClusters);
334 createTableOnClusters(table);
335
336
337 addPeer("1", 0, 1);
338
339 htables = getHTablesOnClusters(tableName);
340
341
342
343 byte[][][] hfileRanges =
344 new byte[][][] { new byte[][] { Bytes.toBytes("mmmm"), Bytes.toBytes("oooo") },
345 new byte[][] { Bytes.toBytes("ppp"), Bytes.toBytes("rrr") }, };
346 int numOfRows = 100;
347
348 int[] expectedCounts =
349 new int[] { hfileRanges.length * numOfRows, hfileRanges.length * numOfRows };
350
351 loadAndValidateHFileReplication("testHFileCyclicReplication_0", 0, new int[] { 1 }, row,
352 famName, htables, hfileRanges, numOfRows, expectedCounts, true);
353
354
355 assertEquals(0, utilities[2].countRows(htables[2]));
356
357 rollWALAndWait(utilities[0], htables[0].getName(), row);
358
359
360 addPeer("2", 0, 2);
361
362
363
364 hfileRanges = new byte[][][] { new byte[][] { Bytes.toBytes("ssss"), Bytes.toBytes("uuuu") },
365 new byte[][] { Bytes.toBytes("vvv"), Bytes.toBytes("xxx") }, };
366 numOfRows = 200;
367
368 int[] newExpectedCounts = new int[] { hfileRanges.length * numOfRows + expectedCounts[0],
369 hfileRanges.length * numOfRows + expectedCounts[1], hfileRanges.length * numOfRows };
370
371 loadAndValidateHFileReplication("testHFileCyclicReplication_1", 0, new int[] { 1, 2 }, row,
372 famName, htables, hfileRanges, numOfRows, newExpectedCounts, true);
373
374 } finally {
375 close(htables);
376 shutDownMiniClusters();
377 }
378 }
379
380
381
382
383
384
385 @Test(timeout = 300000)
386 public void testHFileReplicationForConfiguredTableCfs() throws Exception {
387 LOG.info("testHFileReplicationForConfiguredTableCfs");
388 int numClusters = 2;
389 Table[] htables = null;
390 try {
391 startMiniClusters(numClusters);
392 createTableOnClusters(table);
393
394 htables = getHTablesOnClusters(tableName);
395
396 addPeer("1", 0, 1, tableName.getNameAsString() + ":" + Bytes.toString(famName));
397
398
399 byte[][][] hfileRanges =
400 new byte[][][] { new byte[][] { Bytes.toBytes("aaaa"), Bytes.toBytes("cccc") },
401 new byte[][] { Bytes.toBytes("ddd"), Bytes.toBytes("fff") }, };
402 int numOfRows = 100;
403 int[] expectedCounts =
404 new int[] { hfileRanges.length * numOfRows, hfileRanges.length * numOfRows };
405
406 loadAndValidateHFileReplication("load_f", 0, new int[] { 1 }, row, famName, htables,
407 hfileRanges, numOfRows, expectedCounts, true);
408
409
410 hfileRanges = new byte[][][] { new byte[][] { Bytes.toBytes("gggg"), Bytes.toBytes("iiii") },
411 new byte[][] { Bytes.toBytes("jjj"), Bytes.toBytes("lll") }, };
412 numOfRows = 100;
413
414 int[] newExpectedCounts =
415 new int[] { hfileRanges.length * numOfRows + expectedCounts[0], expectedCounts[1] };
416
417 loadAndValidateHFileReplication("load_f1", 0, new int[] { 1 }, row, famName1, htables,
418 hfileRanges, numOfRows, newExpectedCounts, false);
419
420
421
422
423 wait(0, htables[0], hfileRanges.length * numOfRows + expectedCounts[0]);
424
425
426
427 Thread.sleep((NB_RETRIES / 2) * SLEEP_TIME);
428
429 wait(1, htables[1], expectedCounts[1]);
430 } finally {
431 close(htables);
432 shutDownMiniClusters();
433 }
434 }
435
436
437
438
439 @Test(timeout = 300000)
440 public void testCyclicReplication3() throws Exception {
441 LOG.info("testCyclicReplication2");
442 int numClusters = 3;
443 Table[] htables = null;
444 try {
445 startMiniClusters(numClusters);
446 createTableOnClusters(table);
447
448
449 addPeer("1", 0, 1);
450 addPeer("1", 1, 2);
451 addPeer("1", 2, 1);
452
453 htables = getHTablesOnClusters(tableName);
454
455
456 putAndWait(row, famName, htables[0], htables[2]);
457 putAndWait(row1, famName, htables[1], htables[2]);
458 putAndWait(row2, famName, htables[2], htables[1]);
459
460 deleteAndWait(row, htables[0], htables[2]);
461 deleteAndWait(row1, htables[1], htables[2]);
462 deleteAndWait(row2, htables[2], htables[1]);
463
464 int[] expectedCounts = new int[] { 1, 3, 3 };
465 validateCounts(htables, put, expectedCounts);
466 validateCounts(htables, delete, expectedCounts);
467 } finally {
468 close(htables);
469 shutDownMiniClusters();
470 }
471 }
472
473
474
475
476
477 @Test(timeout = 180000, expected = ServiceException.class)
478 public void testReplicateWALEntryWhenReplicationIsDisabled() throws Exception {
479 LOG.info("testSimplePutDelete");
480 baseConfiguration.setBoolean(HConstants.REPLICATION_ENABLE_KEY, false);
481 Table[] htables = null;
482 try {
483 startMiniClusters(1);
484 createTableOnClusters(table);
485 htables = getHTablesOnClusters(tableName);
486
487 HRegionServer rs = utilities[0].getRSForFirstRegionInTable(tableName);
488 RSRpcServices rsrpc = new RSRpcServices(rs);
489 rsrpc.replicateWALEntry(null, null);
490 } finally {
491 close(htables);
492 shutDownMiniClusters();
493 }
494 }
495
496 @After
497 public void tearDown() throws IOException {
498 configurations = null;
499 utilities = null;
500 }
501
502 @SuppressWarnings("resource")
503 private void startMiniClusters(int numClusters) throws Exception {
504 Random random = new Random();
505 utilities = new HBaseTestingUtility[numClusters];
506 configurations = new Configuration[numClusters];
507 for (int i = 0; i < numClusters; i++) {
508 Configuration conf = new Configuration(baseConfiguration);
509 conf.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/" + i + random.nextInt());
510 HBaseTestingUtility utility = new HBaseTestingUtility(conf);
511 if (i == 0) {
512 utility.startMiniZKCluster();
513 miniZK = utility.getZkCluster();
514 } else {
515 utility.setZkCluster(miniZK);
516 }
517 utility.startMiniCluster();
518 utilities[i] = utility;
519 configurations[i] = conf;
520 new ZooKeeperWatcher(conf, "cluster" + i, null, true);
521 }
522 }
523
524 private void shutDownMiniClusters() throws Exception {
525 int numClusters = utilities.length;
526 for (int i = numClusters - 1; i >= 0; i--) {
527 if (utilities[i] != null) {
528 utilities[i].shutdownMiniCluster();
529 }
530 }
531 miniZK.shutdown();
532 }
533
534 private void createTableOnClusters(HTableDescriptor table) throws Exception {
535 int numClusters = configurations.length;
536 for (int i = 0; i < numClusters; i++) {
537 Admin hbaseAdmin = null;
538 try {
539 hbaseAdmin = new HBaseAdmin(configurations[i]);
540 hbaseAdmin.createTable(table);
541 } finally {
542 close(hbaseAdmin);
543 }
544 }
545 }
546
547 private void addPeer(String id, int masterClusterNumber,
548 int slaveClusterNumber) throws Exception {
549 ReplicationAdmin replicationAdmin = null;
550 try {
551 replicationAdmin = new ReplicationAdmin(
552 configurations[masterClusterNumber]);
553 replicationAdmin.addPeer(id,
554 utilities[slaveClusterNumber].getClusterKey());
555 } finally {
556 close(replicationAdmin);
557 }
558 }
559
560 private void addPeer(String id, int masterClusterNumber, int slaveClusterNumber, String tableCfs)
561 throws Exception {
562 ReplicationAdmin replicationAdmin = null;
563 try {
564 replicationAdmin = new ReplicationAdmin(configurations[masterClusterNumber]);
565 replicationAdmin.addPeer(id, utilities[slaveClusterNumber].getClusterKey(), tableCfs);
566 } finally {
567 close(replicationAdmin);
568 }
569 }
570
571 private void disablePeer(String id, int masterClusterNumber) throws Exception {
572 ReplicationAdmin replicationAdmin = null;
573 try {
574 replicationAdmin = new ReplicationAdmin(
575 configurations[masterClusterNumber]);
576 replicationAdmin.disablePeer(id);
577 } finally {
578 close(replicationAdmin);
579 }
580 }
581
582 private void enablePeer(String id, int masterClusterNumber) throws Exception {
583 ReplicationAdmin replicationAdmin = null;
584 try {
585 replicationAdmin = new ReplicationAdmin(
586 configurations[masterClusterNumber]);
587 replicationAdmin.enablePeer(id);
588 } finally {
589 close(replicationAdmin);
590 }
591 }
592
593 private void close(Closeable... closeables) {
594 try {
595 if (closeables != null) {
596 for (Closeable closeable : closeables) {
597 closeable.close();
598 }
599 }
600 } catch (Exception e) {
601 LOG.warn("Exception occured while closing the object:", e);
602 }
603 }
604
605 @SuppressWarnings("resource")
606 private Table[] getHTablesOnClusters(TableName tableName) throws Exception {
607 int numClusters = utilities.length;
608 Table[] htables = new Table[numClusters];
609 for (int i = 0; i < numClusters; i++) {
610 Table htable = new HTable(configurations[i], tableName);
611 htable.setWriteBufferSize(1024);
612 htables[i] = htable;
613 }
614 return htables;
615 }
616
617 private void validateCounts(Table[] htables, byte[] type,
618 int[] expectedCounts) throws IOException {
619 for (int i = 0; i < htables.length; i++) {
620 assertEquals(Bytes.toString(type) + " were replicated back ",
621 expectedCounts[i], getCount(htables[i], type));
622 }
623 }
624
625 private int getCount(Table t, byte[] type) throws IOException {
626 Get test = new Get(row);
627 test.setAttribute("count", new byte[] {});
628 Result res = t.get(test);
629 return Bytes.toInt(res.getValue(count, type));
630 }
631
632 private void deleteAndWait(byte[] row, Table source, Table target)
633 throws Exception {
634 Delete del = new Delete(row);
635 source.delete(del);
636 wait(row, target, true);
637 }
638
639 private void putAndWait(byte[] row, byte[] fam, Table source, Table target)
640 throws Exception {
641 Put put = new Put(row);
642 put.add(fam, row, row);
643 source.put(put);
644 wait(row, target, false);
645 }
646
647 private void loadAndValidateHFileReplication(String testName, int masterNumber,
648 int[] slaveNumbers, byte[] row, byte[] fam, Table[] tables, byte[][][] hfileRanges,
649 int numOfRows, int[] expectedCounts, boolean toValidate) throws Exception {
650 HBaseTestingUtility util = utilities[masterNumber];
651
652 Path dir = util.getDataTestDirOnTestFS(testName);
653 FileSystem fs = util.getTestFileSystem();
654 dir = dir.makeQualified(fs);
655 Path familyDir = new Path(dir, Bytes.toString(fam));
656
657 int hfileIdx = 0;
658 for (byte[][] range : hfileRanges) {
659 byte[] from = range[0];
660 byte[] to = range[1];
661 HFileTestUtil.createHFile(util.getConfiguration(), fs,
662 new Path(familyDir, "hfile_" + hfileIdx++), fam, row, from, to, numOfRows);
663 }
664
665 Table source = tables[masterNumber];
666 final TableName tableName = source.getName();
667 LoadIncrementalHFiles loader = new LoadIncrementalHFiles(util.getConfiguration());
668 String[] args = { dir.toString(), tableName.toString() };
669 loader.run(args);
670
671 if (toValidate) {
672 for (int slaveClusterNumber : slaveNumbers) {
673 wait(slaveClusterNumber, tables[slaveClusterNumber], expectedCounts[slaveClusterNumber]);
674 }
675 }
676 }
677
678 private void wait(int slaveNumber, Table target, int expectedCount)
679 throws IOException, InterruptedException {
680 int count = 0;
681 for (int i = 0; i < NB_RETRIES; i++) {
682 if (i == NB_RETRIES - 1) {
683 fail("Waited too much time for bulkloaded data replication. Current count=" + count
684 + ", expected count=" + expectedCount);
685 }
686 count = utilities[slaveNumber].countRows(target);
687 if (count != expectedCount) {
688 LOG.info("Waiting more time for bulkloaded data replication.");
689 Thread.sleep(SLEEP_TIME);
690 } else {
691 break;
692 }
693 }
694 }
695
696 private void wait(byte[] row, Table target, boolean isDeleted) throws Exception {
697 Get get = new Get(row);
698 for (int i = 0; i < NB_RETRIES; i++) {
699 if (i == NB_RETRIES - 1) {
700 fail("Waited too much time for replication. Row:" + Bytes.toString(row)
701 + ". IsDeleteReplication:" + isDeleted);
702 }
703 Result res = target.get(get);
704 boolean sleep = isDeleted ? res.size() > 0 : res.size() == 0;
705 if (sleep) {
706 LOG.info("Waiting for more time for replication. Row:"
707 + Bytes.toString(row) + ". IsDeleteReplication:" + isDeleted);
708 Thread.sleep(SLEEP_TIME);
709 } else {
710 if (!isDeleted) {
711 assertArrayEquals(res.value(), row);
712 }
713 LOG.info("Obtained row:"
714 + Bytes.toString(row) + ". IsDeleteReplication:" + isDeleted);
715 break;
716 }
717 }
718 }
719
720 private void rollWALAndWait(final HBaseTestingUtility utility, final TableName table,
721 final byte[] row) throws IOException {
722 final Admin admin = utility.getHBaseAdmin();
723 final MiniHBaseCluster cluster = utility.getMiniHBaseCluster();
724
725
726 HRegion region = null;
727 for (HRegion candidate : cluster.getRegions(table)) {
728 if (HRegion.rowIsInRange(candidate.getRegionInfo(), row)) {
729 region = candidate;
730 break;
731 }
732 }
733 assertNotNull("Couldn't find the region for row '" + Arrays.toString(row) + "'", region);
734
735 final CountDownLatch latch = new CountDownLatch(1);
736
737
738 final WALActionsListener listener = new WALActionsListener.Base() {
739 @Override
740 public void postLogRoll(final Path oldPath, final Path newPath) throws IOException {
741 latch.countDown();
742 }
743 };
744 region.getWAL().registerWALActionsListener(listener);
745
746
747 admin.rollWALWriter(cluster.getServerHoldingRegion(region.getTableDesc().getTableName(),
748 region.getRegionInfo().getRegionName()));
749
750
751 try {
752 latch.await();
753 } catch (InterruptedException exception) {
754 LOG.warn("Interrupted while waiting for the wal of '" + region + "' to roll. If later " +
755 "replication tests fail, it's probably because we should still be waiting.");
756 Thread.currentThread().interrupt();
757 }
758 region.getWAL().unregisterWALActionsListener(listener);
759 }
760
761
762
763
764
765 public static class CoprocessorCounter extends BaseRegionObserver {
766 private int nCount = 0;
767 private int nDelete = 0;
768
769 @Override
770 public void prePut(final ObserverContext<RegionCoprocessorEnvironment> e, final Put put,
771 final WALEdit edit, final Durability durability) throws IOException {
772 nCount++;
773 }
774
775 @Override
776 public void postDelete(final ObserverContext<RegionCoprocessorEnvironment> c,
777 final Delete delete, final WALEdit edit, final Durability durability) throws IOException {
778 nDelete++;
779 }
780
781 @Override
782 public void preGetOp(final ObserverContext<RegionCoprocessorEnvironment> c,
783 final Get get, final List<Cell> result) throws IOException {
784 if (get.getAttribute("count") != null) {
785 result.clear();
786
787 result.add(new KeyValue(count, count, delete, Bytes.toBytes(nDelete)));
788 result.add(new KeyValue(count, count, put, Bytes.toBytes(nCount)));
789 c.bypass();
790 }
791 }
792 }
793
794 }