1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.mapreduce;
20
21 import static org.junit.Assert.assertEquals;
22 import static org.junit.Assert.assertFalse;
23 import static org.junit.Assert.assertNotNull;
24 import static org.junit.Assert.assertNotSame;
25 import static org.junit.Assert.assertTrue;
26 import static org.junit.Assert.fail;
27
28 import java.io.IOException;
29 import java.io.UnsupportedEncodingException;
30 import java.util.Arrays;
31 import java.util.HashMap;
32 import java.util.Iterator;
33 import java.util.List;
34 import java.util.Map;
35 import java.util.Map.Entry;
36 import java.util.Random;
37 import java.util.Set;
38 import java.util.concurrent.Callable;
39
40 import org.apache.commons.logging.Log;
41 import org.apache.commons.logging.LogFactory;
42 import org.apache.hadoop.conf.Configuration;
43 import org.apache.hadoop.fs.FileStatus;
44 import org.apache.hadoop.fs.FileSystem;
45 import org.apache.hadoop.fs.LocatedFileStatus;
46 import org.apache.hadoop.fs.Path;
47 import org.apache.hadoop.fs.RemoteIterator;
48 import org.apache.hadoop.hbase.CategoryBasedTimeout;
49 import org.apache.hadoop.hbase.Cell;
50 import org.apache.hadoop.hbase.CellUtil;
51 import org.apache.hadoop.hbase.CompatibilitySingletonFactory;
52 import org.apache.hadoop.hbase.HBaseConfiguration;
53 import org.apache.hadoop.hbase.HBaseTestingUtility;
54 import org.apache.hadoop.hbase.HColumnDescriptor;
55 import org.apache.hadoop.hbase.HConstants;
56 import org.apache.hadoop.hbase.HDFSBlocksDistribution;
57 import org.apache.hadoop.hbase.HTableDescriptor;
58 import org.apache.hadoop.hbase.HadoopShims;
59 import org.apache.hadoop.hbase.KeyValue;
60 import org.apache.hadoop.hbase.PerformanceEvaluation;
61 import org.apache.hadoop.hbase.TableName;
62 import org.apache.hadoop.hbase.Tag;
63 import org.apache.hadoop.hbase.TagType;
64 import org.apache.hadoop.hbase.client.Admin;
65 import org.apache.hadoop.hbase.client.Connection;
66 import org.apache.hadoop.hbase.client.ConnectionFactory;
67 import org.apache.hadoop.hbase.client.HTable;
68 import org.apache.hadoop.hbase.client.Put;
69 import org.apache.hadoop.hbase.client.RegionLocator;
70 import org.apache.hadoop.hbase.client.Result;
71 import org.apache.hadoop.hbase.client.ResultScanner;
72 import org.apache.hadoop.hbase.client.Scan;
73 import org.apache.hadoop.hbase.client.Table;
74 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
75 import org.apache.hadoop.hbase.io.compress.Compression;
76 import org.apache.hadoop.hbase.io.compress.Compression.Algorithm;
77 import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
78 import org.apache.hadoop.hbase.io.hfile.CacheConfig;
79 import org.apache.hadoop.hbase.io.hfile.HFile;
80 import org.apache.hadoop.hbase.io.hfile.HFile.Reader;
81 import org.apache.hadoop.hbase.mapreduce.TestImportTSVWithTTLs.TTLCheckingObserver;
82 import org.apache.hadoop.hbase.io.hfile.HFileScanner;
83 import org.apache.hadoop.hbase.regionserver.BloomType;
84 import org.apache.hadoop.hbase.regionserver.HRegion;
85 import org.apache.hadoop.hbase.regionserver.Store;
86 import org.apache.hadoop.hbase.regionserver.StoreFile;
87 import org.apache.hadoop.hbase.regionserver.TimeRangeTracker;
88 import org.apache.hadoop.hbase.testclassification.LargeTests;
89 import org.apache.hadoop.hbase.util.Bytes;
90 import org.apache.hadoop.hbase.util.FSUtils;
91 import org.apache.hadoop.hbase.util.Threads;
92 import org.apache.hadoop.hbase.util.Writables;
93 import org.apache.hadoop.io.NullWritable;
94 import org.apache.hadoop.mapreduce.Job;
95 import org.apache.hadoop.mapreduce.Mapper;
96 import org.apache.hadoop.mapreduce.RecordWriter;
97 import org.apache.hadoop.mapreduce.TaskAttemptContext;
98 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
99 import org.junit.Ignore;
100 import org.junit.Rule;
101 import org.junit.Test;
102 import org.junit.experimental.categories.Category;
103 import org.junit.rules.TestRule;
104 import org.mockito.Mockito;
105
106 import com.google.common.collect.Lists;
107
108
109
110
111
112
113
114 @Category(LargeTests.class)
115 public class TestHFileOutputFormat2 {
116 @Rule public final TestRule timeout = CategoryBasedTimeout.builder().
117 withTimeout(this.getClass()).withLookingForStuckThread(true).build();
118 private final static int ROWSPERSPLIT = 1024;
119
120 private static final byte[] FAMILY_NAME = Bytes.toBytes("info");
121 private static final byte[][] FAMILIES = {
122 Bytes.add(FAMILY_NAME, Bytes.toBytes("-A")),
123 Bytes.add(FAMILY_NAME, Bytes.toBytes("-B"))
124 };
125 private static final TableName TABLE_NAME =
126 TableName.valueOf("TestTable");
127
128 private HBaseTestingUtility util = new HBaseTestingUtility();
129
130 private static final Log LOG = LogFactory.getLog(TestHFileOutputFormat2.class);
131
132
133
134
135 static class RandomKVGeneratingMapper
136 extends Mapper<NullWritable, NullWritable,
137 ImmutableBytesWritable, Cell> {
138
139 private int keyLength;
140 private static final int KEYLEN_DEFAULT=10;
141 private static final String KEYLEN_CONF="randomkv.key.length";
142
143 private int valLength;
144 private static final int VALLEN_DEFAULT=10;
145 private static final String VALLEN_CONF="randomkv.val.length";
146 private static final byte [] QUALIFIER = Bytes.toBytes("data");
147
148 @Override
149 protected void setup(Context context) throws IOException,
150 InterruptedException {
151 super.setup(context);
152
153 Configuration conf = context.getConfiguration();
154 keyLength = conf.getInt(KEYLEN_CONF, KEYLEN_DEFAULT);
155 valLength = conf.getInt(VALLEN_CONF, VALLEN_DEFAULT);
156 }
157
158 @Override
159 protected void map(
160 NullWritable n1, NullWritable n2,
161 Mapper<NullWritable, NullWritable,
162 ImmutableBytesWritable,Cell>.Context context)
163 throws java.io.IOException ,InterruptedException
164 {
165
166 byte keyBytes[] = new byte[keyLength];
167 byte valBytes[] = new byte[valLength];
168
169 int taskId = context.getTaskAttemptID().getTaskID().getId();
170 assert taskId < Byte.MAX_VALUE : "Unit tests dont support > 127 tasks!";
171
172 Random random = new Random();
173 for (int i = 0; i < ROWSPERSPLIT; i++) {
174
175 random.nextBytes(keyBytes);
176
177 keyBytes[keyLength - 1] = (byte)(taskId & 0xFF);
178 random.nextBytes(valBytes);
179 ImmutableBytesWritable key = new ImmutableBytesWritable(keyBytes);
180
181 for (byte[] family : TestHFileOutputFormat2.FAMILIES) {
182 Cell kv = new KeyValue(keyBytes, family, QUALIFIER, valBytes);
183 context.write(key, kv);
184 }
185 }
186 }
187 }
188
189
190
191
192 static class RandomPutGeneratingMapper
193 extends Mapper<NullWritable, NullWritable,
194 ImmutableBytesWritable, Put> {
195
196 private int keyLength;
197 private static final int KEYLEN_DEFAULT=10;
198 private static final String KEYLEN_CONF="randomkv.key.length";
199
200 private int valLength;
201 private static final int VALLEN_DEFAULT=10;
202 private static final String VALLEN_CONF="randomkv.val.length";
203 private static final byte [] QUALIFIER = Bytes.toBytes("data");
204
205 @Override
206 protected void setup(Context context) throws IOException,
207 InterruptedException {
208 super.setup(context);
209
210 Configuration conf = context.getConfiguration();
211 keyLength = conf.getInt(KEYLEN_CONF, KEYLEN_DEFAULT);
212 valLength = conf.getInt(VALLEN_CONF, VALLEN_DEFAULT);
213 }
214
215 @Override
216 protected void map(
217 NullWritable n1, NullWritable n2,
218 Mapper<NullWritable, NullWritable,
219 ImmutableBytesWritable,Put>.Context context)
220 throws java.io.IOException ,InterruptedException
221 {
222
223 byte keyBytes[] = new byte[keyLength];
224 byte valBytes[] = new byte[valLength];
225
226 int taskId = context.getTaskAttemptID().getTaskID().getId();
227 assert taskId < Byte.MAX_VALUE : "Unit tests dont support > 127 tasks!";
228
229 Random random = new Random();
230 for (int i = 0; i < ROWSPERSPLIT; i++) {
231
232 random.nextBytes(keyBytes);
233
234 keyBytes[keyLength - 1] = (byte)(taskId & 0xFF);
235 random.nextBytes(valBytes);
236 ImmutableBytesWritable key = new ImmutableBytesWritable(keyBytes);
237
238 for (byte[] family : TestHFileOutputFormat2.FAMILIES) {
239 Put p = new Put(keyBytes);
240 p.addColumn(family, QUALIFIER, valBytes);
241
242 p.setTTL(1l);
243 context.write(key, p);
244 }
245 }
246 }
247 }
248
249 private void setupRandomGeneratorMapper(Job job, boolean putSortReducer) {
250 if (putSortReducer) {
251 job.setInputFormatClass(NMapInputFormat.class);
252 job.setMapperClass(RandomPutGeneratingMapper.class);
253 job.setMapOutputKeyClass(ImmutableBytesWritable.class);
254 job.setMapOutputValueClass(Put.class);
255 } else {
256 job.setInputFormatClass(NMapInputFormat.class);
257 job.setMapperClass(RandomKVGeneratingMapper.class);
258 job.setMapOutputKeyClass(ImmutableBytesWritable.class);
259 job.setMapOutputValueClass(KeyValue.class);
260 }
261 }
262
263
264
265
266
267
268 @Ignore("Goes zombie too frequently; needs work. See HBASE-14563") @Test
269 public void test_LATEST_TIMESTAMP_isReplaced()
270 throws Exception {
271 Configuration conf = new Configuration(this.util.getConfiguration());
272 RecordWriter<ImmutableBytesWritable, Cell> writer = null;
273 TaskAttemptContext context = null;
274 Path dir =
275 util.getDataTestDir("test_LATEST_TIMESTAMP_isReplaced");
276 try {
277 Job job = new Job(conf);
278 FileOutputFormat.setOutputPath(job, dir);
279 context = createTestTaskAttemptContext(job);
280 HFileOutputFormat2 hof = new HFileOutputFormat2();
281 writer = hof.getRecordWriter(context);
282 final byte [] b = Bytes.toBytes("b");
283
284
285
286 KeyValue kv = new KeyValue(b, b, b);
287 KeyValue original = kv.clone();
288 writer.write(new ImmutableBytesWritable(), kv);
289 assertFalse(original.equals(kv));
290 assertTrue(Bytes.equals(CellUtil.cloneRow(original), CellUtil.cloneRow(kv)));
291 assertTrue(Bytes.equals(CellUtil.cloneFamily(original), CellUtil.cloneFamily(kv)));
292 assertTrue(Bytes.equals(CellUtil.cloneQualifier(original), CellUtil.cloneQualifier(kv)));
293 assertNotSame(original.getTimestamp(), kv.getTimestamp());
294 assertNotSame(HConstants.LATEST_TIMESTAMP, kv.getTimestamp());
295
296
297
298 kv = new KeyValue(b, b, b, kv.getTimestamp() - 1, b);
299 original = kv.clone();
300 writer.write(new ImmutableBytesWritable(), kv);
301 assertTrue(original.equals(kv));
302 } finally {
303 if (writer != null && context != null) writer.close(context);
304 dir.getFileSystem(conf).delete(dir, true);
305 }
306 }
307
308 private TaskAttemptContext createTestTaskAttemptContext(final Job job)
309 throws Exception {
310 HadoopShims hadoop = CompatibilitySingletonFactory.getInstance(HadoopShims.class);
311 TaskAttemptContext context = hadoop.createTestTaskAttemptContext(
312 job, "attempt_201402131733_0001_m_000000_0");
313 return context;
314 }
315
316
317
318
319
320 @Ignore("Goes zombie too frequently; needs work. See HBASE-14563") @Test
321 public void test_TIMERANGE() throws Exception {
322 Configuration conf = new Configuration(this.util.getConfiguration());
323 RecordWriter<ImmutableBytesWritable, Cell> writer = null;
324 TaskAttemptContext context = null;
325 Path dir =
326 util.getDataTestDir("test_TIMERANGE_present");
327 LOG.info("Timerange dir writing to dir: " + dir);
328 try {
329
330 Job job = new Job(conf);
331 FileOutputFormat.setOutputPath(job, dir);
332 context = createTestTaskAttemptContext(job);
333 HFileOutputFormat2 hof = new HFileOutputFormat2();
334 writer = hof.getRecordWriter(context);
335
336
337 final byte [] b = Bytes.toBytes("b");
338
339
340 KeyValue kv = new KeyValue(b, b, b, 2000, b);
341 KeyValue original = kv.clone();
342 writer.write(new ImmutableBytesWritable(), kv);
343 assertEquals(original,kv);
344
345
346 kv = new KeyValue(b, b, b, 1000, b);
347 original = kv.clone();
348 writer.write(new ImmutableBytesWritable(), kv);
349 assertEquals(original, kv);
350
351
352 writer.close(context);
353
354
355
356
357 FileSystem fs = FileSystem.get(conf);
358 Path attemptDirectory = hof.getDefaultWorkFile(context, "").getParent();
359 FileStatus[] sub1 = fs.listStatus(attemptDirectory);
360 FileStatus[] file = fs.listStatus(sub1[0].getPath());
361
362
363 HFile.Reader rd = HFile.createReader(fs, file[0].getPath(),
364 new CacheConfig(conf), conf);
365 Map<byte[],byte[]> finfo = rd.loadFileInfo();
366 byte[] range = finfo.get("TIMERANGE".getBytes());
367 assertNotNull(range);
368
369
370 TimeRangeTracker timeRangeTracker = new TimeRangeTracker();
371 Writables.copyWritable(range, timeRangeTracker);
372 LOG.info(timeRangeTracker.getMin() +
373 "...." + timeRangeTracker.getMax());
374 assertEquals(1000, timeRangeTracker.getMin());
375 assertEquals(2000, timeRangeTracker.getMax());
376 rd.close();
377 } finally {
378 if (writer != null && context != null) writer.close(context);
379 dir.getFileSystem(conf).delete(dir, true);
380 }
381 }
382
383
384
385
386 @Ignore("Goes zombie too frequently; needs work. See HBASE-14563") @Test
387 public void testWritingPEData() throws Exception {
388 Configuration conf = util.getConfiguration();
389 Path testDir = util.getDataTestDirOnTestFS("testWritingPEData");
390 FileSystem fs = testDir.getFileSystem(conf);
391
392
393 conf.setInt("mapreduce.task.io.sort.mb", 20);
394
395 conf.setLong(HConstants.HREGION_MAX_FILESIZE, 64 * 1024);
396
397 Job job = new Job(conf, "testWritingPEData");
398 setupRandomGeneratorMapper(job, false);
399
400
401 byte[] startKey = new byte[RandomKVGeneratingMapper.KEYLEN_DEFAULT];
402 byte[] endKey = new byte[RandomKVGeneratingMapper.KEYLEN_DEFAULT];
403
404 Arrays.fill(startKey, (byte)0);
405 Arrays.fill(endKey, (byte)0xff);
406
407 job.setPartitionerClass(SimpleTotalOrderPartitioner.class);
408
409 SimpleTotalOrderPartitioner.setStartKey(job.getConfiguration(), startKey);
410 SimpleTotalOrderPartitioner.setEndKey(job.getConfiguration(), endKey);
411 job.setReducerClass(KeyValueSortReducer.class);
412 job.setOutputFormatClass(HFileOutputFormat2.class);
413 job.setNumReduceTasks(4);
414 job.getConfiguration().setStrings("io.serializations", conf.get("io.serializations"),
415 MutationSerialization.class.getName(), ResultSerialization.class.getName(),
416 KeyValueSerialization.class.getName());
417
418 FileOutputFormat.setOutputPath(job, testDir);
419 assertTrue(job.waitForCompletion(false));
420 FileStatus [] files = fs.listStatus(testDir);
421 assertTrue(files.length > 0);
422 }
423
424
425
426
427
428 @Test
429 public void test_WritingTagData()
430 throws Exception {
431 Configuration conf = new Configuration(this.util.getConfiguration());
432 final String HFILE_FORMAT_VERSION_CONF_KEY = "hfile.format.version";
433 conf.setInt(HFILE_FORMAT_VERSION_CONF_KEY, HFile.MIN_FORMAT_VERSION_WITH_TAGS);
434 RecordWriter<ImmutableBytesWritable, Cell> writer = null;
435 TaskAttemptContext context = null;
436 Path dir =
437 util.getDataTestDir("WritingTagData");
438 try {
439 Job job = new Job(conf);
440 FileOutputFormat.setOutputPath(job, dir);
441 context = createTestTaskAttemptContext(job);
442 HFileOutputFormat2 hof = new HFileOutputFormat2();
443 writer = hof.getRecordWriter(context);
444 final byte [] b = Bytes.toBytes("b");
445
446 KeyValue kv = new KeyValue(b, b, b, HConstants.LATEST_TIMESTAMP, b, new Tag[] {
447 new Tag(TagType.TTL_TAG_TYPE, Bytes.toBytes(978670)) });
448 writer.write(new ImmutableBytesWritable(), kv);
449 writer.close(context);
450 writer = null;
451 FileSystem fs = dir.getFileSystem(conf);
452 RemoteIterator<LocatedFileStatus> iterator = fs.listFiles(dir, true);
453 while(iterator.hasNext()) {
454 LocatedFileStatus keyFileStatus = iterator.next();
455 HFile.Reader reader = HFile.createReader(fs, keyFileStatus.getPath(), new CacheConfig(conf),
456 conf);
457 HFileScanner scanner = reader.getScanner(false, false, false);
458 scanner.seekTo();
459 Cell cell = scanner.getKeyValue();
460
461 Iterator<Tag> tagsIterator = CellUtil.tagsIterator(cell.getTagsArray(),
462 cell.getTagsOffset(), cell.getTagsLength());
463 assertTrue(tagsIterator.hasNext());
464 assertTrue(tagsIterator.next().getType() == TagType.TTL_TAG_TYPE);
465 }
466 } finally {
467 if (writer != null && context != null) writer.close(context);
468 dir.getFileSystem(conf).delete(dir, true);
469 }
470 }
471
472 @Ignore("Goes zombie too frequently; needs work. See HBASE-14563") @Test
473 public void testJobConfiguration() throws Exception {
474 Configuration conf = new Configuration(this.util.getConfiguration());
475 conf.set(HConstants.TEMPORARY_FS_DIRECTORY_KEY, util.getDataTestDir("testJobConfiguration")
476 .toString());
477 Job job = new Job(conf);
478 job.setWorkingDirectory(util.getDataTestDir("testJobConfiguration"));
479 RegionLocator regionLocator = Mockito.mock(RegionLocator.class);
480 setupMockStartKeys(regionLocator);
481 setupMockTableName(regionLocator);
482 HFileOutputFormat2.configureIncrementalLoad(job, new HTableDescriptor(), regionLocator);
483 assertEquals(job.getNumReduceTasks(), 4);
484 }
485
486 private byte [][] generateRandomStartKeys(int numKeys) {
487 Random random = new Random();
488 byte[][] ret = new byte[numKeys][];
489
490 ret[0] = HConstants.EMPTY_BYTE_ARRAY;
491 for (int i = 1; i < numKeys; i++) {
492 ret[i] =
493 PerformanceEvaluation.generateData(random, PerformanceEvaluation.DEFAULT_VALUE_LENGTH);
494 }
495 return ret;
496 }
497
498 private byte[][] generateRandomSplitKeys(int numKeys) {
499 Random random = new Random();
500 byte[][] ret = new byte[numKeys][];
501 for (int i = 0; i < numKeys; i++) {
502 ret[i] =
503 PerformanceEvaluation.generateData(random, PerformanceEvaluation.DEFAULT_VALUE_LENGTH);
504 }
505 return ret;
506 }
507
508 @Ignore("Goes zombie too frequently; needs work. See HBASE-14563") @Test
509 public void testMRIncrementalLoad() throws Exception {
510 LOG.info("\nStarting test testMRIncrementalLoad\n");
511 doIncrementalLoadTest(false, false, false, "testMRIncrementalLoad");
512 }
513
514 @Ignore("Goes zombie too frequently; needs work. See HBASE-14563") @Test
515 public void testMRIncrementalLoadWithSplit() throws Exception {
516 LOG.info("\nStarting test testMRIncrementalLoadWithSplit\n");
517 doIncrementalLoadTest(true, false, false, "testMRIncrementalLoadWithSplit");
518 }
519
520
521
522
523
524
525
526
527
528 @Ignore("Goes zombie too frequently; needs work. See HBASE-14563") @Test
529 public void testMRIncrementalLoadWithLocality() throws Exception {
530 LOG.info("\nStarting test testMRIncrementalLoadWithLocality\n");
531 doIncrementalLoadTest(false, true, false, "testMRIncrementalLoadWithLocality1");
532 doIncrementalLoadTest(true, true, false, "testMRIncrementalLoadWithLocality2");
533 }
534
535 @Ignore("Goes zombie too frequently; needs work. See HBASE-14563") @Test
536 public void testMRIncrementalLoadWithPutSortReducer() throws Exception {
537 LOG.info("\nStarting test testMRIncrementalLoadWithPutSortReducer\n");
538 doIncrementalLoadTest(false, false, true, "testMRIncrementalLoadWithPutSortReducer");
539 }
540
541 private void doIncrementalLoadTest(boolean shouldChangeRegions, boolean shouldKeepLocality,
542 boolean putSortReducer, String tableStr) throws Exception {
543 util = new HBaseTestingUtility();
544 Configuration conf = util.getConfiguration();
545 conf.setBoolean(HFileOutputFormat2.LOCALITY_SENSITIVE_CONF_KEY, shouldKeepLocality);
546 int hostCount = 1;
547 int regionNum = 5;
548 if (shouldKeepLocality) {
549
550
551 hostCount = 3;
552 regionNum = 20;
553 }
554 byte[][] splitKeys = generateRandomSplitKeys(regionNum - 1);
555 String[] hostnames = new String[hostCount];
556 for (int i = 0; i < hostCount; ++i) {
557 hostnames[i] = "datanode_" + i;
558 }
559 util.setJobWithoutMRCluster();
560 util.startMiniCluster(1, hostCount, hostnames);
561
562 HTable table = util.createTable(TABLE_NAME, FAMILIES, splitKeys);
563 Path testDir = util.getDataTestDirOnTestFS("testLocalMRIncrementalLoad");
564 try (RegionLocator r = table.getRegionLocator(); Admin admin = table.getConnection().getAdmin()) {
565 assertEquals("Should start with empty table", 0, util.countRows(table));
566 int numRegions = r.getStartKeys().length;
567 assertEquals("Should make " + regionNum + " regions", numRegions, regionNum);
568
569
570 runIncrementalPELoad(conf, table.getTableDescriptor(), table.getRegionLocator(), testDir,
571 putSortReducer);
572
573 assertEquals("HFOF should not touch actual table", 0, util.countRows(table));
574
575
576 int dir = 0;
577 for (FileStatus f : testDir.getFileSystem(conf).listStatus(testDir)) {
578 for (byte[] family : FAMILIES) {
579 if (Bytes.toString(family).equals(f.getPath().getName())) {
580 ++dir;
581 }
582 }
583 }
584 assertEquals("Column family not found in FS.", FAMILIES.length, dir);
585
586
587 if (shouldChangeRegions) {
588 LOG.info("Changing regions in table");
589 admin.disableTable(table.getName());
590 util.waitUntilNoRegionsInTransition();
591
592 util.deleteTable(table.getName());
593 byte[][] newSplitKeys = generateRandomSplitKeys(14);
594 table = util.createTable(TABLE_NAME, FAMILIES, newSplitKeys);
595
596 while (table.getRegionLocator().getAllRegionLocations().size() != 15
597 || !admin.isTableAvailable(table.getName())) {
598 Thread.sleep(200);
599 LOG.info("Waiting for new region assignment to happen");
600 }
601 }
602
603
604 new LoadIncrementalHFiles(conf).doBulkLoad(testDir, table);
605
606 int expectedRows = 0;
607 if (putSortReducer) {
608
609 assertEquals("LoadIncrementalHFiles should put expected data in table", expectedRows,
610 util.countRows(table));
611 } else {
612
613 expectedRows = NMapInputFormat.getNumMapTasks(conf) * ROWSPERSPLIT;
614 assertEquals("LoadIncrementalHFiles should put expected data in table", expectedRows,
615 util.countRows(table));
616 Scan scan = new Scan();
617 ResultScanner results = table.getScanner(scan);
618 for (Result res : results) {
619 assertEquals(FAMILIES.length, res.rawCells().length);
620 Cell first = res.rawCells()[0];
621 for (Cell kv : res.rawCells()) {
622 assertTrue(CellUtil.matchingRow(first, kv));
623 assertTrue(Bytes.equals(CellUtil.cloneValue(first), CellUtil.cloneValue(kv)));
624 }
625 }
626 results.close();
627 }
628 String tableDigestBefore = util.checksumRows(table);
629
630
631 HDFSBlocksDistribution hbd = new HDFSBlocksDistribution();
632 for (HRegion region : util.getHBaseCluster().getRegions(TABLE_NAME)) {
633 hbd.add(region.getHDFSBlocksDistribution());
634 }
635 for (String hostname : hostnames) {
636 float locality = hbd.getBlockLocalityIndex(hostname);
637 LOG.info("locality of [" + hostname + "]: " + locality);
638 assertEquals(100, (int) (locality * 100));
639 }
640
641
642 admin.disableTable(TABLE_NAME);
643 while (!admin.isTableDisabled(TABLE_NAME)) {
644 Thread.sleep(200);
645 LOG.info("Waiting for table to disable");
646 }
647 admin.enableTable(TABLE_NAME);
648 util.waitTableAvailable(TABLE_NAME);
649 assertEquals("Data should remain after reopening of regions", tableDigestBefore,
650 util.checksumRows(table));
651 } finally {
652 testDir.getFileSystem(conf).delete(testDir, true);
653 util.deleteTable(TABLE_NAME);
654 util.shutdownMiniCluster();
655 }
656 }
657
658 private void runIncrementalPELoad(Configuration conf, HTableDescriptor tableDescriptor,
659 RegionLocator regionLocator, Path outDir, boolean putSortReducer) throws IOException,
660 UnsupportedEncodingException, InterruptedException, ClassNotFoundException {
661 Job job = new Job(conf, "testLocalMRIncrementalLoad");
662 job.setWorkingDirectory(util.getDataTestDirOnTestFS("runIncrementalPELoad"));
663 job.getConfiguration().setStrings("io.serializations", conf.get("io.serializations"),
664 MutationSerialization.class.getName(), ResultSerialization.class.getName(),
665 KeyValueSerialization.class.getName());
666 setupRandomGeneratorMapper(job, putSortReducer);
667 HFileOutputFormat2.configureIncrementalLoad(job, tableDescriptor, regionLocator);
668 FileOutputFormat.setOutputPath(job, outDir);
669
670 assertFalse(util.getTestFileSystem().exists(outDir)) ;
671
672 assertEquals(regionLocator.getAllRegionLocations().size(), job.getNumReduceTasks());
673
674 assertTrue(job.waitForCompletion(true));
675 }
676
677
678
679
680
681
682
683
684
685
686 @Ignore("Goes zombie too frequently; needs work. See HBASE-14563") @Test
687 public void testSerializeDeserializeFamilyCompressionMap() throws IOException {
688 for (int numCfs = 0; numCfs <= 3; numCfs++) {
689 Configuration conf = new Configuration(this.util.getConfiguration());
690 Map<String, Compression.Algorithm> familyToCompression =
691 getMockColumnFamiliesForCompression(numCfs);
692 Table table = Mockito.mock(HTable.class);
693 setupMockColumnFamiliesForCompression(table, familyToCompression);
694 HFileOutputFormat2.configureCompression(conf, table.getTableDescriptor());
695
696
697 Map<byte[], Algorithm> retrievedFamilyToCompressionMap = HFileOutputFormat2
698 .createFamilyCompressionMap(conf);
699
700
701
702 for (Entry<String, Algorithm> entry : familyToCompression.entrySet()) {
703 assertEquals("Compression configuration incorrect for column family:"
704 + entry.getKey(), entry.getValue(),
705 retrievedFamilyToCompressionMap.get(entry.getKey().getBytes()));
706 }
707 }
708 }
709
710 private void setupMockColumnFamiliesForCompression(Table table,
711 Map<String, Compression.Algorithm> familyToCompression) throws IOException {
712 HTableDescriptor mockTableDescriptor = new HTableDescriptor(TABLE_NAME);
713 for (Entry<String, Compression.Algorithm> entry : familyToCompression.entrySet()) {
714 mockTableDescriptor.addFamily(new HColumnDescriptor(entry.getKey())
715 .setMaxVersions(1)
716 .setCompressionType(entry.getValue())
717 .setBlockCacheEnabled(false)
718 .setTimeToLive(0));
719 }
720 Mockito.doReturn(mockTableDescriptor).when(table).getTableDescriptor();
721 }
722
723
724
725
726
727 private Map<String, Compression.Algorithm>
728 getMockColumnFamiliesForCompression (int numCfs) {
729 Map<String, Compression.Algorithm> familyToCompression
730 = new HashMap<String, Compression.Algorithm>();
731
732 if (numCfs-- > 0) {
733 familyToCompression.put("Family1!@#!@#&", Compression.Algorithm.LZO);
734 }
735 if (numCfs-- > 0) {
736 familyToCompression.put("Family2=asdads&!AASD", Compression.Algorithm.SNAPPY);
737 }
738 if (numCfs-- > 0) {
739 familyToCompression.put("Family2=asdads&!AASD", Compression.Algorithm.GZ);
740 }
741 if (numCfs-- > 0) {
742 familyToCompression.put("Family3", Compression.Algorithm.NONE);
743 }
744 return familyToCompression;
745 }
746
747
748
749
750
751
752
753
754
755
756
757 @Ignore("Goes zombie too frequently; needs work. See HBASE-14563") @Test
758 public void testSerializeDeserializeFamilyBloomTypeMap() throws IOException {
759 for (int numCfs = 0; numCfs <= 2; numCfs++) {
760 Configuration conf = new Configuration(this.util.getConfiguration());
761 Map<String, BloomType> familyToBloomType =
762 getMockColumnFamiliesForBloomType(numCfs);
763 Table table = Mockito.mock(HTable.class);
764 setupMockColumnFamiliesForBloomType(table,
765 familyToBloomType);
766 HFileOutputFormat2.configureBloomType(table.getTableDescriptor(), conf);
767
768
769
770 Map<byte[], BloomType> retrievedFamilyToBloomTypeMap =
771 HFileOutputFormat2
772 .createFamilyBloomTypeMap(conf);
773
774
775
776 for (Entry<String, BloomType> entry : familyToBloomType.entrySet()) {
777 assertEquals("BloomType configuration incorrect for column family:"
778 + entry.getKey(), entry.getValue(),
779 retrievedFamilyToBloomTypeMap.get(entry.getKey().getBytes()));
780 }
781 }
782 }
783
784 private void setupMockColumnFamiliesForBloomType(Table table,
785 Map<String, BloomType> familyToDataBlockEncoding) throws IOException {
786 HTableDescriptor mockTableDescriptor = new HTableDescriptor(TABLE_NAME);
787 for (Entry<String, BloomType> entry : familyToDataBlockEncoding.entrySet()) {
788 mockTableDescriptor.addFamily(new HColumnDescriptor(entry.getKey())
789 .setMaxVersions(1)
790 .setBloomFilterType(entry.getValue())
791 .setBlockCacheEnabled(false)
792 .setTimeToLive(0));
793 }
794 Mockito.doReturn(mockTableDescriptor).when(table).getTableDescriptor();
795 }
796
797
798
799
800
801 private Map<String, BloomType>
802 getMockColumnFamiliesForBloomType (int numCfs) {
803 Map<String, BloomType> familyToBloomType =
804 new HashMap<String, BloomType>();
805
806 if (numCfs-- > 0) {
807 familyToBloomType.put("Family1!@#!@#&", BloomType.ROW);
808 }
809 if (numCfs-- > 0) {
810 familyToBloomType.put("Family2=asdads&!AASD",
811 BloomType.ROWCOL);
812 }
813 if (numCfs-- > 0) {
814 familyToBloomType.put("Family3", BloomType.NONE);
815 }
816 return familyToBloomType;
817 }
818
819
820
821
822
823
824
825
826
827
828 @Ignore("Goes zombie too frequently; needs work. See HBASE-14563") @Test
829 public void testSerializeDeserializeFamilyBlockSizeMap() throws IOException {
830 for (int numCfs = 0; numCfs <= 3; numCfs++) {
831 Configuration conf = new Configuration(this.util.getConfiguration());
832 Map<String, Integer> familyToBlockSize =
833 getMockColumnFamiliesForBlockSize(numCfs);
834 Table table = Mockito.mock(HTable.class);
835 setupMockColumnFamiliesForBlockSize(table,
836 familyToBlockSize);
837 HFileOutputFormat2.configureBlockSize(table.getTableDescriptor(), conf);
838
839
840
841 Map<byte[], Integer> retrievedFamilyToBlockSizeMap =
842 HFileOutputFormat2
843 .createFamilyBlockSizeMap(conf);
844
845
846
847 for (Entry<String, Integer> entry : familyToBlockSize.entrySet()
848 ) {
849 assertEquals("BlockSize configuration incorrect for column family:"
850 + entry.getKey(), entry.getValue(),
851 retrievedFamilyToBlockSizeMap.get(entry.getKey().getBytes()));
852 }
853 }
854 }
855
856 private void setupMockColumnFamiliesForBlockSize(Table table,
857 Map<String, Integer> familyToDataBlockEncoding) throws IOException {
858 HTableDescriptor mockTableDescriptor = new HTableDescriptor(TABLE_NAME);
859 for (Entry<String, Integer> entry : familyToDataBlockEncoding.entrySet()) {
860 mockTableDescriptor.addFamily(new HColumnDescriptor(entry.getKey())
861 .setMaxVersions(1)
862 .setBlocksize(entry.getValue())
863 .setBlockCacheEnabled(false)
864 .setTimeToLive(0));
865 }
866 Mockito.doReturn(mockTableDescriptor).when(table).getTableDescriptor();
867 }
868
869
870
871
872
873 private Map<String, Integer>
874 getMockColumnFamiliesForBlockSize (int numCfs) {
875 Map<String, Integer> familyToBlockSize =
876 new HashMap<String, Integer>();
877
878 if (numCfs-- > 0) {
879 familyToBlockSize.put("Family1!@#!@#&", 1234);
880 }
881 if (numCfs-- > 0) {
882 familyToBlockSize.put("Family2=asdads&!AASD",
883 Integer.MAX_VALUE);
884 }
885 if (numCfs-- > 0) {
886 familyToBlockSize.put("Family2=asdads&!AASD",
887 Integer.MAX_VALUE);
888 }
889 if (numCfs-- > 0) {
890 familyToBlockSize.put("Family3", 0);
891 }
892 return familyToBlockSize;
893 }
894
895
896
897
898
899
900
901
902
903 @Ignore("Goes zombie too frequently; needs work. See HBASE-14563") @Test
904 public void testSerializeDeserializeFamilyDataBlockEncodingMap() throws IOException {
905 for (int numCfs = 0; numCfs <= 3; numCfs++) {
906 Configuration conf = new Configuration(this.util.getConfiguration());
907 Map<String, DataBlockEncoding> familyToDataBlockEncoding =
908 getMockColumnFamiliesForDataBlockEncoding(numCfs);
909 Table table = Mockito.mock(HTable.class);
910 setupMockColumnFamiliesForDataBlockEncoding(table,
911 familyToDataBlockEncoding);
912 HTableDescriptor tableDescriptor = table.getTableDescriptor();
913 HFileOutputFormat2.configureDataBlockEncoding(tableDescriptor, conf);
914
915
916
917 Map<byte[], DataBlockEncoding> retrievedFamilyToDataBlockEncodingMap =
918 HFileOutputFormat2
919 .createFamilyDataBlockEncodingMap(conf);
920
921
922
923 for (Entry<String, DataBlockEncoding> entry : familyToDataBlockEncoding.entrySet()) {
924 assertEquals("DataBlockEncoding configuration incorrect for column family:"
925 + entry.getKey(), entry.getValue(),
926 retrievedFamilyToDataBlockEncodingMap.get(entry.getKey().getBytes()));
927 }
928 }
929 }
930
931 private void setupMockColumnFamiliesForDataBlockEncoding(Table table,
932 Map<String, DataBlockEncoding> familyToDataBlockEncoding) throws IOException {
933 HTableDescriptor mockTableDescriptor = new HTableDescriptor(TABLE_NAME);
934 for (Entry<String, DataBlockEncoding> entry : familyToDataBlockEncoding.entrySet()) {
935 mockTableDescriptor.addFamily(new HColumnDescriptor(entry.getKey())
936 .setMaxVersions(1)
937 .setDataBlockEncoding(entry.getValue())
938 .setBlockCacheEnabled(false)
939 .setTimeToLive(0));
940 }
941 Mockito.doReturn(mockTableDescriptor).when(table).getTableDescriptor();
942 }
943
944
945
946
947
948 private Map<String, DataBlockEncoding>
949 getMockColumnFamiliesForDataBlockEncoding (int numCfs) {
950 Map<String, DataBlockEncoding> familyToDataBlockEncoding =
951 new HashMap<String, DataBlockEncoding>();
952
953 if (numCfs-- > 0) {
954 familyToDataBlockEncoding.put("Family1!@#!@#&", DataBlockEncoding.DIFF);
955 }
956 if (numCfs-- > 0) {
957 familyToDataBlockEncoding.put("Family2=asdads&!AASD",
958 DataBlockEncoding.FAST_DIFF);
959 }
960 if (numCfs-- > 0) {
961 familyToDataBlockEncoding.put("Family2=asdads&!AASD",
962 DataBlockEncoding.PREFIX);
963 }
964 if (numCfs-- > 0) {
965 familyToDataBlockEncoding.put("Family3", DataBlockEncoding.NONE);
966 }
967 return familyToDataBlockEncoding;
968 }
969
970 private void setupMockStartKeys(RegionLocator table) throws IOException {
971 byte[][] mockKeys = new byte[][] {
972 HConstants.EMPTY_BYTE_ARRAY,
973 Bytes.toBytes("aaa"),
974 Bytes.toBytes("ggg"),
975 Bytes.toBytes("zzz")
976 };
977 Mockito.doReturn(mockKeys).when(table).getStartKeys();
978 }
979
980 private void setupMockTableName(RegionLocator table) throws IOException {
981 TableName mockTableName = TableName.valueOf("mock_table");
982 Mockito.doReturn(mockTableName).when(table).getName();
983 }
984
985
986
987
988
989 @Ignore("Goes zombie too frequently; needs work. See HBASE-14563") @Test
990 public void testColumnFamilySettings() throws Exception {
991 Configuration conf = new Configuration(this.util.getConfiguration());
992 RecordWriter<ImmutableBytesWritable, Cell> writer = null;
993 TaskAttemptContext context = null;
994 Path dir = util.getDataTestDir("testColumnFamilySettings");
995
996
997 Table table = Mockito.mock(Table.class);
998 RegionLocator regionLocator = Mockito.mock(RegionLocator.class);
999 HTableDescriptor htd = new HTableDescriptor(TABLE_NAME);
1000 Mockito.doReturn(htd).when(table).getTableDescriptor();
1001 for (HColumnDescriptor hcd: HBaseTestingUtility.generateColumnDescriptors()) {
1002 htd.addFamily(hcd);
1003 }
1004
1005
1006 setupMockStartKeys(regionLocator);
1007
1008 try {
1009
1010
1011
1012 conf.set("io.seqfile.compression.type", "NONE");
1013 conf.set("hbase.fs.tmp.dir", dir.toString());
1014
1015 conf.setBoolean(HFileOutputFormat2.LOCALITY_SENSITIVE_CONF_KEY, false);
1016 Job job = new Job(conf, "testLocalMRIncrementalLoad");
1017 job.setWorkingDirectory(util.getDataTestDirOnTestFS("testColumnFamilySettings"));
1018 setupRandomGeneratorMapper(job, false);
1019 HFileOutputFormat2.configureIncrementalLoad(job, table.getTableDescriptor(), regionLocator);
1020 FileOutputFormat.setOutputPath(job, dir);
1021 context = createTestTaskAttemptContext(job);
1022 HFileOutputFormat2 hof = new HFileOutputFormat2();
1023 writer = hof.getRecordWriter(context);
1024
1025
1026 writeRandomKeyValues(writer, context, htd.getFamiliesKeys(), ROWSPERSPLIT);
1027 writer.close(context);
1028
1029
1030 FileSystem fs = dir.getFileSystem(conf);
1031
1032
1033 hof.getOutputCommitter(context).commitTask(context);
1034 hof.getOutputCommitter(context).commitJob(context);
1035 FileStatus[] families = FSUtils.listStatus(fs, dir, new FSUtils.FamilyDirFilter(fs));
1036 assertEquals(htd.getFamilies().size(), families.length);
1037 for (FileStatus f : families) {
1038 String familyStr = f.getPath().getName();
1039 HColumnDescriptor hcd = htd.getFamily(Bytes.toBytes(familyStr));
1040
1041
1042 Path dataFilePath = fs.listStatus(f.getPath())[0].getPath();
1043 Reader reader = HFile.createReader(fs, dataFilePath, new CacheConfig(conf), conf);
1044 Map<byte[], byte[]> fileInfo = reader.loadFileInfo();
1045
1046 byte[] bloomFilter = fileInfo.get(StoreFile.BLOOM_FILTER_TYPE_KEY);
1047 if (bloomFilter == null) bloomFilter = Bytes.toBytes("NONE");
1048 assertEquals("Incorrect bloom filter used for column family " + familyStr +
1049 "(reader: " + reader + ")",
1050 hcd.getBloomFilterType(), BloomType.valueOf(Bytes.toString(bloomFilter)));
1051 assertEquals("Incorrect compression used for column family " + familyStr +
1052 "(reader: " + reader + ")", hcd.getCompression(), reader.getFileContext().getCompression());
1053 }
1054 } finally {
1055 dir.getFileSystem(conf).delete(dir, true);
1056 }
1057 }
1058
1059
1060
1061
1062
1063 private void writeRandomKeyValues(RecordWriter<ImmutableBytesWritable, Cell> writer,
1064 TaskAttemptContext context, Set<byte[]> families, int numRows)
1065 throws IOException, InterruptedException {
1066 byte keyBytes[] = new byte[Bytes.SIZEOF_INT];
1067 int valLength = 10;
1068 byte valBytes[] = new byte[valLength];
1069
1070 int taskId = context.getTaskAttemptID().getTaskID().getId();
1071 assert taskId < Byte.MAX_VALUE : "Unit tests dont support > 127 tasks!";
1072 final byte [] qualifier = Bytes.toBytes("data");
1073 Random random = new Random();
1074 for (int i = 0; i < numRows; i++) {
1075
1076 Bytes.putInt(keyBytes, 0, i);
1077 random.nextBytes(valBytes);
1078 ImmutableBytesWritable key = new ImmutableBytesWritable(keyBytes);
1079
1080 for (byte[] family : families) {
1081 Cell kv = new KeyValue(keyBytes, family, qualifier, valBytes);
1082 writer.write(key, kv);
1083 }
1084 }
1085 }
1086
1087
1088
1089
1090
1091
1092
1093 @Ignore ("Flakey: See HBASE-9051") @Test
1094 public void testExcludeAllFromMinorCompaction() throws Exception {
1095 Configuration conf = util.getConfiguration();
1096 conf.setInt("hbase.hstore.compaction.min", 2);
1097 generateRandomStartKeys(5);
1098 util.setJobWithoutMRCluster();
1099 util.startMiniCluster();
1100 try (Connection conn = ConnectionFactory.createConnection();
1101 Admin admin = conn.getAdmin()) {
1102 final FileSystem fs = util.getDFSCluster().getFileSystem();
1103 HTable table = util.createTable(TABLE_NAME, FAMILIES);
1104 assertEquals("Should start with empty table", 0, util.countRows(table));
1105
1106
1107 final Path storePath = new Path(
1108 FSUtils.getTableDir(FSUtils.getRootDir(conf), TABLE_NAME),
1109 new Path(admin.getTableRegions(TABLE_NAME).get(0).getEncodedName(),
1110 Bytes.toString(FAMILIES[0])));
1111 assertEquals(0, fs.listStatus(storePath).length);
1112
1113
1114 conf.setBoolean("hbase.mapreduce.hfileoutputformat.compaction.exclude",
1115 true);
1116
1117 for (int i = 0; i < 2; i++) {
1118 Path testDir = util.getDataTestDirOnTestFS("testExcludeAllFromMinorCompaction_" + i);
1119 runIncrementalPELoad(conf, table.getTableDescriptor(), conn.getRegionLocator(TABLE_NAME),
1120 testDir, false);
1121
1122 new LoadIncrementalHFiles(conf).doBulkLoad(testDir, table);
1123 }
1124
1125
1126 int expectedRows = 2 * NMapInputFormat.getNumMapTasks(conf) * ROWSPERSPLIT;
1127 assertEquals("LoadIncrementalHFiles should put expected data in table",
1128 expectedRows, util.countRows(table));
1129
1130
1131 assertEquals(2, fs.listStatus(storePath).length);
1132
1133
1134 admin.compact(TABLE_NAME);
1135 try {
1136 quickPoll(new Callable<Boolean>() {
1137 @Override
1138 public Boolean call() throws Exception {
1139 List<HRegion> regions = util.getMiniHBaseCluster().getRegions(TABLE_NAME);
1140 for (HRegion region : regions) {
1141 for (Store store : region.getStores()) {
1142 store.closeAndArchiveCompactedFiles();
1143 }
1144 }
1145 return fs.listStatus(storePath).length == 1;
1146 }
1147 }, 5000);
1148 throw new IOException("SF# = " + fs.listStatus(storePath).length);
1149 } catch (AssertionError ae) {
1150
1151 }
1152
1153
1154 admin.majorCompact(TABLE_NAME);
1155 quickPoll(new Callable<Boolean>() {
1156 @Override
1157 public Boolean call() throws Exception {
1158 List<HRegion> regions = util.getMiniHBaseCluster().getRegions(TABLE_NAME);
1159 for (HRegion region : regions) {
1160 for (Store store : region.getStores()) {
1161 store.closeAndArchiveCompactedFiles();
1162 }
1163 }
1164 return fs.listStatus(storePath).length == 1;
1165 }
1166 }, 5000);
1167
1168 } finally {
1169 util.shutdownMiniCluster();
1170 }
1171 }
1172
1173 @Ignore("Goes zombie too frequently; needs work. See HBASE-14563") @Test
1174 public void testExcludeMinorCompaction() throws Exception {
1175 Configuration conf = util.getConfiguration();
1176 conf.setInt("hbase.hstore.compaction.min", 2);
1177 generateRandomStartKeys(5);
1178 util.setJobWithoutMRCluster();
1179 util.startMiniCluster();
1180 try (Connection conn = ConnectionFactory.createConnection(conf);
1181 Admin admin = conn.getAdmin()){
1182 Path testDir = util.getDataTestDirOnTestFS("testExcludeMinorCompaction");
1183 final FileSystem fs = util.getDFSCluster().getFileSystem();
1184 Table table = util.createTable(TABLE_NAME, FAMILIES);
1185 assertEquals("Should start with empty table", 0, util.countRows(table));
1186
1187
1188 final Path storePath = new Path(
1189 FSUtils.getTableDir(FSUtils.getRootDir(conf), TABLE_NAME),
1190 new Path(admin.getTableRegions(TABLE_NAME).get(0).getEncodedName(),
1191 Bytes.toString(FAMILIES[0])));
1192 assertEquals(0, fs.listStatus(storePath).length);
1193
1194
1195 Put p = new Put(Bytes.toBytes("test"));
1196 p.add(FAMILIES[0], Bytes.toBytes("1"), Bytes.toBytes("1"));
1197 table.put(p);
1198 admin.flush(TABLE_NAME);
1199 assertEquals(1, util.countRows(table));
1200 quickPoll(new Callable<Boolean>() {
1201 @Override
1202 public Boolean call() throws Exception {
1203 return fs.listStatus(storePath).length == 1;
1204 }
1205 }, 5000);
1206
1207
1208 conf.setBoolean("hbase.mapreduce.hfileoutputformat.compaction.exclude",
1209 true);
1210
1211 RegionLocator regionLocator = conn.getRegionLocator(TABLE_NAME);
1212 runIncrementalPELoad(conf, table.getTableDescriptor(), regionLocator, testDir, false);
1213
1214
1215 new LoadIncrementalHFiles(conf).doBulkLoad(testDir, admin, table, regionLocator);
1216
1217
1218 int expectedRows = NMapInputFormat.getNumMapTasks(conf) * ROWSPERSPLIT;
1219 assertEquals("LoadIncrementalHFiles should put expected data in table",
1220 expectedRows + 1, util.countRows(table));
1221
1222
1223 assertEquals(2, fs.listStatus(storePath).length);
1224
1225
1226 admin.compact(TABLE_NAME);
1227 try {
1228 quickPoll(new Callable<Boolean>() {
1229 @Override
1230 public Boolean call() throws Exception {
1231 return fs.listStatus(storePath).length == 1;
1232 }
1233 }, 5000);
1234 throw new IOException("SF# = " + fs.listStatus(storePath).length);
1235 } catch (AssertionError ae) {
1236
1237 }
1238
1239
1240 admin.majorCompact(TABLE_NAME);
1241 quickPoll(new Callable<Boolean>() {
1242 @Override
1243 public Boolean call() throws Exception {
1244 return fs.listStatus(storePath).length == 1;
1245 }
1246 }, 5000);
1247
1248 } finally {
1249 util.shutdownMiniCluster();
1250 }
1251 }
1252
1253 private void quickPoll(Callable<Boolean> c, int waitMs) throws Exception {
1254 int sleepMs = 10;
1255 int retries = (int) Math.ceil(((double) waitMs) / sleepMs);
1256 while (retries-- > 0) {
1257 if (c.call().booleanValue()) {
1258 return;
1259 }
1260 Thread.sleep(sleepMs);
1261 }
1262 fail();
1263 }
1264
1265 public static void main(String args[]) throws Exception {
1266 new TestHFileOutputFormat2().manualTest(args);
1267 }
1268
1269 public void manualTest(String args[]) throws Exception {
1270 Configuration conf = HBaseConfiguration.create();
1271 util = new HBaseTestingUtility(conf);
1272 if ("newtable".equals(args[0])) {
1273 TableName tname = TableName.valueOf(args[1]);
1274 byte[][] splitKeys = generateRandomSplitKeys(4);
1275 try (HTable table = util.createTable(tname, FAMILIES, splitKeys)) {
1276 }
1277 } else if ("incremental".equals(args[0])) {
1278 TableName tname = TableName.valueOf(args[1]);
1279 try(Connection c = ConnectionFactory.createConnection(conf);
1280 Admin admin = c.getAdmin();
1281 RegionLocator regionLocator = c.getRegionLocator(tname)) {
1282 Path outDir = new Path("incremental-out");
1283 runIncrementalPELoad(conf, admin.getTableDescriptor(tname), regionLocator, outDir, false);
1284 }
1285 } else {
1286 throw new RuntimeException(
1287 "usage: TestHFileOutputFormat2 newtable | incremental");
1288 }
1289 }
1290
1291 }
1292