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