1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.test;
20
21 import com.google.common.base.Preconditions;
22 import com.google.common.collect.Sets;
23 import java.io.DataInput;
24 import java.io.DataOutput;
25 import java.io.FileNotFoundException;
26 import java.io.IOException;
27 import java.io.InterruptedIOException;
28 import java.security.SecureRandom;
29 import java.util.ArrayList;
30 import java.util.Arrays;
31 import java.util.Iterator;
32 import java.util.List;
33 import java.util.Random;
34 import java.util.Set;
35 import java.util.SortedSet;
36 import java.util.TreeSet;
37 import java.util.UUID;
38 import java.util.concurrent.ThreadLocalRandom;
39 import java.util.concurrent.atomic.AtomicInteger;
40 import org.apache.commons.cli.CommandLine;
41 import org.apache.commons.cli.GnuParser;
42 import org.apache.commons.cli.HelpFormatter;
43 import org.apache.commons.cli.Options;
44 import org.apache.commons.cli.ParseException;
45 import org.apache.commons.logging.Log;
46 import org.apache.commons.logging.LogFactory;
47 import org.apache.hadoop.conf.Configuration;
48 import org.apache.hadoop.conf.Configured;
49 import org.apache.hadoop.fs.FileSystem;
50 import org.apache.hadoop.fs.LocatedFileStatus;
51 import org.apache.hadoop.fs.Path;
52 import org.apache.hadoop.fs.RemoteIterator;
53 import org.apache.hadoop.hbase.Cell;
54 import org.apache.hadoop.hbase.HBaseConfiguration;
55 import org.apache.hadoop.hbase.HBaseTestingUtility;
56 import org.apache.hadoop.hbase.HColumnDescriptor;
57 import org.apache.hadoop.hbase.HConstants;
58 import org.apache.hadoop.hbase.HRegionLocation;
59 import org.apache.hadoop.hbase.HTableDescriptor;
60 import org.apache.hadoop.hbase.IntegrationTestBase;
61 import org.apache.hadoop.hbase.IntegrationTestingUtility;
62 import org.apache.hadoop.hbase.MasterNotRunningException;
63 import org.apache.hadoop.hbase.TableName;
64 import org.apache.hadoop.hbase.client.Admin;
65 import org.apache.hadoop.hbase.client.BufferedMutator;
66 import org.apache.hadoop.hbase.client.BufferedMutatorParams;
67 import org.apache.hadoop.hbase.client.Connection;
68 import org.apache.hadoop.hbase.client.ConnectionConfiguration;
69 import org.apache.hadoop.hbase.client.ConnectionFactory;
70 import org.apache.hadoop.hbase.client.Get;
71 import org.apache.hadoop.hbase.client.HTable;
72 import org.apache.hadoop.hbase.client.Mutation;
73 import org.apache.hadoop.hbase.client.Put;
74 import org.apache.hadoop.hbase.client.RegionLocator;
75 import org.apache.hadoop.hbase.client.Result;
76 import org.apache.hadoop.hbase.client.ResultScanner;
77 import org.apache.hadoop.hbase.client.Scan;
78 import org.apache.hadoop.hbase.client.ScannerCallable;
79 import org.apache.hadoop.hbase.client.Table;
80 import org.apache.hadoop.hbase.fs.HFileSystem;
81 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
82 import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
83 import org.apache.hadoop.hbase.mapreduce.TableMapper;
84 import org.apache.hadoop.hbase.mapreduce.TableRecordReaderImpl;
85 import org.apache.hadoop.hbase.mapreduce.WALPlayer;
86 import org.apache.hadoop.hbase.regionserver.FlushLargeStoresPolicy;
87 import org.apache.hadoop.hbase.regionserver.FlushPolicyFactory;
88 import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
89 import org.apache.hadoop.hbase.testclassification.IntegrationTests;
90 import org.apache.hadoop.hbase.util.AbstractHBaseTool;
91 import org.apache.hadoop.hbase.util.Bytes;
92 import org.apache.hadoop.hbase.util.Random64;
93 import org.apache.hadoop.hbase.util.RegionSplitter;
94 import org.apache.hadoop.hbase.wal.WALKey;
95 import org.apache.hadoop.io.BytesWritable;
96 import org.apache.hadoop.io.NullWritable;
97 import org.apache.hadoop.io.Writable;
98 import org.apache.hadoop.mapreduce.Counter;
99 import org.apache.hadoop.mapreduce.CounterGroup;
100 import org.apache.hadoop.mapreduce.Counters;
101 import org.apache.hadoop.mapreduce.InputFormat;
102 import org.apache.hadoop.mapreduce.InputSplit;
103 import org.apache.hadoop.mapreduce.Job;
104 import org.apache.hadoop.mapreduce.JobContext;
105 import org.apache.hadoop.mapreduce.Mapper;
106 import org.apache.hadoop.mapreduce.RecordReader;
107 import org.apache.hadoop.mapreduce.Reducer;
108 import org.apache.hadoop.mapreduce.TaskAttemptContext;
109 import org.apache.hadoop.mapreduce.TaskAttemptID;
110 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
111 import org.apache.hadoop.mapreduce.lib.input.FileSplit;
112 import org.apache.hadoop.mapreduce.lib.input.SequenceFileAsBinaryInputFormat;
113 import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
114 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
115 import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
116 import org.apache.hadoop.mapreduce.lib.output.SequenceFileAsBinaryOutputFormat;
117 import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
118 import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
119 import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
120 import org.apache.hadoop.util.Tool;
121 import org.apache.hadoop.util.ToolRunner;
122 import org.junit.Test;
123 import org.junit.experimental.categories.Category;
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194 @Category(IntegrationTests.class)
195 public class IntegrationTestBigLinkedList extends IntegrationTestBase {
196 protected static final byte[] NO_KEY = new byte[1];
197
198 protected static String TABLE_NAME_KEY = "IntegrationTestBigLinkedList.table";
199
200 protected static String DEFAULT_TABLE_NAME = "IntegrationTestBigLinkedList";
201
202 protected static byte[] FAMILY_NAME = Bytes.toBytes("meta");
203 private static byte[] BIG_FAMILY_NAME = Bytes.toBytes("big");
204 private static byte[] TINY_FAMILY_NAME = Bytes.toBytes("tiny");
205
206
207 protected static final byte[] COLUMN_PREV = Bytes.toBytes("prev");
208
209
210 protected static final byte[] COLUMN_CLIENT = Bytes.toBytes("client");
211
212
213 protected static final byte[] COLUMN_COUNT = Bytes.toBytes("count");
214
215
216 private static final String GENERATOR_NUM_ROWS_PER_MAP_KEY
217 = "IntegrationTestBigLinkedList.generator.num_rows";
218
219 private static final String GENERATOR_NUM_MAPPERS_KEY
220 = "IntegrationTestBigLinkedList.generator.map.tasks";
221
222 private static final String GENERATOR_WIDTH_KEY
223 = "IntegrationTestBigLinkedList.generator.width";
224
225 private static final String GENERATOR_WRAP_KEY
226 = "IntegrationTestBigLinkedList.generator.wrap";
227
228 private static final String CONCURRENT_WALKER_KEY
229 = "IntegrationTestBigLinkedList.generator.concurrentwalkers";
230
231 protected int NUM_SLAVES_BASE = 3;
232
233 private static final int MISSING_ROWS_TO_LOG = 10;
234
235 private static final int WIDTH_DEFAULT = 1000000;
236 private static final int WRAP_DEFAULT = 25;
237 private static final int ROWKEY_LENGTH = 16;
238
239 private static final int CONCURRENT_WALKER_DEFAULT = 0;
240
241 protected String toRun;
242 protected String[] otherArgs;
243
244 static class CINode {
245 byte[] key;
246 byte[] prev;
247 String client;
248 long count;
249 }
250
251
252
253
254 static class Generator extends Configured implements Tool {
255
256 private static final Log LOG = LogFactory.getLog(Generator.class);
257
258
259
260
261
262
263
264
265
266
267
268
269 public static final String MULTIPLE_UNEVEN_COLUMNFAMILIES_KEY =
270 "generator.multiple.columnfamilies";
271
272
273
274
275
276
277
278 public static final String BIG_FAMILY_VALUE_SIZE_KEY = "generator.big.family.value.size";
279
280
281 public static enum Counts {
282 SUCCESS, TERMINATING, UNDEFINED, IOEXCEPTION
283 }
284
285 public static final String USAGE = "Usage : " + Generator.class.getSimpleName() +
286 " <num mappers> <num nodes per map> <tmp output dir> [<width> <wrap multiplier>" +
287 " <num walker threads>] \n" +
288 "where <num nodes per map> should be a multiple of width*wrap multiplier, 25M by default \n" +
289 "walkers will verify random flushed loop during Generation.";
290
291 public Job job;
292
293 static class GeneratorInputFormat extends InputFormat<BytesWritable,NullWritable> {
294 static class GeneratorInputSplit extends InputSplit implements Writable {
295 @Override
296 public long getLength() throws IOException, InterruptedException {
297 return 1;
298 }
299 @Override
300 public String[] getLocations() throws IOException, InterruptedException {
301 return new String[0];
302 }
303 @Override
304 public void readFields(DataInput arg0) throws IOException {
305 }
306 @Override
307 public void write(DataOutput arg0) throws IOException {
308 }
309 }
310
311 static class GeneratorRecordReader extends RecordReader<BytesWritable,NullWritable> {
312 private long count;
313 private long numNodes;
314 private Random64 rand;
315
316 @Override
317 public void close() throws IOException {
318 }
319
320 @Override
321 public BytesWritable getCurrentKey() throws IOException, InterruptedException {
322 byte[] bytes = new byte[ROWKEY_LENGTH];
323 rand.nextBytes(bytes);
324 return new BytesWritable(bytes);
325 }
326
327 @Override
328 public NullWritable getCurrentValue() throws IOException, InterruptedException {
329 return NullWritable.get();
330 }
331
332 @Override
333 public float getProgress() throws IOException, InterruptedException {
334 return (float)(count / (double)numNodes);
335 }
336
337 @Override
338 public void initialize(InputSplit arg0, TaskAttemptContext context)
339 throws IOException, InterruptedException {
340 numNodes = context.getConfiguration().getLong(GENERATOR_NUM_ROWS_PER_MAP_KEY, 25000000);
341
342 rand = new Random64();
343 }
344
345 @Override
346 public boolean nextKeyValue() throws IOException, InterruptedException {
347 return count++ < numNodes;
348 }
349
350 }
351
352 @Override
353 public RecordReader<BytesWritable,NullWritable> createRecordReader(
354 InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
355 GeneratorRecordReader rr = new GeneratorRecordReader();
356 rr.initialize(split, context);
357 return rr;
358 }
359
360 @Override
361 public List<InputSplit> getSplits(JobContext job) throws IOException, InterruptedException {
362 int numMappers = job.getConfiguration().getInt(GENERATOR_NUM_MAPPERS_KEY, 1);
363
364 ArrayList<InputSplit> splits = new ArrayList<InputSplit>(numMappers);
365
366 for (int i = 0; i < numMappers; i++) {
367 splits.add(new GeneratorInputSplit());
368 }
369
370 return splits;
371 }
372 }
373
374
375 static class OneFilePerMapperSFIF<K, V> extends SequenceFileInputFormat<K, V> {
376 @Override
377 protected boolean isSplitable(JobContext context, Path filename) {
378 return false;
379 }
380 }
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409 static class GeneratorMapper
410 extends Mapper<BytesWritable, NullWritable, NullWritable, NullWritable> {
411
412 byte[][] first = null;
413 byte[][] prev = null;
414 byte[][] current = null;
415 byte[] id;
416 long count = 0;
417 int i;
418 BufferedMutator mutator;
419 Connection connection;
420 long numNodes;
421 long wrap;
422 int width;
423 boolean multipleUnevenColumnFamilies;
424 byte[] tinyValue = new byte[] { 't' };
425 byte[] bigValue = null;
426 Configuration conf;
427
428 volatile boolean walkersStop;
429 int numWalkers;
430 volatile List<Long> flushedLoops = new ArrayList<>();
431 List<Thread> walkers = new ArrayList<>();
432
433 @Override
434 protected void setup(Context context) throws IOException, InterruptedException {
435 id = Bytes.toBytes("Job: "+context.getJobID() + " Task: " + context.getTaskAttemptID());
436 Configuration conf = context.getConfiguration();
437 connection = ConnectionFactory.createConnection(conf);
438 instantiateHTable();
439 this.width = context.getConfiguration().getInt(GENERATOR_WIDTH_KEY, WIDTH_DEFAULT);
440 current = new byte[this.width][];
441 int wrapMultiplier = context.getConfiguration().getInt(GENERATOR_WRAP_KEY, WRAP_DEFAULT);
442 this.wrap = (long)wrapMultiplier * width;
443 this.numNodes = context.getConfiguration().getLong(
444 GENERATOR_NUM_ROWS_PER_MAP_KEY, (long)WIDTH_DEFAULT * WRAP_DEFAULT);
445 if (this.numNodes < this.wrap) {
446 this.wrap = this.numNodes;
447 }
448 this.multipleUnevenColumnFamilies = isMultiUnevenColumnFamilies(context.getConfiguration());
449 this.numWalkers = context.getConfiguration().getInt(CONCURRENT_WALKER_KEY, CONCURRENT_WALKER_DEFAULT);
450 this.walkersStop = false;
451 this.conf = context.getConfiguration();
452
453 if (multipleUnevenColumnFamilies) {
454 int n = context.getConfiguration().getInt(BIG_FAMILY_VALUE_SIZE_KEY, 256);
455 int limit = context.getConfiguration().getInt(
456 ConnectionConfiguration.MAX_KEYVALUE_SIZE_KEY,
457 ConnectionConfiguration.MAX_KEYVALUE_SIZE_DEFAULT);
458
459 Preconditions.checkArgument(
460 n <= limit,
461 "%s(%s) > %s(%s)",
462 BIG_FAMILY_VALUE_SIZE_KEY, n, ConnectionConfiguration.MAX_KEYVALUE_SIZE_KEY, limit);
463
464 bigValue = new byte[n];
465 ThreadLocalRandom.current().nextBytes(bigValue);
466 LOG.info("Create a bigValue with " + n + " bytes.");
467 }
468
469 Preconditions.checkArgument(
470 numNodes > 0,
471 "numNodes(%s) <= 0",
472 numNodes);
473 Preconditions.checkArgument(
474 numNodes % width == 0,
475 "numNodes(%s) mod width(%s) != 0",
476 numNodes, width);
477 Preconditions.checkArgument(
478 numNodes % wrap == 0,
479 "numNodes(%s) mod wrap(%s) != 0",
480 numNodes, wrap
481 );
482 }
483
484 protected void instantiateHTable() throws IOException {
485 mutator = connection.getBufferedMutator(
486 new BufferedMutatorParams(getTableName(connection.getConfiguration()))
487 .writeBufferSize(4 * 1024 * 1024));
488 }
489
490 @Override
491 protected void cleanup(Context context) throws IOException ,InterruptedException {
492 joinWalkers();
493 mutator.close();
494 connection.close();
495 }
496
497 @Override
498 protected void map(BytesWritable key, NullWritable value, Context output) throws IOException {
499 current[i] = new byte[key.getLength()];
500 System.arraycopy(key.getBytes(), 0, current[i], 0, key.getLength());
501 if (++i == current.length) {
502 LOG.debug("Persisting current.length=" + current.length + ", count=" + count +
503 ", id=" + Bytes.toStringBinary(id) + ", current=" + Bytes.toStringBinary(current[0]) +
504 ", i=" + i);
505 persist(output, count, prev, current, id);
506 i = 0;
507
508 if (first == null) {
509 first = current;
510 }
511 prev = current;
512 current = new byte[this.width][];
513
514 count += current.length;
515 output.setStatus("Count " + count);
516
517 if (count % wrap == 0) {
518
519
520 circularLeftShift(first);
521 persist(output, -1, prev, first, null);
522
523
524 if (numWalkers > 0) {
525 addFlushed(key.getBytes());
526 if (walkers.isEmpty()) {
527 startWalkers(numWalkers, conf, output);
528 }
529 }
530 first = null;
531 prev = null;
532 }
533 }
534 }
535
536 private static <T> void circularLeftShift(T[] first) {
537 T ez = first[0];
538 System.arraycopy(first, 1, first, 0, first.length - 1);
539 first[first.length - 1] = ez;
540 }
541
542 private void addFlushed(byte[] rowKey) {
543 synchronized (flushedLoops) {
544 flushedLoops.add(Bytes.toLong(rowKey));
545 flushedLoops.notifyAll();
546 }
547 }
548
549 protected void persist(Context output, long count, byte[][] prev, byte[][] current, byte[] id)
550 throws IOException {
551 for (int i = 0; i < current.length; i++) {
552
553 if (i % 100 == 0) {
554
555 output.progress();
556 }
557
558 Put put = new Put(current[i]);
559 put.addColumn(FAMILY_NAME, COLUMN_PREV, prev == null ? NO_KEY : prev[i]);
560
561 if (count >= 0) {
562 put.addColumn(FAMILY_NAME, COLUMN_COUNT, Bytes.toBytes(count + i));
563 }
564 if (id != null) {
565 put.addColumn(FAMILY_NAME, COLUMN_CLIENT, id);
566 }
567
568 if (this.multipleUnevenColumnFamilies) {
569
570 put.addColumn(TINY_FAMILY_NAME, TINY_FAMILY_NAME, this.tinyValue);
571
572 put.addColumn(BIG_FAMILY_NAME, BIG_FAMILY_NAME, this.bigValue);
573 }
574 mutator.mutate(put);
575 }
576
577 mutator.flush();
578 }
579
580 private void startWalkers(int numWalkers, Configuration conf, Context context) {
581 LOG.info("Starting " + numWalkers + " concurrent walkers");
582 for (int i = 0; i < numWalkers; i++) {
583 Thread walker = new Thread(new ContinuousConcurrentWalker(conf, context));
584 walker.start();
585 walkers.add(walker);
586 }
587 }
588
589 private void joinWalkers() {
590 walkersStop = true;
591 synchronized (flushedLoops) {
592 flushedLoops.notifyAll();
593 }
594 for (Thread walker : walkers) {
595 try {
596 walker.join();
597 } catch (InterruptedException e) {
598
599 }
600 }
601 }
602
603
604
605
606
607
608
609 public class ContinuousConcurrentWalker implements Runnable {
610
611 ConcurrentWalker walker;
612 Configuration conf;
613 Context context;
614 Random rand;
615
616 public ContinuousConcurrentWalker(Configuration conf, Context context) {
617 this.conf = conf;
618 this.context = context;
619 rand = new Random();
620 }
621
622 @Override
623 public void run() {
624 while (!walkersStop) {
625 try {
626 long node = selectLoop();
627 try {
628 walkLoop(node);
629 } catch (IOException e) {
630 context.getCounter(Counts.IOEXCEPTION).increment(1l);
631 return;
632 }
633 } catch (InterruptedException e) {
634 return;
635 }
636 }
637 }
638
639 private void walkLoop(long node) throws IOException {
640 walker = new ConcurrentWalker(context);
641 walker.setConf(conf);
642 walker.run(node, wrap);
643 }
644
645 private long selectLoop () throws InterruptedException{
646 synchronized (flushedLoops) {
647 while (flushedLoops.isEmpty() && !walkersStop) {
648 flushedLoops.wait();
649 }
650 if (walkersStop) {
651 throw new InterruptedException();
652 }
653 return flushedLoops.get(rand.nextInt(flushedLoops.size()));
654 }
655 }
656 }
657
658 public static class ConcurrentWalker extends WalkerBase {
659
660 Context context;
661
662 public ConcurrentWalker(Context context) {this.context = context;}
663
664 public void run(long startKeyIn, long maxQueriesIn) throws IOException {
665
666 long maxQueries = maxQueriesIn > 0 ? maxQueriesIn : Long.MAX_VALUE;
667 byte[] startKey = Bytes.toBytes(startKeyIn);
668
669 Connection connection = ConnectionFactory.createConnection(getConf());
670 Table table = connection.getTable(getTableName(getConf()));
671 long numQueries = 0;
672
673
674
675
676 CINode node = findStartNode(table, startKey);
677 if (node == null) {
678 LOG.error("Start node not found: " + Bytes.toStringBinary(startKey));
679 throw new IOException("Start node not found: " + startKeyIn);
680 }
681 while (numQueries < maxQueries) {
682 numQueries++;
683 byte[] prev = node.prev;
684 long t1 = System.currentTimeMillis();
685 node = getNode(prev, table, node);
686 long t2 = System.currentTimeMillis();
687 if (node == null) {
688 LOG.error("ConcurrentWalker found UNDEFINED NODE: " + Bytes.toStringBinary(prev));
689 context.getCounter(Counts.UNDEFINED).increment(1l);
690 } else if (node.prev.length == NO_KEY.length) {
691 LOG.error("ConcurrentWalker found TERMINATING NODE: " +
692 Bytes.toStringBinary(node.key));
693 context.getCounter(Counts.TERMINATING).increment(1l);
694 } else {
695
696 context.getCounter(Counts.SUCCESS).increment(1l);
697 }
698 }
699 table.close();
700 connection.close();
701 }
702 }
703 }
704
705 @Override
706 public int run(String[] args) throws Exception {
707 if (args.length < 3) {
708 System.err.println(USAGE);
709 return 1;
710 }
711 try {
712 int numMappers = Integer.parseInt(args[0]);
713 long numNodes = Long.parseLong(args[1]);
714 Path tmpOutput = new Path(args[2]);
715 Integer width = (args.length < 4) ? null : Integer.parseInt(args[3]);
716 Integer wrapMultiplier = (args.length < 5) ? null : Integer.parseInt(args[4]);
717 Integer numWalkers = (args.length < 6) ? null : Integer.parseInt(args[5]);
718 return run(numMappers, numNodes, tmpOutput, width, wrapMultiplier, numWalkers);
719 } catch (NumberFormatException e) {
720 System.err.println("Parsing generator arguments failed: " + e.getMessage());
721 System.err.println(USAGE);
722 return 1;
723 }
724 }
725
726 protected void createSchema() throws IOException {
727 Configuration conf = getConf();
728 TableName tableName = getTableName(conf);
729 try (Connection conn = ConnectionFactory.createConnection(conf);
730 Admin admin = conn.getAdmin()) {
731 if (!admin.tableExists(tableName)) {
732 HTableDescriptor htd = new HTableDescriptor(getTableName(getConf()));
733 htd.addFamily(new HColumnDescriptor(FAMILY_NAME));
734
735 htd.addFamily(new HColumnDescriptor(BIG_FAMILY_NAME));
736 htd.addFamily(new HColumnDescriptor(TINY_FAMILY_NAME));
737
738
739 if (conf.getBoolean(HBaseTestingUtility.PRESPLIT_TEST_TABLE_KEY,
740 HBaseTestingUtility.PRESPLIT_TEST_TABLE)) {
741 int numberOfServers = admin.getClusterStatus().getServers().size();
742 if (numberOfServers == 0) {
743 throw new IllegalStateException("No live regionservers");
744 }
745 int regionsPerServer = conf.getInt(HBaseTestingUtility.REGIONS_PER_SERVER_KEY,
746 HBaseTestingUtility.DEFAULT_REGIONS_PER_SERVER);
747 int totalNumberOfRegions = numberOfServers * regionsPerServer;
748 LOG.info("Number of live regionservers: " + numberOfServers + ", " +
749 "pre-splitting table into " + totalNumberOfRegions + " regions " +
750 "(default regions per server: " + regionsPerServer + ")");
751
752
753 byte[][] splits = new RegionSplitter.UniformSplit().split(totalNumberOfRegions);
754
755 admin.createTable(htd, splits);
756 } else {
757
758
759
760 admin.createTable(htd);
761 }
762 }
763 } catch (MasterNotRunningException e) {
764 LOG.error("Master not running", e);
765 throw new IOException(e);
766 }
767 }
768
769 public int runRandomInputGenerator(int numMappers, long numNodes, Path tmpOutput,
770 Integer width, Integer wrapMultiplier, Integer numWalkers)
771 throws Exception {
772 LOG.info("Running RandomInputGenerator with numMappers=" + numMappers
773 + ", numNodes=" + numNodes);
774 Job job = Job.getInstance(getConf());
775
776 job.setJobName("Random Input Generator");
777 job.setNumReduceTasks(0);
778 job.setJarByClass(getClass());
779
780 job.setInputFormatClass(GeneratorInputFormat.class);
781 job.setOutputKeyClass(BytesWritable.class);
782 job.setOutputValueClass(NullWritable.class);
783
784 setJobConf(job, numMappers, numNodes, width, wrapMultiplier, numWalkers);
785
786 job.setMapperClass(Mapper.class);
787
788 FileOutputFormat.setOutputPath(job, tmpOutput);
789 job.setOutputFormatClass(SequenceFileOutputFormat.class);
790 TableMapReduceUtil.addDependencyJarsForClasses(job.getConfiguration(), Random64.class);
791
792 boolean success = jobCompletion(job);
793
794 return success ? 0 : 1;
795 }
796
797 public int runGenerator(int numMappers, long numNodes, Path tmpOutput,
798 Integer width, Integer wrapMultiplier, Integer numWalkers)
799 throws Exception {
800 LOG.info("Running Generator with numMappers=" + numMappers +", numNodes=" + numNodes);
801 createSchema();
802 job = Job.getInstance(getConf());
803
804 job.setJobName("Link Generator");
805 job.setNumReduceTasks(0);
806 job.setJarByClass(getClass());
807
808 FileInputFormat.setInputPaths(job, tmpOutput);
809 job.setInputFormatClass(OneFilePerMapperSFIF.class);
810 job.setOutputKeyClass(NullWritable.class);
811 job.setOutputValueClass(NullWritable.class);
812
813 setJobConf(job, numMappers, numNodes, width, wrapMultiplier, numWalkers);
814
815 setMapperForGenerator(job);
816
817 job.setOutputFormatClass(NullOutputFormat.class);
818
819 job.getConfiguration().setBoolean("mapreduce.map.speculative", false);
820 TableMapReduceUtil.addDependencyJars(job);
821 TableMapReduceUtil.addDependencyJarsForClasses(job.getConfiguration(),
822 AbstractHBaseTool.class);
823 TableMapReduceUtil.initCredentials(job);
824
825 boolean success = jobCompletion(job);
826
827 return success ? 0 : 1;
828 }
829
830 protected boolean jobCompletion(Job job) throws IOException, InterruptedException,
831 ClassNotFoundException {
832 boolean success = job.waitForCompletion(true);
833 return success;
834 }
835
836 protected void setMapperForGenerator(Job job) {
837 job.setMapperClass(GeneratorMapper.class);
838 }
839
840 public int run(int numMappers, long numNodes, Path tmpOutput,
841 Integer width, Integer wrapMultiplier, Integer numWalkers)
842 throws Exception {
843 int ret = runRandomInputGenerator(numMappers, numNodes, tmpOutput, width, wrapMultiplier,
844 numWalkers);
845 if (ret > 0) {
846 return ret;
847 }
848 return runGenerator(numMappers, numNodes, tmpOutput, width, wrapMultiplier, numWalkers);
849 }
850
851 public boolean verify() {
852 try {
853 Counters counters = job.getCounters();
854 if (counters == null) {
855 LOG.info("Counters object was null, Generator verification cannot be performed."
856 + " This is commonly a result of insufficient YARN configuration.");
857 return false;
858 }
859
860 if (counters.findCounter(Counts.TERMINATING).getValue() > 0 ||
861 counters.findCounter(Counts.UNDEFINED).getValue() > 0 ||
862 counters.findCounter(Counts.IOEXCEPTION).getValue() > 0) {
863 LOG.error("Concurrent walker failed to verify during Generation phase");
864 LOG.error("TERMINATING nodes: " + counters.findCounter(Counts.TERMINATING).getValue());
865 LOG.error("UNDEFINED nodes: " + counters.findCounter(Counts.UNDEFINED).getValue());
866 LOG.error("IOEXCEPTION nodes: " + counters.findCounter(Counts.IOEXCEPTION).getValue());
867 return false;
868 }
869 } catch (IOException e) {
870 LOG.info("Generator verification could not find counter");
871 return false;
872 }
873 return true;
874 }
875 }
876
877
878
879
880
881
882
883 static class Search extends Configured implements Tool {
884 private static final Log LOG = LogFactory.getLog(Search.class);
885 protected Job job;
886
887 private static void printUsage(final String error) {
888 if (error != null && error.length() > 0) System.out.println("ERROR: " + error);
889 System.err.println("Usage: search <KEYS_DIR> [<MAPPERS_COUNT>]");
890 }
891
892 @Override
893 public int run(String[] args) throws Exception {
894 if (args.length < 1 || args.length > 2) {
895 printUsage(null);
896 return 1;
897 }
898 Path inputDir = new Path(args[0]);
899 int numMappers = 1;
900 if (args.length > 1) {
901 numMappers = Integer.parseInt(args[1]);
902 }
903 return run(inputDir, numMappers);
904 }
905
906
907
908
909 public static class WALSearcher extends WALPlayer {
910 public WALSearcher(Configuration conf) {
911 super(conf);
912 }
913
914
915
916
917 public static class WALMapperSearcher extends WALMapper {
918 private SortedSet<byte []> keysToFind;
919 private AtomicInteger rows = new AtomicInteger(0);
920
921 @Override
922 public void setup(Mapper<WALKey, WALEdit, ImmutableBytesWritable, Mutation>.Context context)
923 throws IOException {
924 super.setup(context);
925 try {
926 this.keysToFind = readKeysToSearch(context.getConfiguration());
927 LOG.info("Loaded keys to find: count=" + this.keysToFind.size());
928 } catch (InterruptedException e) {
929 throw new InterruptedIOException(e.toString());
930 }
931 }
932
933 @Override
934 protected boolean filter(Context context, Cell cell) {
935
936 byte [] row = new byte [cell.getRowLength()];
937 System.arraycopy(cell.getRowArray(), cell.getRowOffset(), row, 0, cell.getRowLength());
938 boolean b = this.keysToFind.contains(row);
939 if (b) {
940 String keyStr = Bytes.toStringBinary(row);
941 try {
942 LOG.info("Found cell=" + cell + " , walKey=" + context.getCurrentKey());
943 } catch (IOException|InterruptedException e) {
944 LOG.warn(e);
945 }
946 if (rows.addAndGet(1) < MISSING_ROWS_TO_LOG) {
947 context.getCounter(FOUND_GROUP_KEY, keyStr).increment(1);
948 }
949 context.getCounter(FOUND_GROUP_KEY, "CELL_WITH_MISSING_ROW").increment(1);
950 }
951 return b;
952 }
953 }
954
955
956 @Override
957 public Job createSubmittableJob(String[] args) throws IOException {
958 Job job = super.createSubmittableJob(args);
959
960 job.setJarByClass(WALMapperSearcher.class);
961 job.setMapperClass(WALMapperSearcher.class);
962 job.setOutputFormatClass(NullOutputFormat.class);
963 return job;
964 }
965 }
966
967 static final String FOUND_GROUP_KEY = "Found";
968 static final String SEARCHER_INPUTDIR_KEY = "searcher.keys.inputdir";
969
970 public int run(Path inputDir, int numMappers) throws Exception {
971 getConf().set(SEARCHER_INPUTDIR_KEY, inputDir.toString());
972 SortedSet<byte []> keys = readKeysToSearch(getConf());
973 if (keys.isEmpty()) throw new RuntimeException("No keys to find");
974 LOG.info("Count of keys to find: " + keys.size());
975 for(byte [] key: keys) LOG.info("Key: " + Bytes.toStringBinary(key));
976 Path hbaseDir = new Path(getConf().get(HConstants.HBASE_DIR));
977
978 Path walsDir = new Path(hbaseDir, HConstants.HREGION_LOGDIR_NAME);
979 Path oldWalsDir = new Path(hbaseDir, HConstants.HREGION_OLDLOGDIR_NAME);
980 LOG.info("Running Search with keys inputDir=" + inputDir +", numMappers=" + numMappers +
981 " against " + getConf().get(HConstants.HBASE_DIR));
982 int ret = ToolRunner.run(new WALSearcher(getConf()), new String [] {walsDir.toString(), ""});
983 if (ret != 0) return ret;
984 return ToolRunner.run(new WALSearcher(getConf()), new String [] {oldWalsDir.toString(), ""});
985 }
986
987 static SortedSet<byte []> readKeysToSearch(final Configuration conf)
988 throws IOException, InterruptedException {
989 Path keysInputDir = new Path(conf.get(SEARCHER_INPUTDIR_KEY));
990 FileSystem fs = FileSystem.get(conf);
991 SortedSet<byte []> result = new TreeSet<byte []>(Bytes.BYTES_COMPARATOR);
992 if (!fs.exists(keysInputDir)) {
993 throw new FileNotFoundException(keysInputDir.toString());
994 }
995 if (!fs.isDirectory(keysInputDir)) {
996 throw new UnsupportedOperationException("TODO");
997 } else {
998 RemoteIterator<LocatedFileStatus> iterator = fs.listFiles(keysInputDir, false);
999 while(iterator.hasNext()) {
1000 LocatedFileStatus keyFileStatus = iterator.next();
1001
1002 if (keyFileStatus.getPath().getName().startsWith("_")) continue;
1003 result.addAll(readFileToSearch(conf, fs, keyFileStatus));
1004 }
1005 }
1006 return result;
1007 }
1008
1009 private static SortedSet<byte[]> readFileToSearch(final Configuration conf,
1010 final FileSystem fs, final LocatedFileStatus keyFileStatus) throws IOException,
1011 InterruptedException {
1012 SortedSet<byte []> result = new TreeSet<byte []>(Bytes.BYTES_COMPARATOR);
1013
1014
1015 TaskAttemptContext context = new TaskAttemptContextImpl(conf, new TaskAttemptID());
1016 try (SequenceFileAsBinaryInputFormat.SequenceFileAsBinaryRecordReader rr =
1017 new SequenceFileAsBinaryInputFormat.SequenceFileAsBinaryRecordReader()) {
1018 InputSplit is =
1019 new FileSplit(keyFileStatus.getPath(), 0, keyFileStatus.getLen(), new String [] {});
1020 rr.initialize(is, context);
1021 while (rr.nextKeyValue()) {
1022 rr.getCurrentKey();
1023 BytesWritable bw = rr.getCurrentValue();
1024 if (Verify.VerifyReducer.whichType(bw.getBytes()) == Verify.Counts.UNDEFINED) {
1025 byte[] key = new byte[rr.getCurrentKey().getLength()];
1026 System.arraycopy(rr.getCurrentKey().getBytes(), 0, key, 0, rr.getCurrentKey()
1027 .getLength());
1028 result.add(key);
1029 }
1030 }
1031 }
1032 return result;
1033 }
1034 }
1035
1036
1037
1038
1039
1040 static class Verify extends Configured implements Tool {
1041
1042 private static final Log LOG = LogFactory.getLog(Verify.class);
1043 protected static final BytesWritable DEF = new BytesWritable(new byte[] { 0 });
1044 protected static final BytesWritable DEF_LOST_FAMILIES = new BytesWritable(new byte[] { 1 });
1045
1046 protected Job job;
1047
1048 public static class VerifyMapper extends TableMapper<BytesWritable, BytesWritable> {
1049 private BytesWritable row = new BytesWritable();
1050 private BytesWritable ref = new BytesWritable();
1051
1052 private boolean multipleUnevenColumnFamilies;
1053
1054 @Override
1055 protected void setup(
1056 Mapper<ImmutableBytesWritable, Result, BytesWritable, BytesWritable>.Context context)
1057 throws IOException, InterruptedException {
1058 this.multipleUnevenColumnFamilies = isMultiUnevenColumnFamilies(context.getConfiguration());
1059 }
1060
1061 @Override
1062 protected void map(ImmutableBytesWritable key, Result value, Context context)
1063 throws IOException ,InterruptedException {
1064 byte[] rowKey = key.get();
1065 row.set(rowKey, 0, rowKey.length);
1066 if (multipleUnevenColumnFamilies
1067 && (!value.containsColumn(BIG_FAMILY_NAME, BIG_FAMILY_NAME) || !value.containsColumn(
1068 TINY_FAMILY_NAME, TINY_FAMILY_NAME))) {
1069 context.write(row, DEF_LOST_FAMILIES);
1070 } else {
1071 context.write(row, DEF);
1072 }
1073 byte[] prev = value.getValue(FAMILY_NAME, COLUMN_PREV);
1074 if (prev != null && prev.length > 0) {
1075 ref.set(prev, 0, prev.length);
1076 context.write(ref, row);
1077 } else {
1078 LOG.warn(String.format("Prev is not set for: %s", Bytes.toStringBinary(rowKey)));
1079 }
1080 }
1081 }
1082
1083
1084
1085
1086
1087 public static enum Counts {
1088 UNREFERENCED, UNDEFINED, REFERENCED, CORRUPT, EXTRAREFERENCES, EXTRA_UNDEF_REFERENCES,
1089 LOST_FAMILIES
1090 }
1091
1092
1093
1094
1095
1096
1097 public static class VerifyReducer extends
1098 Reducer<BytesWritable, BytesWritable, BytesWritable, BytesWritable> {
1099 private ArrayList<byte[]> refs = new ArrayList<byte[]>();
1100 private final BytesWritable UNREF = new BytesWritable(addPrefixFlag(
1101 Counts.UNREFERENCED.ordinal(), new byte[] {}));
1102 private final BytesWritable LOSTFAM = new BytesWritable(addPrefixFlag(
1103 Counts.LOST_FAMILIES.ordinal(), new byte[] {}));
1104
1105 private AtomicInteger rows = new AtomicInteger(0);
1106 private Connection connection;
1107
1108 @Override
1109 protected void setup(Reducer<BytesWritable, BytesWritable, BytesWritable, BytesWritable>.Context context)
1110 throws IOException, InterruptedException {
1111 super.setup(context);
1112 this.connection = ConnectionFactory.createConnection(context.getConfiguration());
1113 }
1114
1115 @Override
1116 protected void cleanup(
1117 Reducer<BytesWritable, BytesWritable, BytesWritable, BytesWritable>.Context context)
1118 throws IOException, InterruptedException {
1119 if (this.connection != null) {
1120 this.connection.close();
1121 }
1122 super.cleanup(context);
1123 }
1124
1125
1126
1127
1128
1129
1130
1131 public static byte[] addPrefixFlag(final int ordinal, final byte [] r) {
1132 byte[] prefix = Bytes.toBytes((short)ordinal);
1133 if (prefix.length != Bytes.SIZEOF_SHORT) {
1134 throw new RuntimeException("Unexpected size: " + prefix.length);
1135 }
1136 byte[] result = new byte[prefix.length + r.length];
1137 System.arraycopy(prefix, 0, result, 0, prefix.length);
1138 System.arraycopy(r, 0, result, prefix.length, r.length);
1139 return result;
1140 }
1141
1142
1143
1144
1145
1146
1147 public static Counts whichType(final byte [] bs) {
1148 int ordinal = Bytes.toShort(bs, 0, Bytes.SIZEOF_SHORT);
1149 return Counts.values()[ordinal];
1150 }
1151
1152
1153
1154
1155
1156 public static byte[] getRowOnly(BytesWritable bw) {
1157 byte[] bytes = new byte [bw.getLength() - Bytes.SIZEOF_SHORT];
1158 System.arraycopy(bw.getBytes(), Bytes.SIZEOF_SHORT, bytes, 0, bytes.length);
1159 return bytes;
1160 }
1161
1162 @Override
1163 public void reduce(BytesWritable key, Iterable<BytesWritable> values, Context context)
1164 throws IOException, InterruptedException {
1165 int defCount = 0;
1166 boolean lostFamilies = false;
1167 refs.clear();
1168 for (BytesWritable type : values) {
1169 if (type.getLength() == DEF.getLength()) {
1170 defCount++;
1171 if (type.getBytes()[0] == 1) {
1172 lostFamilies = true;
1173 }
1174 } else {
1175 byte[] bytes = new byte[type.getLength()];
1176 System.arraycopy(type.getBytes(), 0, bytes, 0, type.getLength());
1177 refs.add(bytes);
1178 }
1179 }
1180
1181
1182 StringBuilder refsSb = null;
1183 if (defCount == 0 || refs.size() != 1) {
1184 String keyString = Bytes.toStringBinary(key.getBytes(), 0, key.getLength());
1185 refsSb = dumpExtraInfoOnRefs(key, context, refs);
1186 LOG.error("LinkedListError: key=" + keyString + ", reference(s)=" +
1187 (refsSb != null? refsSb.toString(): ""));
1188 }
1189 if (lostFamilies) {
1190 String keyString = Bytes.toStringBinary(key.getBytes(), 0, key.getLength());
1191 LOG.error("LinkedListError: key=" + keyString + ", lost big or tiny families");
1192 context.getCounter(Counts.LOST_FAMILIES).increment(1);
1193 context.write(key, LOSTFAM);
1194 }
1195
1196 if (defCount == 0 && refs.size() > 0) {
1197
1198
1199
1200 for (int i = 0; i < refs.size(); i++) {
1201 byte[] bs = refs.get(i);
1202 int ordinal;
1203 if (i <= 0) {
1204 ordinal = Counts.UNDEFINED.ordinal();
1205 context.write(key, new BytesWritable(addPrefixFlag(ordinal, bs)));
1206 context.getCounter(Counts.UNDEFINED).increment(1);
1207 } else {
1208 ordinal = Counts.EXTRA_UNDEF_REFERENCES.ordinal();
1209 context.write(key, new BytesWritable(addPrefixFlag(ordinal, bs)));
1210 }
1211 }
1212 if (rows.addAndGet(1) < MISSING_ROWS_TO_LOG) {
1213
1214
1215
1216
1217 String keyString = Bytes.toStringBinary(key.getBytes(), 0, key.getLength());
1218 context.getCounter("undef", keyString).increment(1);
1219 }
1220 } else if (defCount > 0 && refs.size() == 0) {
1221
1222 context.write(key, UNREF);
1223 context.getCounter(Counts.UNREFERENCED).increment(1);
1224 if (rows.addAndGet(1) < MISSING_ROWS_TO_LOG) {
1225 String keyString = Bytes.toStringBinary(key.getBytes(), 0, key.getLength());
1226 context.getCounter("unref", keyString).increment(1);
1227 }
1228 } else {
1229 if (refs.size() > 1) {
1230
1231 for (int i = 1; i < refs.size(); i++) {
1232 context.write(key,
1233 new BytesWritable(addPrefixFlag(Counts.EXTRAREFERENCES.ordinal(), refs.get(i))));
1234 }
1235 context.getCounter(Counts.EXTRAREFERENCES).increment(refs.size() - 1);
1236 }
1237
1238 context.getCounter(Counts.REFERENCED).increment(1);
1239 }
1240 }
1241
1242
1243
1244
1245
1246
1247 private StringBuilder dumpExtraInfoOnRefs(final BytesWritable key, final Context context,
1248 final List<byte []> refs)
1249 throws IOException {
1250 StringBuilder refsSb = null;
1251 if (refs.isEmpty()) return refsSb;
1252 refsSb = new StringBuilder();
1253 String comma = "";
1254
1255
1256
1257 TableName tn = getTableName(context.getConfiguration());
1258 try (Table t = this.connection.getTable(tn)) {
1259 for (byte [] ref : refs) {
1260 Result r = t.get(new Get(ref));
1261 List<Cell> cells = r.listCells();
1262 String ts = (cells != null && !cells.isEmpty())?
1263 new java.util.Date(cells.get(0).getTimestamp()).toString(): "";
1264 byte [] b = r.getValue(FAMILY_NAME, COLUMN_CLIENT);
1265 String jobStr = (b != null && b.length > 0)? Bytes.toString(b): "";
1266 b = r.getValue(FAMILY_NAME, COLUMN_COUNT);
1267 long count = (b != null && b.length > 0)? Bytes.toLong(b): -1;
1268 b = r.getValue(FAMILY_NAME, COLUMN_PREV);
1269 String refRegionLocation = "";
1270 String keyRegionLocation = "";
1271 if (b != null && b.length > 0) {
1272 try (RegionLocator rl = this.connection.getRegionLocator(tn)) {
1273 HRegionLocation hrl = rl.getRegionLocation(b);
1274 if (hrl != null) refRegionLocation = hrl.toString();
1275
1276 hrl = rl.getRegionLocation(key.getBytes());
1277 if (hrl != null) keyRegionLocation = hrl.toString();
1278 }
1279 }
1280 LOG.error("Extras on ref without a def, ref=" + Bytes.toStringBinary(ref) +
1281 ", refPrevEqualsKey=" +
1282 (Bytes.compareTo(key.getBytes(), 0, key.getLength(), b, 0, b.length) == 0) +
1283 ", key=" + Bytes.toStringBinary(key.getBytes(), 0, key.getLength()) +
1284 ", ref row date=" + ts + ", jobStr=" + jobStr +
1285 ", ref row count=" + count +
1286 ", ref row regionLocation=" + refRegionLocation +
1287 ", key row regionLocation=" + keyRegionLocation);
1288 refsSb.append(comma);
1289 comma = ",";
1290 refsSb.append(Bytes.toStringBinary(ref));
1291 }
1292 }
1293 return refsSb;
1294 }
1295 }
1296
1297 @Override
1298 public int run(String[] args) throws Exception {
1299 if (args.length != 2) {
1300 System.out.println("Usage : " + Verify.class.getSimpleName()
1301 + " <output dir> <num reducers>");
1302 return 0;
1303 }
1304
1305 String outputDir = args[0];
1306 int numReducers = Integer.parseInt(args[1]);
1307
1308 return run(outputDir, numReducers);
1309 }
1310
1311 public int run(String outputDir, int numReducers) throws Exception {
1312 return run(new Path(outputDir), numReducers);
1313 }
1314
1315 public int run(Path outputDir, int numReducers) throws Exception {
1316 LOG.info("Running Verify with outputDir=" + outputDir +", numReducers=" + numReducers);
1317
1318 job = Job.getInstance(getConf());
1319
1320 job.setJobName("Link Verifier");
1321 job.setNumReduceTasks(numReducers);
1322 job.setJarByClass(getClass());
1323
1324 setJobScannerConf(job);
1325
1326 Scan scan = new Scan();
1327 scan.addColumn(FAMILY_NAME, COLUMN_PREV);
1328 scan.setCaching(10000);
1329 scan.setCacheBlocks(false);
1330 if (isMultiUnevenColumnFamilies(getConf())) {
1331 scan.addColumn(BIG_FAMILY_NAME, BIG_FAMILY_NAME);
1332 scan.addColumn(TINY_FAMILY_NAME, TINY_FAMILY_NAME);
1333 }
1334
1335 TableMapReduceUtil.initTableMapperJob(getTableName(getConf()).getName(), scan,
1336 VerifyMapper.class, BytesWritable.class, BytesWritable.class, job);
1337 TableMapReduceUtil.addDependencyJarsForClasses(job.getConfiguration(),
1338 AbstractHBaseTool.class);
1339
1340 job.getConfiguration().setBoolean("mapreduce.map.speculative", false);
1341
1342 job.setReducerClass(VerifyReducer.class);
1343 job.setOutputFormatClass(SequenceFileAsBinaryOutputFormat.class);
1344 job.setOutputKeyClass(BytesWritable.class);
1345 job.setOutputValueClass(BytesWritable.class);
1346 TextOutputFormat.setOutputPath(job, outputDir);
1347
1348 boolean success = job.waitForCompletion(true);
1349
1350 if (success) {
1351 Counters counters = job.getCounters();
1352 if (null == counters) {
1353 LOG.warn("Counters were null, cannot verify Job completion."
1354 + " This is commonly a result of insufficient YARN configuration.");
1355
1356 return 0;
1357 }
1358
1359
1360 if (verifyUnexpectedValues(counters)) {
1361
1362 return 0;
1363 }
1364 }
1365
1366
1367 return 1;
1368 }
1369
1370 public boolean verify(long expectedReferenced) throws Exception {
1371 if (job == null) {
1372 throw new IllegalStateException("You should call run() first");
1373 }
1374
1375 Counters counters = job.getCounters();
1376 if (counters == null) {
1377 LOG.info("Counters object was null, write verification cannot be performed."
1378 + " This is commonly a result of insufficient YARN configuration.");
1379 return false;
1380 }
1381
1382
1383 boolean success = verifyExpectedValues(expectedReferenced, counters);
1384
1385 if (!verifyUnexpectedValues(counters)) {
1386
1387 success = false;
1388 }
1389
1390 if (!success) {
1391 handleFailure(counters);
1392 }
1393 return success;
1394 }
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405 protected boolean verifyExpectedValues(long expectedReferenced, Counters counters) {
1406 final Counter referenced = counters.findCounter(Counts.REFERENCED);
1407 final Counter unreferenced = counters.findCounter(Counts.UNREFERENCED);
1408 boolean success = true;
1409
1410 if (expectedReferenced != referenced.getValue()) {
1411 LOG.error("Expected referenced count does not match with actual referenced count. " +
1412 "expected referenced=" + expectedReferenced + " ,actual=" + referenced.getValue());
1413 success = false;
1414 }
1415
1416 if (unreferenced.getValue() > 0) {
1417 final Counter multiref = counters.findCounter(Counts.EXTRAREFERENCES);
1418 boolean couldBeMultiRef = (multiref.getValue() == unreferenced.getValue());
1419 LOG.error("Unreferenced nodes were not expected. Unreferenced count=" + unreferenced.getValue()
1420 + (couldBeMultiRef ? "; could be due to duplicate random numbers" : ""));
1421 success = false;
1422 }
1423
1424 return success;
1425 }
1426
1427
1428
1429
1430
1431
1432
1433
1434 protected boolean verifyUnexpectedValues(Counters counters) {
1435 final Counter undefined = counters.findCounter(Counts.UNDEFINED);
1436 final Counter lostfamilies = counters.findCounter(Counts.LOST_FAMILIES);
1437 boolean success = true;
1438
1439 if (undefined.getValue() > 0) {
1440 LOG.error("Found an undefined node. Undefined count=" + undefined.getValue());
1441 success = false;
1442 }
1443
1444 if (lostfamilies.getValue() > 0) {
1445 LOG.error("Found nodes which lost big or tiny families, count=" + lostfamilies.getValue());
1446 success = false;
1447 }
1448
1449 return success;
1450 }
1451
1452 protected void handleFailure(Counters counters) throws IOException {
1453 Configuration conf = job.getConfiguration();
1454 TableName tableName = getTableName(conf);
1455 try (Connection conn = ConnectionFactory.createConnection(conf)) {
1456 try (RegionLocator rl = conn.getRegionLocator(tableName)) {
1457 CounterGroup g = counters.getGroup("undef");
1458 Iterator<Counter> it = g.iterator();
1459 while (it.hasNext()) {
1460 String keyString = it.next().getName();
1461 byte[] key = Bytes.toBytes(keyString);
1462 HRegionLocation loc = rl.getRegionLocation(key, true);
1463 LOG.error("undefined row " + keyString + ", " + loc);
1464 }
1465 g = counters.getGroup("unref");
1466 it = g.iterator();
1467 while (it.hasNext()) {
1468 String keyString = it.next().getName();
1469 byte[] key = Bytes.toBytes(keyString);
1470 HRegionLocation loc = rl.getRegionLocation(key, true);
1471 LOG.error("unreferred row " + keyString + ", " + loc);
1472 }
1473 }
1474 }
1475 }
1476 }
1477
1478
1479
1480
1481
1482 static class Loop extends Configured implements Tool {
1483
1484 private static final Log LOG = LogFactory.getLog(Loop.class);
1485 private static final String USAGE = "Usage: Loop <num iterations> <num mappers> " +
1486 "<num nodes per mapper> <output dir> <num reducers> [<width> <wrap multiplier>" +
1487 " <num walker threads>] \n" +
1488 "where <num nodes per map> should be a multiple of width*wrap multiplier, 25M by default \n" +
1489 "walkers will select and verify random flushed loop during Generation.";
1490
1491 IntegrationTestBigLinkedList it;
1492
1493 protected void runGenerator(int numMappers, long numNodes,
1494 String outputDir, Integer width, Integer wrapMultiplier, Integer numWalkers)
1495 throws Exception {
1496 Path outputPath = new Path(outputDir);
1497 UUID uuid = UUID.randomUUID();
1498 Path generatorOutput = new Path(outputPath, uuid.toString());
1499
1500 Generator generator = new Generator();
1501 generator.setConf(getConf());
1502 int retCode = generator.run(numMappers, numNodes, generatorOutput, width, wrapMultiplier,
1503 numWalkers);
1504 if (retCode > 0) {
1505 throw new RuntimeException("Generator failed with return code: " + retCode);
1506 }
1507 if (numWalkers > 0) {
1508 if (!generator.verify()) {
1509 throw new RuntimeException("Generator.verify failed");
1510 }
1511 }
1512 }
1513
1514 protected void runVerify(String outputDir,
1515 int numReducers, long expectedNumNodes) throws Exception {
1516 Path outputPath = new Path(outputDir);
1517 UUID uuid = UUID.randomUUID();
1518 Path iterationOutput = new Path(outputPath, uuid.toString());
1519
1520 Verify verify = new Verify();
1521 verify.setConf(getConf());
1522 int retCode = verify.run(iterationOutput, numReducers);
1523 if (retCode > 0) {
1524 throw new RuntimeException("Verify.run failed with return code: " + retCode);
1525 }
1526
1527 if (!verify.verify(expectedNumNodes)) {
1528 throw new RuntimeException("Verify.verify failed");
1529 }
1530 LOG.info("Verify finished with success. Total nodes=" + expectedNumNodes);
1531 }
1532
1533 @Override
1534 public int run(String[] args) throws Exception {
1535 if (args.length < 5) {
1536 System.err.println(USAGE);
1537 return 1;
1538 }
1539 try {
1540 int numIterations = Integer.parseInt(args[0]);
1541 int numMappers = Integer.parseInt(args[1]);
1542 long numNodes = Long.parseLong(args[2]);
1543 String outputDir = args[3];
1544 int numReducers = Integer.parseInt(args[4]);
1545 Integer width = (args.length < 6) ? null : Integer.parseInt(args[5]);
1546 Integer wrapMultiplier = (args.length < 7) ? null : Integer.parseInt(args[6]);
1547 Integer numWalkers = (args.length < 8) ? 0 : Integer.parseInt(args[7]);
1548
1549 long expectedNumNodes = 0;
1550
1551 if (numIterations < 0) {
1552 numIterations = Integer.MAX_VALUE;
1553 }
1554 LOG.info("Running Loop with args:" + Arrays.deepToString(args));
1555 for (int i = 0; i < numIterations; i++) {
1556 LOG.info("Starting iteration = " + i);
1557 runGenerator(numMappers, numNodes, outputDir, width, wrapMultiplier, numWalkers);
1558 expectedNumNodes += numMappers * numNodes;
1559 runVerify(outputDir, numReducers, expectedNumNodes);
1560 }
1561 return 0;
1562 } catch (NumberFormatException e) {
1563 System.err.println("Parsing loop arguments failed: " + e.getMessage());
1564 System.err.println(USAGE);
1565 return 1;
1566 }
1567 }
1568 }
1569
1570
1571
1572
1573 private static class Print extends Configured implements Tool {
1574 @Override
1575 public int run(String[] args) throws Exception {
1576 Options options = new Options();
1577 options.addOption("s", "start", true, "start key");
1578 options.addOption("e", "end", true, "end key");
1579 options.addOption("l", "limit", true, "number to print");
1580
1581 GnuParser parser = new GnuParser();
1582 CommandLine cmd = null;
1583 try {
1584 cmd = parser.parse(options, args);
1585 if (cmd.getArgs().length != 0) {
1586 throw new ParseException("Command takes no arguments");
1587 }
1588 } catch (ParseException e) {
1589 System.err.println("Failed to parse command line " + e.getMessage());
1590 System.err.println();
1591 HelpFormatter formatter = new HelpFormatter();
1592 formatter.printHelp(getClass().getSimpleName(), options);
1593 System.exit(-1);
1594 }
1595
1596 Table table = new HTable(getConf(), getTableName(getConf()));
1597
1598 Scan scan = new Scan();
1599 scan.setBatch(10000);
1600
1601 if (cmd.hasOption("s"))
1602 scan.setStartRow(Bytes.toBytesBinary(cmd.getOptionValue("s")));
1603
1604 if (cmd.hasOption("e"))
1605 scan.setStopRow(Bytes.toBytesBinary(cmd.getOptionValue("e")));
1606
1607 int limit = 0;
1608 if (cmd.hasOption("l"))
1609 limit = Integer.parseInt(cmd.getOptionValue("l"));
1610 else
1611 limit = 100;
1612
1613 ResultScanner scanner = table.getScanner(scan);
1614
1615 CINode node = new CINode();
1616 Result result = scanner.next();
1617 int count = 0;
1618 while (result != null && count++ < limit) {
1619 node = getCINode(result, node);
1620 System.out.printf("%s:%s:%012d:%s\n", Bytes.toStringBinary(node.key),
1621 Bytes.toStringBinary(node.prev), node.count, node.client);
1622 result = scanner.next();
1623 }
1624 scanner.close();
1625 table.close();
1626
1627 return 0;
1628 }
1629 }
1630
1631
1632
1633
1634 private static class Delete extends Configured implements Tool {
1635 @Override
1636 public int run(String[] args) throws Exception {
1637 if (args.length != 1) {
1638 System.out.println("Usage : " + Delete.class.getSimpleName() + " <node to delete>");
1639 return 0;
1640 }
1641 byte[] val = Bytes.toBytesBinary(args[0]);
1642
1643 org.apache.hadoop.hbase.client.Delete delete
1644 = new org.apache.hadoop.hbase.client.Delete(val);
1645
1646 Table table = new HTable(getConf(), getTableName(getConf()));
1647 table.delete(delete);
1648 table.close();
1649
1650 System.out.println("Delete successful");
1651 return 0;
1652 }
1653 }
1654
1655 abstract static class WalkerBase extends Configured{
1656 protected static CINode findStartNode(Table table, byte[] startKey) throws IOException {
1657 Scan scan = new Scan();
1658 scan.setStartRow(startKey);
1659 scan.setBatch(1);
1660 scan.addColumn(FAMILY_NAME, COLUMN_PREV);
1661
1662 long t1 = System.currentTimeMillis();
1663 ResultScanner scanner = table.getScanner(scan);
1664 Result result = scanner.next();
1665 long t2 = System.currentTimeMillis();
1666 scanner.close();
1667
1668 if ( result != null) {
1669 CINode node = getCINode(result, new CINode());
1670 System.out.printf("FSR %d %s\n", t2 - t1, Bytes.toStringBinary(node.key));
1671 return node;
1672 }
1673
1674 System.out.println("FSR " + (t2 - t1));
1675
1676 return null;
1677 }
1678 protected CINode getNode(byte[] row, Table table, CINode node) throws IOException {
1679 Get get = new Get(row);
1680 get.addColumn(FAMILY_NAME, COLUMN_PREV);
1681 Result result = table.get(get);
1682 return getCINode(result, node);
1683 }
1684 }
1685
1686
1687
1688
1689 private static class Walker extends WalkerBase implements Tool {
1690
1691 public Walker(){}
1692
1693 @Override
1694 public int run(String[] args) throws IOException {
1695
1696 Options options = new Options();
1697 options.addOption("n", "num", true, "number of queries");
1698 options.addOption("s", "start", true, "key to start at, binary string");
1699 options.addOption("l", "logevery", true, "log every N queries");
1700
1701 GnuParser parser = new GnuParser();
1702 CommandLine cmd = null;
1703 try {
1704 cmd = parser.parse(options, args);
1705 if (cmd.getArgs().length != 0) {
1706 throw new ParseException("Command takes no arguments");
1707 }
1708 } catch (ParseException e) {
1709 System.err.println("Failed to parse command line " + e.getMessage());
1710 System.err.println();
1711 HelpFormatter formatter = new HelpFormatter();
1712 formatter.printHelp(getClass().getSimpleName(), options);
1713 System.exit(-1);
1714 }
1715
1716 long maxQueries = Long.MAX_VALUE;
1717 if (cmd.hasOption('n')) {
1718 maxQueries = Long.parseLong(cmd.getOptionValue("n"));
1719 }
1720 Random rand = new SecureRandom();
1721 boolean isSpecificStart = cmd.hasOption('s');
1722
1723 byte[] startKey = isSpecificStart ? Bytes.toBytesBinary(cmd.getOptionValue('s')) : null;
1724 int logEvery = cmd.hasOption('l') ? Integer.parseInt(cmd.getOptionValue('l')) : 1;
1725
1726 Table table = new HTable(getConf(), getTableName(getConf()));
1727 long numQueries = 0;
1728
1729
1730
1731 while (numQueries < maxQueries && (numQueries == 0 || !isSpecificStart)) {
1732 if (!isSpecificStart) {
1733 startKey = new byte[ROWKEY_LENGTH];
1734 rand.nextBytes(startKey);
1735 }
1736 CINode node = findStartNode(table, startKey);
1737 if (node == null && isSpecificStart) {
1738 System.err.printf("Start node not found: %s \n", Bytes.toStringBinary(startKey));
1739 }
1740 numQueries++;
1741 while (node != null && node.prev.length != NO_KEY.length &&
1742 numQueries < maxQueries) {
1743 byte[] prev = node.prev;
1744 long t1 = System.currentTimeMillis();
1745 node = getNode(prev, table, node);
1746 long t2 = System.currentTimeMillis();
1747 if (logEvery > 0 && numQueries % logEvery == 0) {
1748 System.out.printf("CQ %d: %d %s \n", numQueries, t2 - t1, Bytes.toStringBinary(prev));
1749 }
1750 numQueries++;
1751 if (node == null) {
1752 System.err.printf("UNDEFINED NODE %s \n", Bytes.toStringBinary(prev));
1753 } else if (node.prev.length == NO_KEY.length) {
1754 System.err.printf("TERMINATING NODE %s \n", Bytes.toStringBinary(node.key));
1755 }
1756 }
1757 }
1758 table.close();
1759 return 0;
1760 }
1761 }
1762
1763 private static class Clean extends Configured implements Tool {
1764 @Override public int run(String[] args) throws Exception {
1765 if (args.length < 1) {
1766 System.err.println("Usage: Clean <output dir>");
1767 return -1;
1768 }
1769
1770 Path p = new Path(args[0]);
1771 Configuration conf = getConf();
1772 TableName tableName = getTableName(conf);
1773 try (FileSystem fs = HFileSystem.get(conf);
1774 Connection conn = ConnectionFactory.createConnection(conf);
1775 Admin admin = conn.getAdmin()) {
1776 if (admin.tableExists(tableName)) {
1777 admin.disableTable(tableName);
1778 admin.deleteTable(tableName);
1779 }
1780
1781 if (fs.exists(p)) {
1782 fs.delete(p, true);
1783 }
1784 }
1785
1786 return 0;
1787 }
1788 }
1789
1790 static TableName getTableName(Configuration conf) {
1791 return TableName.valueOf(conf.get(TABLE_NAME_KEY, DEFAULT_TABLE_NAME));
1792 }
1793
1794 private static CINode getCINode(Result result, CINode node) {
1795 node.key = Bytes.copy(result.getRow());
1796 if (result.containsColumn(FAMILY_NAME, COLUMN_PREV)) {
1797 node.prev = Bytes.copy(result.getValue(FAMILY_NAME, COLUMN_PREV));
1798 } else {
1799 node.prev = NO_KEY;
1800 }
1801 if (result.containsColumn(FAMILY_NAME, COLUMN_COUNT)) {
1802 node.count = Bytes.toLong(result.getValue(FAMILY_NAME, COLUMN_COUNT));
1803 } else {
1804 node.count = -1;
1805 }
1806 if (result.containsColumn(FAMILY_NAME, COLUMN_CLIENT)) {
1807 node.client = Bytes.toString(result.getValue(FAMILY_NAME, COLUMN_CLIENT));
1808 } else {
1809 node.client = "";
1810 }
1811 return node;
1812 }
1813
1814 protected IntegrationTestingUtility util;
1815
1816 @Override
1817 public void setUpCluster() throws Exception {
1818 util = getTestingUtil(getConf());
1819 boolean isDistributed = util.isDistributedCluster();
1820 util.initializeCluster(isDistributed ? 1 : this.NUM_SLAVES_BASE);
1821 if (!isDistributed) {
1822 util.startMiniMapReduceCluster();
1823 }
1824 this.setConf(util.getConfiguration());
1825 }
1826
1827 @Override
1828 public void cleanUpCluster() throws Exception {
1829 super.cleanUpCluster();
1830 if (util.isDistributedCluster()) {
1831 util.shutdownMiniMapReduceCluster();
1832 }
1833 }
1834
1835 private static boolean isMultiUnevenColumnFamilies(Configuration conf) {
1836 return conf.getBoolean(Generator.MULTIPLE_UNEVEN_COLUMNFAMILIES_KEY,true);
1837 }
1838
1839 @Test
1840 public void testContinuousIngest() throws IOException, Exception {
1841
1842 Configuration conf = getTestingUtil(getConf()).getConfiguration();
1843 if (isMultiUnevenColumnFamilies(getConf())) {
1844
1845 conf.set(FlushPolicyFactory.HBASE_FLUSH_POLICY_KEY, FlushLargeStoresPolicy.class.getName());
1846 }
1847 int ret =
1848 ToolRunner.run(conf, new Loop(), new String[] { "1", "1", "2000000",
1849 util.getDataTestDirOnTestFS("IntegrationTestBigLinkedList").toString(), "1" });
1850 org.junit.Assert.assertEquals(0, ret);
1851 }
1852
1853 private void usage() {
1854 System.err.println("Usage: " + this.getClass().getSimpleName() + " COMMAND [COMMAND options]");
1855 printCommands();
1856 }
1857
1858 private void printCommands() {
1859 System.err.println("Commands:");
1860 System.err.println(" generator Map only job that generates data.");
1861 System.err.println(" verify A map reduce job that looks for holes. Check return code and");
1862 System.err.println(" look at the counts after running. See REFERENCED and");
1863 System.err.println(" UNREFERENCED are ok. Any UNDEFINED counts are bad. Do not run");
1864 System.err.println(" with the Generator.");
1865 System.err.println(" walker " +
1866 "Standalone program that starts following a linked list & emits timing info.");
1867 System.err.println(" print Standalone program that prints nodes in the linked list.");
1868 System.err.println(" delete Standalone program that deletes a·single node.");
1869 System.err.println(" loop Program to Loop through Generator and Verify steps");
1870 System.err.println(" clean Program to clean all left over detritus.");
1871 System.err.println(" search Search for missing keys.");
1872 System.err.flush();
1873 }
1874
1875 @Override
1876 protected void processOptions(CommandLine cmd) {
1877 super.processOptions(cmd);
1878 String[] args = cmd.getArgs();
1879
1880 if (args.length < 1) {
1881 printUsage(this.getClass().getSimpleName() +
1882 " <general options> COMMAND [<COMMAND options>]", "General options:", "");
1883 printCommands();
1884
1885 throw new RuntimeException("Incorrect Number of args.");
1886 }
1887 toRun = args[0];
1888 otherArgs = Arrays.copyOfRange(args, 1, args.length);
1889 }
1890
1891 @Override
1892 public int runTestFromCommandLine() throws Exception {
1893 Tool tool = null;
1894 if (toRun.equalsIgnoreCase("Generator")) {
1895 tool = new Generator();
1896 } else if (toRun.equalsIgnoreCase("Verify")) {
1897 tool = new Verify();
1898 } else if (toRun.equalsIgnoreCase("Loop")) {
1899 Loop loop = new Loop();
1900 loop.it = this;
1901 tool = loop;
1902 } else if (toRun.equalsIgnoreCase("Walker")) {
1903 tool = new Walker();
1904 } else if (toRun.equalsIgnoreCase("Print")) {
1905 tool = new Print();
1906 } else if (toRun.equalsIgnoreCase("Delete")) {
1907 tool = new Delete();
1908 } else if (toRun.equalsIgnoreCase("Clean")) {
1909 tool = new Clean();
1910 } else if (toRun.equalsIgnoreCase("Search")) {
1911 tool = new Search();
1912 } else {
1913 usage();
1914 throw new RuntimeException("Unknown arg");
1915 }
1916
1917 return ToolRunner.run(getConf(), tool, otherArgs);
1918 }
1919
1920 @Override
1921 public TableName getTablename() {
1922 Configuration c = getConf();
1923 return TableName.valueOf(c.get(TABLE_NAME_KEY, DEFAULT_TABLE_NAME));
1924 }
1925
1926 @Override
1927 protected Set<String> getColumnFamilies() {
1928 if (isMultiUnevenColumnFamilies(getConf())) {
1929 return Sets.newHashSet(Bytes.toString(FAMILY_NAME), Bytes.toString(BIG_FAMILY_NAME),
1930 Bytes.toString(TINY_FAMILY_NAME));
1931 } else {
1932 return Sets.newHashSet(Bytes.toString(FAMILY_NAME));
1933 }
1934 }
1935
1936 private static void setJobConf(Job job, int numMappers, long numNodes,
1937 Integer width, Integer wrapMultiplier, Integer numWalkers) {
1938 job.getConfiguration().setInt(GENERATOR_NUM_MAPPERS_KEY, numMappers);
1939 job.getConfiguration().setLong(GENERATOR_NUM_ROWS_PER_MAP_KEY, numNodes);
1940 if (width != null) {
1941 job.getConfiguration().setInt(GENERATOR_WIDTH_KEY, width);
1942 }
1943 if (wrapMultiplier != null) {
1944 job.getConfiguration().setInt(GENERATOR_WRAP_KEY, wrapMultiplier);
1945 }
1946 if (numWalkers != null) {
1947 job.getConfiguration().setInt(CONCURRENT_WALKER_KEY, numWalkers);
1948 }
1949 }
1950
1951 public static void setJobScannerConf(Job job) {
1952
1953 job.getConfiguration().setBoolean(ScannerCallable.LOG_SCANNER_ACTIVITY, true);
1954 job.getConfiguration().setInt(TableRecordReaderImpl.LOG_PER_ROW_COUNT, 100000);
1955 }
1956
1957 public static void main(String[] args) throws Exception {
1958 Configuration conf = HBaseConfiguration.create();
1959 IntegrationTestingUtility.setUseDistributedCluster(conf);
1960 int ret = ToolRunner.run(conf, new IntegrationTestBigLinkedList(), args);
1961 System.exit(ret);
1962 }
1963 }