1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.regionserver.wal;
20
21 import static org.junit.Assert.assertEquals;
22 import static org.junit.Assert.assertNotNull;
23 import static org.junit.Assert.assertTrue;
24 import static org.junit.Assert.fail;
25 import static org.mockito.Mockito.when;
26
27 import java.io.IOException;
28 import java.security.PrivilegedExceptionAction;
29 import java.util.ArrayList;
30 import java.util.Collection;
31 import java.util.HashSet;
32 import java.util.List;
33 import java.util.Set;
34 import java.util.concurrent.atomic.AtomicBoolean;
35 import java.util.concurrent.atomic.AtomicInteger;
36 import java.util.concurrent.atomic.AtomicLong;
37
38 import org.apache.commons.logging.Log;
39 import org.apache.commons.logging.LogFactory;
40 import org.apache.hadoop.conf.Configuration;
41 import org.apache.hadoop.fs.FileStatus;
42 import org.apache.hadoop.fs.FileSystem;
43 import org.apache.hadoop.fs.Path;
44 import org.apache.hadoop.fs.PathFilter;
45 import org.apache.hadoop.hbase.Cell;
46 import org.apache.hadoop.hbase.HBaseConfiguration;
47 import org.apache.hadoop.hbase.HBaseTestingUtility;
48 import org.apache.hadoop.hbase.HColumnDescriptor;
49 import org.apache.hadoop.hbase.HConstants;
50 import org.apache.hadoop.hbase.HRegionInfo;
51 import org.apache.hadoop.hbase.HTableDescriptor;
52 import org.apache.hadoop.hbase.KeyValue;
53 import org.apache.hadoop.hbase.MasterNotRunningException;
54 import org.apache.hadoop.hbase.testclassification.MediumTests;
55 import org.apache.hadoop.hbase.MiniHBaseCluster;
56 import org.apache.hadoop.hbase.ServerName;
57 import org.apache.hadoop.hbase.TableName;
58 import org.apache.hadoop.hbase.ZooKeeperConnectionException;
59 import org.apache.hadoop.hbase.client.Delete;
60 import org.apache.hadoop.hbase.client.Get;
61 import org.apache.hadoop.hbase.client.HTable;
62 import org.apache.hadoop.hbase.client.Put;
63 import org.apache.hadoop.hbase.client.Result;
64 import org.apache.hadoop.hbase.client.ResultScanner;
65 import org.apache.hadoop.hbase.client.Scan;
66 import org.apache.hadoop.hbase.client.Table;
67 import org.apache.hadoop.hbase.master.HMaster;
68 import org.apache.hadoop.hbase.monitoring.MonitoredTask;
69 import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos.SplitLogTask.RecoveryMode;
70 import org.apache.hadoop.hbase.regionserver.DefaultStoreEngine;
71 import org.apache.hadoop.hbase.regionserver.DefaultStoreFlusher;
72 import org.apache.hadoop.hbase.regionserver.FlushRequestListener;
73 import org.apache.hadoop.hbase.regionserver.FlushRequester;
74 import org.apache.hadoop.hbase.regionserver.HRegion;
75 import org.apache.hadoop.hbase.regionserver.HRegionServer;
76 import org.apache.hadoop.hbase.regionserver.MemStoreSnapshot;
77 import org.apache.hadoop.hbase.regionserver.Region;
78 import org.apache.hadoop.hbase.regionserver.RegionScanner;
79 import org.apache.hadoop.hbase.regionserver.RegionServerServices;
80 import org.apache.hadoop.hbase.regionserver.Store;
81 import org.apache.hadoop.hbase.security.User;
82 import org.apache.hadoop.hbase.util.Bytes;
83 import org.apache.hadoop.hbase.util.EnvironmentEdge;
84 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
85 import org.apache.hadoop.hbase.util.FSUtils;
86 import org.apache.hadoop.hbase.util.HFileTestUtil;
87 import org.apache.hadoop.hbase.util.Pair;
88 import org.apache.hadoop.hbase.wal.DefaultWALProvider;
89 import org.apache.hadoop.hbase.wal.WAL;
90 import org.apache.hadoop.hbase.wal.WALFactory;
91 import org.apache.hadoop.hbase.wal.WALKey;
92 import org.apache.hadoop.hbase.wal.WALSplitter;
93 import org.junit.After;
94 import org.junit.AfterClass;
95 import org.junit.Before;
96 import org.junit.BeforeClass;
97 import org.junit.Rule;
98 import org.junit.Test;
99 import org.junit.experimental.categories.Category;
100 import org.junit.rules.TestName;
101 import org.mockito.Mockito;
102
103
104
105
106 @Category(MediumTests.class)
107 public class TestWALReplay {
108 public static final Log LOG = LogFactory.getLog(TestWALReplay.class);
109 static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
110 private final EnvironmentEdge ee = EnvironmentEdgeManager.getDelegate();
111 private Path hbaseRootDir = null;
112 private String logName;
113 private Path oldLogDir;
114 private Path logDir;
115 private FileSystem fs;
116 private Configuration conf;
117 private RecoveryMode mode;
118 private WALFactory wals;
119
120 @Rule
121 public final TestName currentTest = new TestName();
122
123
124 @BeforeClass
125 public static void setUpBeforeClass() throws Exception {
126 Configuration conf = TEST_UTIL.getConfiguration();
127 conf.setBoolean("dfs.support.append", true);
128
129 conf.setInt("dfs.client.block.recovery.retries", 2);
130 TEST_UTIL.startMiniCluster(3);
131 Path hbaseRootDir =
132 TEST_UTIL.getDFSCluster().getFileSystem().makeQualified(new Path("/hbase"));
133 LOG.info("hbase.rootdir=" + hbaseRootDir);
134 FSUtils.setRootDir(conf, hbaseRootDir);
135 }
136
137 @AfterClass
138 public static void tearDownAfterClass() throws Exception {
139 TEST_UTIL.shutdownMiniCluster();
140 }
141
142 @Before
143 public void setUp() throws Exception {
144 this.conf = HBaseConfiguration.create(TEST_UTIL.getConfiguration());
145 this.fs = TEST_UTIL.getDFSCluster().getFileSystem();
146 this.hbaseRootDir = FSUtils.getRootDir(this.conf);
147 this.oldLogDir = new Path(this.hbaseRootDir, HConstants.HREGION_OLDLOGDIR_NAME);
148 this.logName = DefaultWALProvider.getWALDirectoryName(currentTest.getMethodName() + "-manual");
149 this.logDir = new Path(this.hbaseRootDir, logName);
150 if (TEST_UTIL.getDFSCluster().getFileSystem().exists(this.hbaseRootDir)) {
151 TEST_UTIL.getDFSCluster().getFileSystem().delete(this.hbaseRootDir, true);
152 }
153 this.mode = (conf.getBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY, false) ?
154 RecoveryMode.LOG_REPLAY : RecoveryMode.LOG_SPLITTING);
155 this.wals = new WALFactory(conf, null, currentTest.getMethodName());
156 }
157
158 @After
159 public void tearDown() throws Exception {
160 this.wals.close();
161 TEST_UTIL.getDFSCluster().getFileSystem().delete(this.hbaseRootDir, true);
162 }
163
164
165
166
167 private void deleteDir(final Path p) throws IOException {
168 if (this.fs.exists(p)) {
169 if (!this.fs.delete(p, true)) {
170 throw new IOException("Failed remove of " + p);
171 }
172 }
173 }
174
175
176
177
178
179 @Test
180 public void testReplayEditsAfterRegionMovedWithMultiCF() throws Exception {
181 final TableName tableName =
182 TableName.valueOf("testReplayEditsAfterRegionMovedWithMultiCF");
183 byte[] family1 = Bytes.toBytes("cf1");
184 byte[] family2 = Bytes.toBytes("cf2");
185 byte[] qualifier = Bytes.toBytes("q");
186 byte[] value = Bytes.toBytes("testV");
187 byte[][] familys = { family1, family2 };
188 TEST_UTIL.createTable(tableName, familys);
189 Table htable = new HTable(TEST_UTIL.getConfiguration(), tableName);
190 Put put = new Put(Bytes.toBytes("r1"));
191 put.add(family1, qualifier, value);
192 htable.put(put);
193 ResultScanner resultScanner = htable.getScanner(new Scan());
194 int count = 0;
195 while (resultScanner.next() != null) {
196 count++;
197 }
198 resultScanner.close();
199 assertEquals(1, count);
200
201 MiniHBaseCluster hbaseCluster = TEST_UTIL.getMiniHBaseCluster();
202 List<HRegion> regions = hbaseCluster.getRegions(tableName);
203 assertEquals(1, regions.size());
204
205
206 Region destRegion = regions.get(0);
207 int originServerNum = hbaseCluster
208 .getServerWith(destRegion.getRegionInfo().getRegionName());
209 assertTrue("Please start more than 1 regionserver", hbaseCluster
210 .getRegionServerThreads().size() > 1);
211 int destServerNum = 0;
212 while (destServerNum == originServerNum) {
213 destServerNum++;
214 }
215 HRegionServer originServer = hbaseCluster.getRegionServer(originServerNum);
216 HRegionServer destServer = hbaseCluster.getRegionServer(destServerNum);
217
218 moveRegionAndWait(destRegion, destServer);
219
220
221 Delete del = new Delete(Bytes.toBytes("r1"));
222 htable.delete(del);
223 resultScanner = htable.getScanner(new Scan());
224 count = 0;
225 while (resultScanner.next() != null) {
226 count++;
227 }
228 resultScanner.close();
229 assertEquals(0, count);
230
231
232 Region region = destServer.getOnlineRegion(destRegion.getRegionInfo().getRegionName());
233 region.flush(true);
234
235 for (Store store : region.getStores()) {
236 store.triggerMajorCompaction();
237 }
238 region.compact(true);
239
240
241 moveRegionAndWait(destRegion, originServer);
242
243 originServer.abort("testing");
244
245
246 Result result = htable.get(new Get(Bytes.toBytes("r1")));
247 if (result != null) {
248 assertTrue("Row is deleted, but we get" + result.toString(),
249 (result == null) || result.isEmpty());
250 }
251 resultScanner.close();
252 }
253
254 private void moveRegionAndWait(Region destRegion, HRegionServer destServer)
255 throws InterruptedException, MasterNotRunningException,
256 ZooKeeperConnectionException, IOException {
257 HMaster master = TEST_UTIL.getMiniHBaseCluster().getMaster();
258 TEST_UTIL.getHBaseAdmin().move(
259 destRegion.getRegionInfo().getEncodedNameAsBytes(),
260 Bytes.toBytes(destServer.getServerName().getServerName()));
261 while (true) {
262 ServerName serverName = master.getAssignmentManager()
263 .getRegionStates().getRegionServerOfRegion(destRegion.getRegionInfo());
264 if (serverName != null && serverName.equals(destServer.getServerName())) {
265 TEST_UTIL.assertRegionOnServer(
266 destRegion.getRegionInfo(), serverName, 200);
267 break;
268 }
269 Thread.sleep(10);
270 }
271 }
272
273
274
275
276
277
278 @Test
279 public void test2727() throws Exception {
280
281
282 final TableName tableName =
283 TableName.valueOf("test2727");
284 HRegionInfo hri = createBasic3FamilyHRegionInfo(tableName);
285 Path basedir = FSUtils.getTableDir(hbaseRootDir, tableName);
286 deleteDir(basedir);
287
288 HTableDescriptor htd = createBasic3FamilyHTD(tableName);
289 HRegion region2 = HRegion.createHRegion(hri, hbaseRootDir, this.conf, htd);
290 HRegion.closeHRegion(region2);
291 final byte [] rowName = tableName.getName();
292
293 WAL wal1 = createWAL(this.conf);
294
295 final int countPerFamily = 1000;
296 final AtomicLong sequenceId = new AtomicLong(1);
297 for (HColumnDescriptor hcd: htd.getFamilies()) {
298 addWALEdits(tableName, hri, rowName, hcd.getName(), countPerFamily, ee,
299 wal1, htd, sequenceId);
300 }
301 wal1.shutdown();
302 runWALSplit(this.conf);
303
304 WAL wal2 = createWAL(this.conf);
305
306 for (HColumnDescriptor hcd: htd.getFamilies()) {
307 addWALEdits(tableName, hri, rowName, hcd.getName(), countPerFamily,
308 ee, wal2, htd, sequenceId);
309 }
310 wal2.shutdown();
311 runWALSplit(this.conf);
312
313 WAL wal3 = createWAL(this.conf);
314 try {
315 HRegion region = HRegion.openHRegion(this.conf, this.fs, hbaseRootDir, hri, htd, wal3);
316 long seqid = region.getOpenSeqNum();
317
318
319 assertTrue(seqid > sequenceId.get());
320 assertEquals(seqid - 1, sequenceId.get());
321 LOG.debug("region.getOpenSeqNum(): " + region.getOpenSeqNum() + ", wal3.id: "
322 + sequenceId.get());
323
324
325 region.close();
326 } finally {
327 wal3.close();
328 }
329 }
330
331
332
333
334
335
336
337
338
339
340 @Test
341 public void testRegionMadeOfBulkLoadedFilesOnly()
342 throws IOException, SecurityException, IllegalArgumentException,
343 NoSuchFieldException, IllegalAccessException, InterruptedException {
344 final TableName tableName =
345 TableName.valueOf("testRegionMadeOfBulkLoadedFilesOnly");
346 final HRegionInfo hri = createBasic3FamilyHRegionInfo(tableName);
347 final Path basedir = new Path(this.hbaseRootDir, tableName.getNameAsString());
348 deleteDir(basedir);
349 final HTableDescriptor htd = createBasic3FamilyHTD(tableName);
350 HRegion region2 = HRegion.createHRegion(hri, hbaseRootDir, this.conf, htd);
351 HRegion.closeHRegion(region2);
352 WAL wal = createWAL(this.conf);
353 HRegion region = HRegion.openHRegion(hri, htd, wal, this.conf);
354
355 byte [] family = htd.getFamilies().iterator().next().getName();
356 Path f = new Path(basedir, "hfile");
357 HFileTestUtil.createHFile(this.conf, fs, f, family, family, Bytes.toBytes(""),
358 Bytes.toBytes("z"), 10);
359 List <Pair<byte[],String>> hfs= new ArrayList<Pair<byte[],String>>(1);
360 hfs.add(Pair.newPair(family, f.toString()));
361 region.bulkLoadHFiles(hfs, true, null);
362
363
364 byte [] row = tableName.getName();
365 region.put((new Put(row)).add(family, family, family));
366 wal.sync();
367 final int rowsInsertedCount = 11;
368
369 assertEquals(rowsInsertedCount, getScannedCount(region.getScanner(new Scan())));
370
371
372 final Configuration newConf = HBaseConfiguration.create(this.conf);
373 User user = HBaseTestingUtility.getDifferentUser(newConf,
374 tableName.getNameAsString());
375 user.runAs(new PrivilegedExceptionAction() {
376 @Override
377 public Object run() throws Exception {
378 runWALSplit(newConf);
379 WAL wal2 = createWAL(newConf);
380
381 HRegion region2 = HRegion.openHRegion(newConf, FileSystem.get(newConf),
382 hbaseRootDir, hri, htd, wal2);
383 long seqid2 = region2.getOpenSeqNum();
384 assertTrue(seqid2 > -1);
385 assertEquals(rowsInsertedCount, getScannedCount(region2.getScanner(new Scan())));
386
387
388 region2.close();
389 wal2.close();
390 return null;
391 }
392 });
393 }
394
395
396
397
398
399
400
401
402
403
404
405
406 @Test
407 public void testCompactedBulkLoadedFiles()
408 throws IOException, SecurityException, IllegalArgumentException,
409 NoSuchFieldException, IllegalAccessException, InterruptedException {
410 final TableName tableName =
411 TableName.valueOf("testCompactedBulkLoadedFiles");
412 final HRegionInfo hri = createBasic3FamilyHRegionInfo(tableName);
413 final Path basedir = new Path(this.hbaseRootDir, tableName.getNameAsString());
414 deleteDir(basedir);
415 final HTableDescriptor htd = createBasic3FamilyHTD(tableName);
416 HRegion region2 = HRegion.createHRegion(hri,
417 hbaseRootDir, this.conf, htd);
418 HRegion.closeHRegion(region2);
419 WAL wal = createWAL(this.conf);
420 HRegion region = HRegion.openHRegion(hri, htd, wal, this.conf);
421
422
423 byte [] row = tableName.getName();
424 byte [] family = htd.getFamilies().iterator().next().getName();
425 region.put((new Put(row)).add(family, family, family));
426 wal.sync();
427
428 List <Pair<byte[],String>> hfs= new ArrayList<Pair<byte[],String>>(1);
429 for (int i = 0; i < 3; i++) {
430 Path f = new Path(basedir, "hfile"+i);
431 HFileTestUtil.createHFile(this.conf, fs, f, family, family, Bytes.toBytes(i + "00"),
432 Bytes.toBytes(i + "50"), 10);
433 hfs.add(Pair.newPair(family, f.toString()));
434 }
435 region.bulkLoadHFiles(hfs, true, null);
436 final int rowsInsertedCount = 31;
437 assertEquals(rowsInsertedCount, getScannedCount(region.getScanner(new Scan())));
438
439
440 region.compact(true);
441 assertEquals(rowsInsertedCount, getScannedCount(region.getScanner(new Scan())));
442
443
444 final Configuration newConf = HBaseConfiguration.create(this.conf);
445 User user = HBaseTestingUtility.getDifferentUser(newConf,
446 tableName.getNameAsString());
447 user.runAs(new PrivilegedExceptionAction() {
448 @Override
449 public Object run() throws Exception {
450 runWALSplit(newConf);
451 WAL wal2 = createWAL(newConf);
452
453 HRegion region2 = HRegion.openHRegion(newConf, FileSystem.get(newConf),
454 hbaseRootDir, hri, htd, wal2);
455 long seqid2 = region2.getOpenSeqNum();
456 assertTrue(seqid2 > -1);
457 assertEquals(rowsInsertedCount, getScannedCount(region2.getScanner(new Scan())));
458
459
460 region2.close();
461 wal2.close();
462 return null;
463 }
464 });
465 }
466
467
468
469
470
471
472
473
474
475
476
477 @Test
478 public void testReplayEditsWrittenViaHRegion()
479 throws IOException, SecurityException, IllegalArgumentException,
480 NoSuchFieldException, IllegalAccessException, InterruptedException {
481 final TableName tableName =
482 TableName.valueOf("testReplayEditsWrittenViaHRegion");
483 final HRegionInfo hri = createBasic3FamilyHRegionInfo(tableName);
484 final Path basedir = FSUtils.getTableDir(this.hbaseRootDir, tableName);
485 deleteDir(basedir);
486 final byte[] rowName = tableName.getName();
487 final int countPerFamily = 10;
488 final HTableDescriptor htd = createBasic3FamilyHTD(tableName);
489 HRegion region3 = HRegion.createHRegion(hri,
490 hbaseRootDir, this.conf, htd);
491 HRegion.closeHRegion(region3);
492
493
494
495 WAL wal = createWAL(this.conf);
496 HRegion region = HRegion.openHRegion(this.conf, this.fs, hbaseRootDir, hri, htd, wal);
497 long seqid = region.getOpenSeqNum();
498 boolean first = true;
499 for (HColumnDescriptor hcd: htd.getFamilies()) {
500 addRegionEdits(rowName, hcd.getName(), countPerFamily, this.ee, region, "x");
501 if (first ) {
502
503 region.flush(true);
504 first = false;
505 }
506 }
507
508 final Get g = new Get(rowName);
509 Result result = region.get(g);
510 assertEquals(countPerFamily * htd.getFamilies().size(),
511 result.size());
512
513
514
515 region.close(true);
516 wal.shutdown();
517 runWALSplit(this.conf);
518 WAL wal2 = createWAL(this.conf);
519 HRegion region2 = HRegion.openHRegion(conf, this.fs, hbaseRootDir, hri, htd, wal2);
520 long seqid2 = region2.getOpenSeqNum();
521 assertTrue(seqid + result.size() < seqid2);
522 final Result result1b = region2.get(g);
523 assertEquals(result.size(), result1b.size());
524
525
526
527
528 for (HColumnDescriptor hcd: htd.getFamilies()) {
529 addRegionEdits(rowName, hcd.getName(), countPerFamily, this.ee, region2, "y");
530 }
531
532 final Result result2 = region2.get(g);
533 assertEquals(2 * result.size(), result2.size());
534 wal2.sync();
535 final Configuration newConf = HBaseConfiguration.create(this.conf);
536 User user = HBaseTestingUtility.getDifferentUser(newConf,
537 tableName.getNameAsString());
538 user.runAs(new PrivilegedExceptionAction() {
539 @Override
540 public Object run() throws Exception {
541 runWALSplit(newConf);
542 FileSystem newFS = FileSystem.get(newConf);
543
544 WAL wal3 = createWAL(newConf);
545 final AtomicInteger countOfRestoredEdits = new AtomicInteger(0);
546 HRegion region3 = new HRegion(basedir, wal3, newFS, newConf, hri, htd, null) {
547 @Override
548 protected boolean restoreEdit(Store s, Cell cell) {
549 boolean b = super.restoreEdit(s, cell);
550 countOfRestoredEdits.incrementAndGet();
551 return b;
552 }
553 };
554 long seqid3 = region3.initialize();
555 Result result3 = region3.get(g);
556
557 assertEquals(result2.size(), result3.size());
558 assertEquals(htd.getFamilies().size() * countPerFamily,
559 countOfRestoredEdits.get());
560
561
562 region3.close();
563 wal3.close();
564 return null;
565 }
566 });
567 }
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587 @Test
588 public void testReplayEditsAfterPartialFlush()
589 throws IOException, SecurityException, IllegalArgumentException,
590 NoSuchFieldException, IllegalAccessException, InterruptedException {
591 final TableName tableName =
592 TableName.valueOf("testReplayEditsWrittenViaHRegion");
593 final HRegionInfo hri = createBasic3FamilyHRegionInfo(tableName);
594 final Path basedir = FSUtils.getTableDir(this.hbaseRootDir, tableName);
595 deleteDir(basedir);
596 final byte[] rowName = tableName.getName();
597 final int countPerFamily = 10;
598 final HTableDescriptor htd = createBasic3FamilyHTD(tableName);
599 HRegion region3 = HRegion.createHRegion(hri,
600 hbaseRootDir, this.conf, htd);
601 HRegion.closeHRegion(region3);
602
603
604
605 WAL wal = createWAL(this.conf);
606 HRegion region = HRegion.openHRegion(this.conf, this.fs, hbaseRootDir, hri, htd, wal);
607 long seqid = region.getOpenSeqNum();
608 for (HColumnDescriptor hcd: htd.getFamilies()) {
609 addRegionEdits(rowName, hcd.getName(), countPerFamily, this.ee, region, "x");
610 }
611
612
613 final Get g = new Get(rowName);
614 Result result = region.get(g);
615 assertEquals(countPerFamily * htd.getFamilies().size(),
616 result.size());
617
618
619 region.flush(true);
620 region.close(true);
621 wal.shutdown();
622
623
624
625
626
627 int cf_count = 0;
628 for (HColumnDescriptor hcd: htd.getFamilies()) {
629 cf_count++;
630 if (cf_count == 2) {
631 region.getRegionFileSystem().deleteFamily(hcd.getNameAsString());
632 }
633 }
634
635
636
637 runWALSplit(this.conf);
638 WAL wal2 = createWAL(this.conf);
639 HRegion region2 = HRegion.openHRegion(this.conf, this.fs, hbaseRootDir, hri, htd, wal2);
640 long seqid2 = region2.getOpenSeqNum();
641 assertTrue(seqid + result.size() < seqid2);
642
643 final Result result1b = region2.get(g);
644 assertEquals(result.size(), result1b.size());
645 }
646
647
648
649
650 public static class CustomStoreFlusher extends DefaultStoreFlusher {
651
652 static final AtomicBoolean throwExceptionWhenFlushing = new AtomicBoolean(false);
653
654 public CustomStoreFlusher(Configuration conf, Store store) {
655 super(conf, store);
656 }
657 @Override
658 public List<Path> flushSnapshot(MemStoreSnapshot snapshot, long cacheFlushId,
659 MonitoredTask status) throws IOException {
660 if (throwExceptionWhenFlushing.get()) {
661 throw new IOException("Simulated exception by tests");
662 }
663 return super.flushSnapshot(snapshot, cacheFlushId, status);
664 }
665
666 };
667
668
669
670
671
672
673
674 @Test
675 public void testReplayEditsAfterAbortingFlush() throws IOException {
676 final TableName tableName =
677 TableName.valueOf("testReplayEditsAfterAbortingFlush");
678 final HRegionInfo hri = createBasic3FamilyHRegionInfo(tableName);
679 final Path basedir = FSUtils.getTableDir(this.hbaseRootDir, tableName);
680 deleteDir(basedir);
681 final HTableDescriptor htd = createBasic3FamilyHTD(tableName);
682 HRegion region3 = HRegion.createHRegion(hri, hbaseRootDir, this.conf, htd);
683 region3.close();
684 region3.getWAL().close();
685
686
687
688 WAL wal = createWAL(this.conf);
689 RegionServerServices rsServices = Mockito.mock(RegionServerServices.class);
690 Mockito.doReturn(false).when(rsServices).isAborted();
691 when(rsServices.getServerName()).thenReturn(ServerName.valueOf("foo", 10, 10));
692 Configuration customConf = new Configuration(this.conf);
693 customConf.set(DefaultStoreEngine.DEFAULT_STORE_FLUSHER_CLASS_KEY,
694 CustomStoreFlusher.class.getName());
695 HRegion region =
696 HRegion.openHRegion(this.hbaseRootDir, hri, htd, wal, customConf, rsServices, null);
697 int writtenRowCount = 10;
698 List<HColumnDescriptor> families = new ArrayList<HColumnDescriptor>(
699 htd.getFamilies());
700 for (int i = 0; i < writtenRowCount; i++) {
701 Put put = new Put(Bytes.toBytes(tableName + Integer.toString(i)));
702 put.add(families.get(i % families.size()).getName(), Bytes.toBytes("q"),
703 Bytes.toBytes("val"));
704 region.put(put);
705 }
706
707
708 RegionScanner scanner = region.getScanner(new Scan());
709 assertEquals(writtenRowCount, getScannedCount(scanner));
710
711
712 CustomStoreFlusher.throwExceptionWhenFlushing.set(true);
713 try {
714 region.flush(true);
715 fail("Injected exception hasn't been thrown");
716 } catch (Throwable t) {
717 LOG.info("Expected simulated exception when flushing region,"
718 + t.getMessage());
719
720 Mockito.doReturn(true).when(rsServices).isAborted();
721 region.setClosing(false);
722
723 }
724
725 int moreRow = 10;
726 for (int i = writtenRowCount; i < writtenRowCount + moreRow; i++) {
727 Put put = new Put(Bytes.toBytes(tableName + Integer.toString(i)));
728 put.add(families.get(i % families.size()).getName(), Bytes.toBytes("q"),
729 Bytes.toBytes("val"));
730 region.put(put);
731 }
732 writtenRowCount += moreRow;
733
734 CustomStoreFlusher.throwExceptionWhenFlushing.set(false);
735 try {
736 region.flush(true);
737 } catch (IOException t) {
738 LOG.info("Expected exception when flushing region because server is stopped,"
739 + t.getMessage());
740 }
741
742 region.close(true);
743 wal.shutdown();
744
745
746 runWALSplit(this.conf);
747 WAL wal2 = createWAL(this.conf);
748 Mockito.doReturn(false).when(rsServices).isAborted();
749 HRegion region2 =
750 HRegion.openHRegion(this.hbaseRootDir, hri, htd, wal2, this.conf, rsServices, null);
751 scanner = region2.getScanner(new Scan());
752 assertEquals(writtenRowCount, getScannedCount(scanner));
753 }
754
755 private int getScannedCount(RegionScanner scanner) throws IOException {
756 int scannedCount = 0;
757 List<Cell> results = new ArrayList<Cell>();
758 while (true) {
759 boolean existMore = scanner.next(results);
760 if (!results.isEmpty())
761 scannedCount++;
762 if (!existMore)
763 break;
764 results.clear();
765 }
766 return scannedCount;
767 }
768
769
770
771
772
773
774 @Test
775 public void testReplayEditsWrittenIntoWAL() throws Exception {
776 final TableName tableName =
777 TableName.valueOf("testReplayEditsWrittenIntoWAL");
778 final HRegionInfo hri = createBasic3FamilyHRegionInfo(tableName);
779 final Path basedir = FSUtils.getTableDir(hbaseRootDir, tableName);
780 deleteDir(basedir);
781
782 final HTableDescriptor htd = createBasic3FamilyHTD(tableName);
783 HRegion region2 = HRegion.createHRegion(hri,
784 hbaseRootDir, this.conf, htd);
785 HRegion.closeHRegion(region2);
786 final WAL wal = createWAL(this.conf);
787 final byte[] rowName = tableName.getName();
788 final byte[] regionName = hri.getEncodedNameAsBytes();
789 final AtomicLong sequenceId = new AtomicLong(1);
790
791
792 final int countPerFamily = 1000;
793 Set<byte[]> familyNames = new HashSet<byte[]>();
794 for (HColumnDescriptor hcd: htd.getFamilies()) {
795 addWALEdits(tableName, hri, rowName, hcd.getName(), countPerFamily,
796 ee, wal, htd, sequenceId);
797 familyNames.add(hcd.getName());
798 }
799
800
801 wal.startCacheFlush(regionName, familyNames);
802 wal.completeCacheFlush(regionName);
803
804
805 WALEdit edit = new WALEdit();
806 long now = ee.currentTime();
807 edit.add(new KeyValue(rowName, Bytes.toBytes("another family"), rowName,
808 now, rowName));
809 wal.append(htd, hri, new WALKey(hri.getEncodedNameAsBytes(), tableName, now), edit, sequenceId,
810 true, null);
811
812
813 edit = new WALEdit();
814 now = ee.currentTime();
815 edit.add(new KeyValue(rowName, Bytes.toBytes("c"), null, now,
816 KeyValue.Type.DeleteFamily));
817 wal.append(htd, hri, new WALKey(hri.getEncodedNameAsBytes(), tableName, now), edit, sequenceId,
818 true, null);
819
820
821 wal.sync();
822
823
824 final Configuration newConf = HBaseConfiguration.create(this.conf);
825 User user = HBaseTestingUtility.getDifferentUser(newConf,
826 ".replay.wal.secondtime");
827 user.runAs(new PrivilegedExceptionAction() {
828 @Override
829 public Object run() throws Exception {
830 runWALSplit(newConf);
831 FileSystem newFS = FileSystem.get(newConf);
832
833 newConf.setInt(HConstants.HREGION_MEMSTORE_FLUSH_SIZE, 1024 * 100);
834
835 WAL newWal = createWAL(newConf);
836 final AtomicInteger flushcount = new AtomicInteger(0);
837 try {
838 final HRegion region =
839 new HRegion(basedir, newWal, newFS, newConf, hri, htd, null) {
840 @Override
841 protected FlushResult internalFlushcache(final WAL wal, final long myseqid,
842 final Collection<Store> storesToFlush, MonitoredTask status,
843 boolean writeFlushWalMarker)
844 throws IOException {
845 LOG.info("InternalFlushCache Invoked");
846 FlushResult fs = super.internalFlushcache(wal, myseqid, storesToFlush,
847 Mockito.mock(MonitoredTask.class), writeFlushWalMarker);
848 flushcount.incrementAndGet();
849 return fs;
850 };
851 };
852 long seqid = region.initialize();
853
854 assertTrue("Flushcount=" + flushcount.get(), flushcount.get() > 0);
855 assertTrue(seqid - 1 == sequenceId.get());
856
857 Get get = new Get(rowName);
858 Result result = region.get(get);
859
860 assertEquals(countPerFamily * (htd.getFamilies().size() - 1),
861 result.size());
862 region.close();
863 } finally {
864 newWal.close();
865 }
866 return null;
867 }
868 });
869 }
870
871 @Test
872
873 public void testSequentialEditLogSeqNum() throws IOException {
874 final TableName tableName = TableName.valueOf(currentTest.getMethodName());
875 final HRegionInfo hri = createBasic3FamilyHRegionInfo(tableName);
876 final Path basedir =
877 FSUtils.getTableDir(this.hbaseRootDir, tableName);
878 deleteDir(basedir);
879 final byte[] rowName = tableName.getName();
880 final int countPerFamily = 10;
881 final HTableDescriptor htd = createBasic1FamilyHTD(tableName);
882
883
884 MockWAL wal = createMockWAL();
885
886 HRegion region = HRegion.openHRegion(this.conf, this.fs, hbaseRootDir, hri, htd, wal);
887 for (HColumnDescriptor hcd : htd.getFamilies()) {
888 addRegionEdits(rowName, hcd.getName(), countPerFamily, this.ee, region, "x");
889 }
890
891
892
893 region.flush(true);
894 for (HColumnDescriptor hcd : htd.getFamilies()) {
895 addRegionEdits(rowName, hcd.getName(), 5, this.ee, region, "x");
896 }
897 long lastestSeqNumber = region.getSequenceId().get();
898
899 wal.doCompleteCacheFlush = true;
900
901
902 wal.completeCacheFlush(hri.getEncodedNameAsBytes());
903 wal.shutdown();
904 FileStatus[] listStatus = wal.getFiles();
905 assertNotNull(listStatus);
906 assertTrue(listStatus.length > 0);
907 WALSplitter.splitLogFile(hbaseRootDir, listStatus[0],
908 this.fs, this.conf, null, null, null, mode, wals);
909 FileStatus[] listStatus1 = this.fs.listStatus(
910 new Path(FSUtils.getTableDir(hbaseRootDir, tableName), new Path(hri.getEncodedName(),
911 "recovered.edits")), new PathFilter() {
912 @Override
913 public boolean accept(Path p) {
914 if (WALSplitter.isSequenceIdFile(p)) {
915 return false;
916 }
917 return true;
918 }
919 });
920 int editCount = 0;
921 for (FileStatus fileStatus : listStatus1) {
922 editCount = Integer.parseInt(fileStatus.getPath().getName());
923 }
924
925 assertEquals(
926 "The sequence number of the recoverd.edits and the current edit seq should be same",
927 lastestSeqNumber, editCount);
928 }
929
930 static class MockWAL extends FSHLog {
931 boolean doCompleteCacheFlush = false;
932
933 public MockWAL(FileSystem fs, Path rootDir, String logName, Configuration conf)
934 throws IOException {
935 super(fs, rootDir, logName, HConstants.HREGION_OLDLOGDIR_NAME, conf, null, true, null, null);
936 }
937
938 @Override
939 public void completeCacheFlush(byte[] encodedRegionName) {
940 if (!doCompleteCacheFlush) {
941 return;
942 }
943 super.completeCacheFlush(encodedRegionName);
944 }
945 }
946
947 private HTableDescriptor createBasic1FamilyHTD(final TableName tableName) {
948 HTableDescriptor htd = new HTableDescriptor(tableName);
949 HColumnDescriptor a = new HColumnDescriptor(Bytes.toBytes("a"));
950 htd.addFamily(a);
951 return htd;
952 }
953
954 private MockWAL createMockWAL() throws IOException {
955 MockWAL wal = new MockWAL(fs, hbaseRootDir, logName, conf);
956
957
958 HBaseTestingUtility.setMaxRecoveryErrorCount(wal.getOutputStream(), 1);
959 return wal;
960 }
961
962
963
964 class TestFlusher implements FlushRequester {
965 private HRegion r;
966
967 @Override
968 public void requestFlush(Region region, boolean force) {
969 try {
970 r.flush(force);
971 } catch (IOException e) {
972 throw new RuntimeException("Exception flushing", e);
973 }
974 }
975
976 @Override
977 public void requestDelayedFlush(Region region, long when, boolean forceFlushAllStores) {
978
979
980 }
981
982 @Override
983 public void registerFlushRequestListener(FlushRequestListener listener) {
984
985 }
986
987 @Override
988 public boolean unregisterFlushRequestListener(FlushRequestListener listener) {
989 return false;
990 }
991
992 @Override
993 public void setGlobalMemstoreLimit(long globalMemStoreSize) {
994
995 }
996 }
997
998 private void addWALEdits(final TableName tableName, final HRegionInfo hri, final byte[] rowName,
999 final byte[] family, final int count, EnvironmentEdge ee, final WAL wal,
1000 final HTableDescriptor htd, final AtomicLong sequenceId)
1001 throws IOException {
1002 String familyStr = Bytes.toString(family);
1003 for (int j = 0; j < count; j++) {
1004 byte[] qualifierBytes = Bytes.toBytes(Integer.toString(j));
1005 byte[] columnBytes = Bytes.toBytes(familyStr + ":" + Integer.toString(j));
1006 WALEdit edit = new WALEdit();
1007 edit.add(new KeyValue(rowName, family, qualifierBytes,
1008 ee.currentTime(), columnBytes));
1009 wal.append(htd, hri, new WALKey(hri.getEncodedNameAsBytes(), tableName, ee.currentTime()),
1010 edit, sequenceId, true, null);
1011 }
1012 wal.sync();
1013 }
1014
1015 static List<Put> addRegionEdits (final byte [] rowName, final byte [] family,
1016 final int count, EnvironmentEdge ee, final Region r,
1017 final String qualifierPrefix)
1018 throws IOException {
1019 List<Put> puts = new ArrayList<Put>();
1020 for (int j = 0; j < count; j++) {
1021 byte[] qualifier = Bytes.toBytes(qualifierPrefix + Integer.toString(j));
1022 Put p = new Put(rowName);
1023 p.add(family, qualifier, ee.currentTime(), rowName);
1024 r.put(p);
1025 puts.add(p);
1026 }
1027 return puts;
1028 }
1029
1030
1031
1032
1033
1034
1035 private HRegionInfo createBasic3FamilyHRegionInfo(final TableName tableName) {
1036 return new HRegionInfo(tableName, null, null, false);
1037 }
1038
1039
1040
1041
1042
1043
1044
1045 private Path runWALSplit(final Configuration c) throws IOException {
1046 List<Path> splits = WALSplitter.split(
1047 hbaseRootDir, logDir, oldLogDir, FileSystem.get(c), c, wals);
1048
1049 assertEquals("splits=" + splits, 1, splits.size());
1050
1051 assertTrue(fs.exists(splits.get(0)));
1052 LOG.info("Split file=" + splits.get(0));
1053 return splits.get(0);
1054 }
1055
1056
1057
1058
1059
1060
1061 private WAL createWAL(final Configuration c) throws IOException {
1062 FSHLog wal = new FSHLog(FileSystem.get(c), hbaseRootDir, logName, c);
1063
1064
1065 HBaseTestingUtility.setMaxRecoveryErrorCount(wal.getOutputStream(), 1);
1066 return wal;
1067 }
1068
1069 private HTableDescriptor createBasic3FamilyHTD(final TableName tableName) {
1070 HTableDescriptor htd = new HTableDescriptor(tableName);
1071 HColumnDescriptor a = new HColumnDescriptor(Bytes.toBytes("a"));
1072 htd.addFamily(a);
1073 HColumnDescriptor b = new HColumnDescriptor(Bytes.toBytes("b"));
1074 htd.addFamily(b);
1075 HColumnDescriptor c = new HColumnDescriptor(Bytes.toBytes("c"));
1076 htd.addFamily(c);
1077 return htd;
1078 }
1079 }