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 com.google.protobuf.ServiceException;
26
27 import java.io.Closeable;
28 import java.io.IOException;
29 import java.util.Arrays;
30 import java.util.List;
31 import java.util.Random;
32 import java.util.concurrent.CountDownLatch;
33
34 import org.apache.commons.logging.Log;
35 import org.apache.commons.logging.LogFactory;
36 import org.apache.hadoop.conf.Configuration;
37 import org.apache.hadoop.fs.FileSystem;
38 import org.apache.hadoop.fs.Path;
39 import org.apache.hadoop.hbase.Cell;
40 import org.apache.hadoop.hbase.ClusterStatus;
41 import org.apache.hadoop.hbase.HBaseConfiguration;
42 import org.apache.hadoop.hbase.HBaseTestingUtility;
43 import org.apache.hadoop.hbase.HColumnDescriptor;
44 import org.apache.hadoop.hbase.HConstants;
45 import org.apache.hadoop.hbase.HTableDescriptor;
46 import org.apache.hadoop.hbase.KeyValue;
47 import org.apache.hadoop.hbase.MiniHBaseCluster;
48 import org.apache.hadoop.hbase.ServerLoad;
49 import org.apache.hadoop.hbase.ServerName;
50 import org.apache.hadoop.hbase.TableName;
51 import org.apache.hadoop.hbase.Waiter;
52 import org.apache.hadoop.hbase.client.Admin;
53 import org.apache.hadoop.hbase.client.Delete;
54 import org.apache.hadoop.hbase.client.Durability;
55 import org.apache.hadoop.hbase.client.Get;
56 import org.apache.hadoop.hbase.client.HBaseAdmin;
57 import org.apache.hadoop.hbase.client.HTable;
58 import org.apache.hadoop.hbase.client.Put;
59 import org.apache.hadoop.hbase.client.Result;
60 import org.apache.hadoop.hbase.client.Table;
61 import org.apache.hadoop.hbase.client.replication.ReplicationAdmin;
62 import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
63 import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
64 import org.apache.hadoop.hbase.coprocessor.ObserverContext;
65 import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
66 import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles;
67 import org.apache.hadoop.hbase.regionserver.HRegion;
68 import org.apache.hadoop.hbase.regionserver.HRegionServer;
69 import org.apache.hadoop.hbase.regionserver.RSRpcServices;
70 import org.apache.hadoop.hbase.regionserver.wal.WALActionsListener;
71 import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
72 import org.apache.hadoop.hbase.replication.regionserver.TestSourceFSConfigurationProvider;
73 import org.apache.hadoop.hbase.testclassification.LargeTests;
74 import org.apache.hadoop.hbase.util.Bytes;
75 import org.apache.hadoop.hbase.util.HFileTestUtil;
76 import org.apache.hadoop.hbase.zookeeper.MiniZooKeeperCluster;
77 import org.apache.hadoop.hbase.zookeeper.ZKUtil;
78 import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
79 import org.junit.After;
80 import org.junit.Before;
81 import org.junit.Test;
82 import org.junit.experimental.categories.Category;
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("testCyclicReplication2");
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], false);
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 ReplicationPeerConfig rpc = new ReplicationPeerConfig();
554 rpc.setClusterKey(utilities[slaveClusterNumber].getClusterKey());
555 replicationAdmin.addPeer(id, rpc);
556 } finally {
557 close(replicationAdmin);
558 }
559 }
560
561 private void addPeer(String id, int masterClusterNumber, int slaveClusterNumber, String tableCfs)
562 throws Exception {
563 ReplicationAdmin replicationAdmin = null;
564 try {
565 replicationAdmin = new ReplicationAdmin(configurations[masterClusterNumber]);
566 ReplicationPeerConfig rpc = new ReplicationPeerConfig();
567 rpc.setClusterKey(utilities[slaveClusterNumber].getClusterKey());
568 replicationAdmin.addPeer(id, rpc, ReplicationSerDeHelper.parseTableCFsFromConfig(tableCfs));
569 } finally {
570 close(replicationAdmin);
571 }
572 }
573
574 private void disablePeer(String id, int masterClusterNumber) throws Exception {
575 ReplicationAdmin replicationAdmin = null;
576 try {
577 replicationAdmin = new ReplicationAdmin(
578 configurations[masterClusterNumber]);
579 replicationAdmin.disablePeer(id);
580 } finally {
581 close(replicationAdmin);
582 }
583 }
584
585 private void enablePeer(String id, int masterClusterNumber) throws Exception {
586 ReplicationAdmin replicationAdmin = null;
587 try {
588 replicationAdmin = new ReplicationAdmin(
589 configurations[masterClusterNumber]);
590 replicationAdmin.enablePeer(id);
591 } finally {
592 close(replicationAdmin);
593 }
594 }
595
596 private void close(Closeable... closeables) {
597 try {
598 if (closeables != null) {
599 for (Closeable closeable : closeables) {
600 closeable.close();
601 }
602 }
603 } catch (Exception e) {
604 LOG.warn("Exception occured while closing the object:", e);
605 }
606 }
607
608 @SuppressWarnings("resource")
609 private Table[] getHTablesOnClusters(TableName tableName) throws Exception {
610 int numClusters = utilities.length;
611 Table[] htables = new Table[numClusters];
612 for (int i = 0; i < numClusters; i++) {
613 Table htable = new HTable(configurations[i], tableName);
614 htable.setWriteBufferSize(1024);
615 htables[i] = htable;
616 }
617 return htables;
618 }
619
620 private void validateCounts(Table[] htables, byte[] type,
621 int[] expectedCounts) throws IOException {
622 for (int i = 0; i < htables.length; i++) {
623 assertEquals(Bytes.toString(type) + " were replicated back ",
624 expectedCounts[i], getCount(htables[i], type));
625 }
626 }
627
628 private int getCount(Table t, byte[] type) throws IOException {
629 Get test = new Get(row);
630 test.setAttribute("count", new byte[] {});
631 Result res = t.get(test);
632 return Bytes.toInt(res.getValue(count, type));
633 }
634
635 private void deleteAndWait(byte[] row, Table source, Table target)
636 throws Exception {
637 Delete del = new Delete(row);
638 source.delete(del);
639 wait(row, target, true);
640 }
641
642 private void putAndWait(byte[] row, byte[] fam, Table source, Table target)
643 throws Exception {
644 Put put = new Put(row);
645 put.add(fam, row, row);
646 source.put(put);
647 wait(row, target, false);
648 }
649
650 private void loadAndValidateHFileReplication(String testName, int masterNumber,
651 int[] slaveNumbers, byte[] row, byte[] fam, Table[] tables, byte[][][] hfileRanges,
652 int numOfRows, int[] expectedCounts, boolean toValidate) throws Exception {
653 HBaseTestingUtility util = utilities[masterNumber];
654
655 Path dir = util.getDataTestDirOnTestFS(testName);
656 FileSystem fs = util.getTestFileSystem();
657 dir = dir.makeQualified(fs);
658 Path familyDir = new Path(dir, Bytes.toString(fam));
659
660 int hfileIdx = 0;
661 for (byte[][] range : hfileRanges) {
662 byte[] from = range[0];
663 byte[] to = range[1];
664 HFileTestUtil.createHFile(util.getConfiguration(), fs,
665 new Path(familyDir, "hfile_" + hfileIdx++), fam, row, from, to, numOfRows);
666 }
667
668 Table source = tables[masterNumber];
669 final TableName tableName = source.getName();
670 LoadIncrementalHFiles loader = new LoadIncrementalHFiles(util.getConfiguration());
671 String[] args = { dir.toString(), tableName.toString() };
672 loader.run(args);
673
674 if (toValidate) {
675 for (int slaveClusterNumber : slaveNumbers) {
676 wait(slaveClusterNumber, tables[slaveClusterNumber], expectedCounts[slaveClusterNumber]);
677 }
678 }
679 }
680
681 private void wait(int slaveNumber, Table target, int expectedCount)
682 throws IOException, InterruptedException {
683 int count = 0;
684 for (int i = 0; i < NB_RETRIES; i++) {
685 if (i == NB_RETRIES - 1) {
686 fail("Waited too much time for bulkloaded data replication. Current count=" + count
687 + ", expected count=" + expectedCount);
688 }
689 count = utilities[slaveNumber].countRows(target);
690 if (count != expectedCount) {
691 LOG.info("Waiting more time for bulkloaded data replication.");
692 Thread.sleep(SLEEP_TIME);
693 } else {
694 break;
695 }
696 }
697 }
698
699 private void wait(byte[] row, Table target, boolean isDeleted) throws Exception {
700 Get get = new Get(row);
701 for (int i = 0; i < NB_RETRIES; i++) {
702 if (i == NB_RETRIES - 1) {
703 fail("Waited too much time for replication. Row:" + Bytes.toString(row)
704 + ". IsDeleteReplication:" + isDeleted);
705 }
706 Result res = target.get(get);
707 boolean sleep = isDeleted ? res.size() > 0 : res.size() == 0;
708 if (sleep) {
709 LOG.info("Waiting for more time for replication. Row:"
710 + Bytes.toString(row) + ". IsDeleteReplication:" + isDeleted);
711 Thread.sleep(SLEEP_TIME);
712 } else {
713 if (!isDeleted) {
714 assertArrayEquals(res.value(), row);
715 }
716 LOG.info("Obtained row:"
717 + Bytes.toString(row) + ". IsDeleteReplication:" + isDeleted);
718 break;
719 }
720 }
721 }
722
723 private void rollWALAndWait(final HBaseTestingUtility utility, final TableName table,
724 final byte[] row) throws IOException {
725 final Admin admin = utility.getHBaseAdmin();
726 final MiniHBaseCluster cluster = utility.getMiniHBaseCluster();
727
728
729 HRegion region = null;
730 for (HRegion candidate : cluster.getRegions(table)) {
731 if (HRegion.rowIsInRange(candidate.getRegionInfo(), row)) {
732 region = candidate;
733 break;
734 }
735 }
736 assertNotNull("Couldn't find the region for row '" + Arrays.toString(row) + "'", region);
737
738 final CountDownLatch latch = new CountDownLatch(1);
739
740
741 final WALActionsListener listener = new WALActionsListener.Base() {
742 @Override
743 public void postLogRoll(final Path oldPath, final Path newPath) throws IOException {
744 latch.countDown();
745 }
746 };
747 region.getWAL().registerWALActionsListener(listener);
748
749
750 admin.rollWALWriter(cluster.getServerHoldingRegion(region.getTableDesc().getTableName(),
751 region.getRegionInfo().getRegionName()));
752
753
754 try {
755 latch.await();
756 } catch (InterruptedException exception) {
757 LOG.warn("Interrupted while waiting for the wal of '" + region + "' to roll. If later " +
758 "replication tests fail, it's probably because we should still be waiting.");
759 Thread.currentThread().interrupt();
760 }
761 region.getWAL().unregisterWALActionsListener(listener);
762 }
763
764
765
766
767
768 public static class CoprocessorCounter extends BaseRegionObserver {
769 private int nCount = 0;
770 private int nDelete = 0;
771
772 @Override
773 public void prePut(final ObserverContext<RegionCoprocessorEnvironment> e, final Put put,
774 final WALEdit edit, final Durability durability) throws IOException {
775 nCount++;
776 }
777
778 @Override
779 public void postDelete(final ObserverContext<RegionCoprocessorEnvironment> c,
780 final Delete delete, final WALEdit edit, final Durability durability) throws IOException {
781 nDelete++;
782 }
783
784 @Override
785 public void preGetOp(final ObserverContext<RegionCoprocessorEnvironment> c,
786 final Get get, final List<Cell> result) throws IOException {
787 if (get.getAttribute("count") != null) {
788 result.clear();
789
790 result.add(new KeyValue(count, count, delete, Bytes.toBytes(nDelete)));
791 result.add(new KeyValue(count, count, put, Bytes.toBytes(nCount)));
792 c.bypass();
793 }
794 }
795 }
796
797 }