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