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;
20
21 import static org.apache.hadoop.hbase.HBaseTestingUtility.START_KEY;
22 import static org.apache.hadoop.hbase.HBaseTestingUtility.START_KEY_BYTES;
23 import static org.apache.hadoop.hbase.HBaseTestingUtility.fam1;
24 import static org.junit.Assert.assertEquals;
25 import static org.junit.Assert.assertTrue;
26 import static org.junit.Assert.fail;
27 import static org.mockito.Matchers.any;
28 import static org.mockito.Mockito.doAnswer;
29 import static org.mockito.Mockito.mock;
30 import static org.mockito.Mockito.spy;
31 import static org.mockito.Mockito.when;
32
33 import java.io.IOException;
34 import java.util.ArrayList;
35 import java.util.Collection;
36 import java.util.Collections;
37 import java.util.List;
38 import java.util.concurrent.CountDownLatch;
39 import java.util.concurrent.TimeUnit;
40
41 import org.apache.commons.logging.Log;
42 import org.apache.commons.logging.LogFactory;
43 import org.apache.hadoop.conf.Configuration;
44 import org.apache.hadoop.fs.FSDataOutputStream;
45 import org.apache.hadoop.fs.FileStatus;
46 import org.apache.hadoop.fs.FileSystem;
47 import org.apache.hadoop.fs.Path;
48 import org.apache.hadoop.hbase.ChoreService;
49 import org.apache.hadoop.hbase.HBaseConfiguration;
50 import org.apache.hadoop.hbase.HBaseTestCase;
51 import org.apache.hadoop.hbase.HBaseTestCase.HRegionIncommon;
52 import org.apache.hadoop.hbase.HBaseTestingUtility;
53 import org.apache.hadoop.hbase.HColumnDescriptor;
54 import org.apache.hadoop.hbase.HConstants;
55 import org.apache.hadoop.hbase.HTableDescriptor;
56 import org.apache.hadoop.hbase.client.Delete;
57 import org.apache.hadoop.hbase.client.Durability;
58 import org.apache.hadoop.hbase.client.Put;
59 import org.apache.hadoop.hbase.io.hfile.HFileScanner;
60 import org.apache.hadoop.hbase.regionserver.compactions.CompactionContext;
61 import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequest;
62 import org.apache.hadoop.hbase.regionserver.compactions.DefaultCompactor;
63 import org.apache.hadoop.hbase.regionserver.throttle.CompactionThroughputControllerFactory;
64 import org.apache.hadoop.hbase.regionserver.throttle.NoLimitThroughputController;
65 import org.apache.hadoop.hbase.regionserver.throttle.ThroughputController;
66 import org.apache.hadoop.hbase.security.User;
67 import org.apache.hadoop.hbase.testclassification.MediumTests;
68 import org.apache.hadoop.hbase.util.Bytes;
69 import org.apache.hadoop.hbase.util.Pair;
70 import org.apache.hadoop.hbase.util.Threads;
71 import org.apache.hadoop.hbase.wal.WAL;
72 import org.junit.After;
73 import org.junit.Assume;
74 import org.junit.Before;
75 import org.junit.Rule;
76 import org.junit.Test;
77 import org.junit.experimental.categories.Category;
78 import org.junit.rules.TestName;
79 import org.mockito.Mockito;
80 import org.mockito.invocation.InvocationOnMock;
81 import org.mockito.stubbing.Answer;
82
83
84
85
86
87 @Category(MediumTests.class)
88 public class TestCompaction {
89 @Rule public TestName name = new TestName();
90 private static final Log LOG = LogFactory.getLog(TestCompaction.class.getName());
91 private static final HBaseTestingUtility UTIL = HBaseTestingUtility.createLocalHTU();
92 protected Configuration conf = UTIL.getConfiguration();
93
94 private HRegion r = null;
95 private HTableDescriptor htd = null;
96 private static final byte [] COLUMN_FAMILY = fam1;
97 private final byte [] STARTROW = Bytes.toBytes(START_KEY);
98 private static final byte [] COLUMN_FAMILY_TEXT = COLUMN_FAMILY;
99 private int compactionThreshold;
100 private byte[] secondRowBytes, thirdRowBytes;
101 private static final long MAX_FILES_TO_COMPACT = 10;
102 private final byte[] FAMILY = Bytes.toBytes("cf");
103
104
105 public TestCompaction() {
106 super();
107
108
109 conf.setInt(HConstants.HREGION_MEMSTORE_FLUSH_SIZE, 1024 * 1024);
110 conf.setInt(HConstants.HREGION_MEMSTORE_BLOCK_MULTIPLIER, 100);
111 conf.set(CompactionThroughputControllerFactory.HBASE_THROUGHPUT_CONTROLLER_KEY,
112 NoLimitThroughputController.class.getName());
113 compactionThreshold = conf.getInt("hbase.hstore.compactionThreshold", 3);
114
115 secondRowBytes = START_KEY_BYTES.clone();
116
117 secondRowBytes[START_KEY_BYTES.length - 1]++;
118 thirdRowBytes = START_KEY_BYTES.clone();
119 thirdRowBytes[START_KEY_BYTES.length - 1] += 2;
120 }
121
122 @Before
123 public void setUp() throws Exception {
124 this.htd = UTIL.createTableDescriptor(name.getMethodName());
125 if (name.getMethodName().equals("testCompactionSeqId")) {
126 UTIL.getConfiguration().set("hbase.hstore.compaction.kv.max", "10");
127 UTIL.getConfiguration().set(
128 DefaultStoreEngine.DEFAULT_COMPACTOR_CLASS_KEY,
129 DummyCompactor.class.getName());
130 HColumnDescriptor hcd = new HColumnDescriptor(FAMILY);
131 hcd.setMaxVersions(65536);
132 this.htd.addFamily(hcd);
133 }
134 this.r = UTIL.createLocalHRegion(htd, null, null);
135 }
136
137 @After
138 public void tearDown() throws Exception {
139 WAL wal = r.getWAL();
140 this.r.close();
141 wal.close();
142 }
143
144
145
146
147
148
149 @Test
150 public void testInterruptCompaction() throws Exception {
151 assertEquals(0, count());
152
153
154 int origWI = HStore.closeCheckInterval;
155 HStore.closeCheckInterval = 10*1000;
156
157 try {
158
159 int jmax = (int) Math.ceil(15.0/compactionThreshold);
160 byte [] pad = new byte[1000];
161 for (int i = 0; i < compactionThreshold; i++) {
162 HRegionIncommon loader = new HRegionIncommon(r);
163 Put p = new Put(Bytes.add(STARTROW, Bytes.toBytes(i)));
164 p.setDurability(Durability.SKIP_WAL);
165 for (int j = 0; j < jmax; j++) {
166 p.add(COLUMN_FAMILY, Bytes.toBytes(j), pad);
167 }
168 HBaseTestCase.addContent(loader, Bytes.toString(COLUMN_FAMILY));
169 loader.put(p);
170 loader.flushcache();
171 }
172
173 HRegion spyR = spy(r);
174 doAnswer(new Answer() {
175 @Override
176 public Object answer(InvocationOnMock invocation) throws Throwable {
177 r.writestate.writesEnabled = false;
178 return invocation.callRealMethod();
179 }
180 }).when(spyR).doRegionCompactionPrep();
181
182
183 spyR.compactStores();
184
185
186 Store s = r.stores.get(COLUMN_FAMILY);
187 assertEquals(compactionThreshold, s.getStorefilesCount());
188 assertTrue(s.getStorefilesSize() > 15*1000);
189
190 FileStatus[] ls = r.getFilesystem().listStatus(r.getRegionFileSystem().getTempDir());
191 assertEquals(0, ls.length);
192
193 } finally {
194
195 r.writestate.writesEnabled = true;
196 HStore.closeCheckInterval = origWI;
197
198
199 for (int i = 0; i < compactionThreshold; i++) {
200 Delete delete = new Delete(Bytes.add(STARTROW, Bytes.toBytes(i)));
201 byte [][] famAndQf = {COLUMN_FAMILY, null};
202 delete.deleteFamily(famAndQf[0]);
203 r.delete(delete);
204 }
205 r.flush(true);
206
207
208
209 final int ttl = 1000;
210 for (Store hstore: this.r.stores.values()) {
211 HStore store = (HStore)hstore;
212 ScanInfo old = store.getScanInfo();
213 ScanInfo si = new ScanInfo(old.getConfiguration(), old.getFamily(),
214 old.getMinVersions(), old.getMaxVersions(), ttl,
215 old.getKeepDeletedCells(), 0, old.getComparator());
216 store.setScanInfo(si);
217 }
218 Thread.sleep(ttl);
219
220 r.compact(true);
221 assertEquals(0, count());
222 }
223 }
224
225 private int count() throws IOException {
226 int count = 0;
227 for (StoreFile f: this.r.stores.
228 get(COLUMN_FAMILY_TEXT).getStorefiles()) {
229 HFileScanner scanner = f.getReader().getScanner(false, false);
230 if (!scanner.seekTo()) {
231 continue;
232 }
233 do {
234 count++;
235 } while(scanner.next());
236 }
237 return count;
238 }
239
240 private void createStoreFile(final HRegion region) throws IOException {
241 createStoreFile(region, Bytes.toString(COLUMN_FAMILY));
242 }
243
244 private void createStoreFile(final HRegion region, String family) throws IOException {
245 HRegionIncommon loader = new HRegionIncommon(region);
246 HBaseTestCase.addContent(loader, family);
247 loader.flushcache();
248 }
249
250 @Test
251 public void testCompactionWithCorruptResult() throws Exception {
252 int nfiles = 10;
253 for (int i = 0; i < nfiles; i++) {
254 createStoreFile(r);
255 }
256 HStore store = (HStore) r.getStore(COLUMN_FAMILY);
257
258 Collection<StoreFile> storeFiles = store.getStorefiles();
259 DefaultCompactor tool = (DefaultCompactor)store.storeEngine.getCompactor();
260 tool.compactForTesting(storeFiles, false);
261
262
263 FileSystem fs = store.getFileSystem();
264
265 Path dstPath = store.getRegionFileSystem().createTempName();
266 FSDataOutputStream stream = fs.create(dstPath, null, true, 512, (short)3, (long)1024, null);
267 stream.writeChars("CORRUPT FILE!!!!");
268 stream.close();
269 Path origPath = store.getRegionFileSystem().commitStoreFile(
270 Bytes.toString(COLUMN_FAMILY), dstPath);
271
272 try {
273 ((HStore)store).moveFileIntoPlace(origPath);
274 } catch (Exception e) {
275
276
277 assert (fs.exists(origPath));
278 assert (!fs.exists(dstPath));
279 System.out.println("testCompactionWithCorruptResult Passed");
280 return;
281 }
282 fail("testCompactionWithCorruptResult failed since no exception was" +
283 "thrown while completing a corrupt file");
284 }
285
286
287
288
289
290 @Test
291 public void testTrackingCompactionRequest() throws Exception {
292
293 HRegionServer mockServer = Mockito.mock(HRegionServer.class);
294 Mockito.when(mockServer.getConfiguration()).thenReturn(r.getBaseConf());
295 CompactSplitThread thread = new CompactSplitThread(mockServer);
296 Mockito.when(mockServer.getCompactSplitThread()).thenReturn(thread);
297
298
299 Store store = r.getStore(COLUMN_FAMILY);
300 createStoreFile(r);
301 for (int i = 0; i < MAX_FILES_TO_COMPACT + 1; i++) {
302 createStoreFile(r);
303 }
304
305 CountDownLatch latch = new CountDownLatch(1);
306 TrackableCompactionRequest request = new TrackableCompactionRequest(latch);
307 thread.requestCompaction(r, store, "test custom comapction", Store.PRIORITY_USER, request,null);
308
309 latch.await();
310
311 thread.interruptIfNecessary();
312 }
313
314 @Test
315 public void testCompactionFailure() throws Exception {
316
317 HRegionServer mockServer = Mockito.mock(HRegionServer.class);
318 Mockito.when(mockServer.getConfiguration()).thenReturn(r.getBaseConf());
319 CompactSplitThread thread = new CompactSplitThread(mockServer);
320 Mockito.when(mockServer.getCompactSplitThread()).thenReturn(thread);
321
322
323 Store store = r.getStore(COLUMN_FAMILY);
324 createStoreFile(r);
325 for (int i = 0; i < HStore.DEFAULT_BLOCKING_STOREFILE_COUNT - 1; i++) {
326 createStoreFile(r);
327 }
328
329 HRegion mockRegion = Mockito.spy(r);
330 Mockito.when(mockRegion.checkSplit()).thenThrow(new IndexOutOfBoundsException());
331
332 MetricsRegionWrapper metricsWrapper = new MetricsRegionWrapperImpl(r);
333
334 long preCompletedCount = metricsWrapper.getNumCompactionsCompleted();
335 long preFailedCount = metricsWrapper.getNumCompactionsFailed();
336
337 CountDownLatch latch = new CountDownLatch(1);
338 TrackableCompactionRequest request = new TrackableCompactionRequest(latch);
339 thread.requestCompaction(mockRegion, store, "test custom comapction", Store.PRIORITY_USER,
340 request, null);
341
342 latch.await(120, TimeUnit.SECONDS);
343
344
345 long postCompletedCount = metricsWrapper.getNumCompactionsCompleted();
346 long postFailedCount = metricsWrapper.getNumCompactionsFailed();
347
348 assertTrue("Completed count should have increased (pre=" + preCompletedCount +
349 ", post="+postCompletedCount+")",
350 postCompletedCount > preCompletedCount);
351 assertTrue("Failed count should have increased (pre=" + preFailedCount +
352 ", post=" + postFailedCount + ")",
353 postFailedCount > preFailedCount);
354 }
355
356
357
358
359
360
361 @Test
362 public void testMultipleCustomCompactionRequests() throws Exception {
363
364 HRegionServer mockServer = Mockito.mock(HRegionServer.class);
365 Mockito.when(mockServer.getConfiguration()).thenReturn(r.getBaseConf());
366 CompactSplitThread thread = new CompactSplitThread(mockServer);
367 Mockito.when(mockServer.getCompactSplitThread()).thenReturn(thread);
368
369
370 int numStores = r.getStores().size();
371 List<Pair<CompactionRequest, Store>> requests =
372 new ArrayList<Pair<CompactionRequest, Store>>(numStores);
373 CountDownLatch latch = new CountDownLatch(numStores);
374
375
376 for (Store store : r.getStores()) {
377 createStoreFile(r, store.getColumnFamilyName());
378 createStoreFile(r, store.getColumnFamilyName());
379 createStoreFile(r, store.getColumnFamilyName());
380 requests
381 .add(new Pair<CompactionRequest, Store>(new TrackableCompactionRequest(latch), store));
382 }
383
384 thread.requestCompaction(r, "test mulitple custom comapctions", Store.PRIORITY_USER,
385 Collections.unmodifiableList(requests), null);
386
387
388 latch.await();
389
390 thread.interruptIfNecessary();
391 }
392
393 private class StoreMockMaker extends StatefulStoreMockMaker {
394 public ArrayList<StoreFile> compacting = new ArrayList<StoreFile>();
395 public ArrayList<StoreFile> notCompacting = new ArrayList<StoreFile>();
396 private ArrayList<Integer> results;
397
398 public StoreMockMaker(ArrayList<Integer> results) {
399 this.results = results;
400 }
401
402 public class TestCompactionContext extends CompactionContext {
403 private List<StoreFile> selectedFiles;
404 public TestCompactionContext(List<StoreFile> selectedFiles) {
405 super();
406 this.selectedFiles = selectedFiles;
407 }
408
409 @Override
410 public List<StoreFile> preSelect(List<StoreFile> filesCompacting) {
411 return new ArrayList<StoreFile>();
412 }
413
414 @Override
415 public boolean select(List<StoreFile> filesCompacting, boolean isUserCompaction,
416 boolean mayUseOffPeak, boolean forceMajor) throws IOException {
417 this.request = new CompactionRequest(selectedFiles);
418 this.request.setPriority(getPriority());
419 return true;
420 }
421
422 @Override
423 public List<Path> compact(ThroughputController throughputController, User user)
424 throws IOException {
425 finishCompaction(this.selectedFiles);
426 return new ArrayList<Path>();
427 }
428 }
429
430 @Override
431 public synchronized CompactionContext selectCompaction() {
432 CompactionContext ctx = new TestCompactionContext(new ArrayList<StoreFile>(notCompacting));
433 compacting.addAll(notCompacting);
434 notCompacting.clear();
435 try {
436 ctx.select(null, false, false, false);
437 } catch (IOException ex) {
438 fail("Shouldn't happen");
439 }
440 return ctx;
441 }
442
443 @Override
444 public synchronized void cancelCompaction(Object object) {
445 TestCompactionContext ctx = (TestCompactionContext)object;
446 compacting.removeAll(ctx.selectedFiles);
447 notCompacting.addAll(ctx.selectedFiles);
448 }
449
450 public synchronized void finishCompaction(List<StoreFile> sfs) {
451 if (sfs.isEmpty()) return;
452 synchronized (results) {
453 results.add(sfs.size());
454 }
455 compacting.removeAll(sfs);
456 }
457
458 @Override
459 public int getPriority() {
460 return 7 - compacting.size() - notCompacting.size();
461 }
462 }
463
464 public class BlockingStoreMockMaker extends StatefulStoreMockMaker {
465 BlockingCompactionContext blocked = null;
466
467 public class BlockingCompactionContext extends CompactionContext {
468 public volatile boolean isInCompact = false;
469
470 public void unblock() {
471 synchronized (this) { this.notifyAll(); }
472 }
473
474 @Override
475 public List<Path> compact(ThroughputController throughputController, User user)
476 throws IOException {
477 try {
478 isInCompact = true;
479 synchronized (this) {
480
481 this.wait();
482 }
483 } catch (InterruptedException e) {
484 Assume.assumeNoException(e);
485 }
486 return new ArrayList<Path>();
487 }
488
489 @Override
490 public List<StoreFile> preSelect(List<StoreFile> filesCompacting) {
491 return new ArrayList<StoreFile>();
492 }
493
494 @Override
495 public boolean select(List<StoreFile> f, boolean i, boolean m, boolean e)
496 throws IOException {
497 this.request = new CompactionRequest(new ArrayList<StoreFile>());
498 return true;
499 }
500 }
501
502 @Override
503 public CompactionContext selectCompaction() {
504 this.blocked = new BlockingCompactionContext();
505 try {
506 this.blocked.select(null, false, false, false);
507 } catch (IOException ex) {
508 fail("Shouldn't happen");
509 }
510 return this.blocked;
511 }
512
513 @Override
514 public void cancelCompaction(Object object) {}
515
516 @Override
517 public int getPriority() {
518 return Integer.MIN_VALUE;
519 }
520
521 public BlockingCompactionContext waitForBlocking() {
522 while (this.blocked == null || !this.blocked.isInCompact) {
523 Threads.sleepWithoutInterrupt(50);
524 }
525 BlockingCompactionContext ctx = this.blocked;
526 this.blocked = null;
527 return ctx;
528 }
529
530 @Override
531 public Store createStoreMock(String name) throws Exception {
532 return createStoreMock(Integer.MIN_VALUE, name);
533 }
534
535 public Store createStoreMock(int priority, String name) throws Exception {
536
537 Store s = super.createStoreMock(name);
538 when(s.getCompactPriority()).thenReturn(priority);
539 return s;
540 }
541 }
542
543
544 @Test
545 public void testCompactionQueuePriorities() throws Exception {
546
547 final Configuration conf = HBaseConfiguration.create();
548 HRegionServer mockServer = mock(HRegionServer.class);
549 when(mockServer.isStopped()).thenReturn(false);
550 when(mockServer.getConfiguration()).thenReturn(conf);
551 when(mockServer.getChoreService()).thenReturn(new ChoreService("test"));
552 CompactSplitThread cst = new CompactSplitThread(mockServer);
553 when(mockServer.getCompactSplitThread()).thenReturn(cst);
554
555 cst.shutdownLongCompactions();
556
557 HRegion r = mock(HRegion.class);
558 when(
559 r.compact(any(CompactionContext.class), any(Store.class),
560 any(ThroughputController.class), any(User.class))).then(new Answer<Boolean>() {
561 @Override
562 public Boolean answer(InvocationOnMock invocation) throws Throwable {
563 invocation.getArgumentAt(0, CompactionContext.class).compact(
564 invocation.getArgumentAt(2, ThroughputController.class), null);
565 return true;
566 }
567 });
568
569
570 ArrayList<Integer> results = new ArrayList<Integer>();
571 StoreMockMaker sm = new StoreMockMaker(results), sm2 = new StoreMockMaker(results);
572 Store store = sm.createStoreMock("store1"), store2 = sm2.createStoreMock("store2");
573 BlockingStoreMockMaker blocker = new BlockingStoreMockMaker();
574
575
576 cst.requestSystemCompaction(r, blocker.createStoreMock(1, "b-pri1"), "b-pri1");
577 BlockingStoreMockMaker.BlockingCompactionContext currentBlock = blocker.waitForBlocking();
578
579
580 for (int i = 0; i < 4; ++i) {
581 sm.notCompacting.add(createFile());
582 }
583 cst.requestSystemCompaction(r, store, "s1-pri3");
584 for (int i = 0; i < 3; ++i) {
585 sm2.notCompacting.add(createFile());
586 }
587 cst.requestSystemCompaction(r, store2, "s2-pri4");
588
589 for (int i = 0; i < 2; ++i) {
590 sm.notCompacting.add(createFile());
591 }
592 cst.requestSystemCompaction(r, store, "s1-pri1");
593
594 cst.requestSystemCompaction(r, blocker.createStoreMock(2, "b-pri2"), "b-pri2");
595
596
597 currentBlock.unblock();
598 currentBlock = blocker.waitForBlocking();
599
600 assertEquals(1, results.size());
601 assertEquals(6, results.get(0).intValue());
602
603 for (int i = 0; i < 2; ++i) {
604 sm.notCompacting.add(createFile());
605 }
606
607
608 cst.requestSystemCompaction(r, blocker.createStoreMock(7, "b-pri7"), "b-pri7");
609 currentBlock.unblock();
610 currentBlock = blocker.waitForBlocking();
611 assertEquals(3, results.size());
612 assertEquals(3, results.get(1).intValue());
613 assertEquals(2, results.get(2).intValue());
614
615 currentBlock.unblock();
616 cst.interruptIfNecessary();
617 }
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635 @Test
636 public void testCompactionSeqId() throws Exception {
637 final byte[] ROW = Bytes.toBytes("row");
638 final byte[] QUALIFIER = Bytes.toBytes("qualifier");
639
640 long timestamp = 10000;
641
642
643
644
645
646
647
648
649
650
651
652 for (int i = 0; i < 10; i++) {
653 Put put = new Put(ROW);
654 put.addColumn(FAMILY, QUALIFIER, timestamp + i, Bytes.toBytes("v" + i));
655 r.put(put);
656 }
657 r.flush(true);
658
659
660
661
662
663
664
665
666
667
668
669 for (int i = 18; i > 8; i--) {
670 Put put = new Put(ROW);
671 put.addColumn(FAMILY, QUALIFIER, timestamp + i, Bytes.toBytes("v" + i));
672 r.put(put);
673 }
674 r.flush(true);
675 r.compact(true);
676 }
677
678 public static class DummyCompactor extends DefaultCompactor {
679 public DummyCompactor(Configuration conf, Store store) {
680 super(conf, store);
681 this.keepSeqIdPeriod = 0;
682 }
683 }
684
685 private static StoreFile createFile() throws Exception {
686 StoreFile sf = mock(StoreFile.class);
687 when(sf.getPath()).thenReturn(new Path("file"));
688 StoreFile.Reader r = mock(StoreFile.Reader.class);
689 when(r.length()).thenReturn(10L);
690 when(sf.getReader()).thenReturn(r);
691 return sf;
692 }
693
694
695
696
697 public static class TrackableCompactionRequest extends CompactionRequest {
698 private CountDownLatch done;
699
700
701
702
703
704 public TrackableCompactionRequest(CountDownLatch finished) {
705 super();
706 this.done = finished;
707 }
708
709 @Override
710 public void afterExecute() {
711 super.afterExecute();
712 this.done.countDown();
713 }
714 }
715 }