View Javadoc

1   /**
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
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  * This is an integration test borrowed from goraci, written by Keith Turner,
127  * which is in turn inspired by the Accumulo test called continous ingest (ci).
128  * The original source code can be found here:
129  * https://github.com/keith-turner/goraci
130  * https://github.com/enis/goraci/
131  *
132  * Apache Accumulo [0] has a simple test suite that verifies that data is not
133  * lost at scale. This test suite is called continuous ingest. This test runs
134  * many ingest clients that continually create linked lists containing 25
135  * million nodes. At some point the clients are stopped and a map reduce job is
136  * run to ensure no linked list has a hole. A hole indicates data was lost.··
137  *
138  * The nodes in the linked list are random. This causes each linked list to
139  * spread across the table. Therefore if one part of a table loses data, then it
140  * will be detected by references in another part of the table.
141  *
142  * THE ANATOMY OF THE TEST
143  *
144  * Below is rough sketch of how data is written. For specific details look at
145  * the Generator code.
146  *
147  * 1 Write out 1 million nodes· 2 Flush the client· 3 Write out 1 million that
148  * reference previous million· 4 If this is the 25th set of 1 million nodes,
149  * then update 1st set of million to point to last· 5 goto 1
150  *
151  * The key is that nodes only reference flushed nodes. Therefore a node should
152  * never reference a missing node, even if the ingest client is killed at any
153  * point in time.
154  *
155  * When running this test suite w/ Accumulo there is a script running in
156  * parallel called the Aggitator that randomly and continuously kills server
157  * processes.·· The outcome was that many data loss bugs were found in Accumulo
158  * by doing this.· This test suite can also help find bugs that impact uptime
159  * and stability when· run for days or weeks.··
160  *
161  * This test suite consists the following· - a few Java programs· - a little
162  * helper script to run the java programs - a maven script to build it.··
163  *
164  * When generating data, its best to have each map task generate a multiple of
165  * 25 million. The reason for this is that circular linked list are generated
166  * every 25M. Not generating a multiple in 25M will result in some nodes in the
167  * linked list not having references. The loss of an unreferenced node can not
168  * be detected.
169  *
170  *
171  * Below is a description of the Java programs
172  *
173  * Generator - A map only job that generates data. As stated previously,·its best to generate data
174  * in multiples of 25M. An option is also available to allow concurrent walkers to select and walk
175  * random flushed loops during this phase.
176  *
177  * Verify - A map reduce job that looks for holes. Look at the counts after running. REFERENCED and
178  * UNREFERENCED are· ok, any UNDEFINED counts are bad. Do not run at the· same
179  * time as the Generator.
180  *
181  * Walker - A standalone program that start following a linked list· and emits timing info.··
182  *
183  * Print - A standalone program that prints nodes in the linked list
184  *
185  * Delete - A standalone program that deletes a single node
186  *
187  * This class can be run as a unit test, as an integration test, or from the command line
188  *
189  * ex:
190  * ./hbase org.apache.hadoop.hbase.test.IntegrationTestBigLinkedList
191  *    loop 2 1 100000 /temp 1 1000 50 1 0
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   //link to the id of the prev node in the linked list
207   protected static final byte[] COLUMN_PREV = Bytes.toBytes("prev");
208 
209   //identifier of the mapred task that generated this row
210   protected static final byte[] COLUMN_CLIENT = Bytes.toBytes("client");
211 
212   //the id of the row within the same client.
213   protected static final byte[] COLUMN_COUNT = Bytes.toBytes("count");
214 
215   /** How many rows to write per map task. This has to be a multiple of 25M */
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; // number of slaves for the cluster
232 
233   private static final int MISSING_ROWS_TO_LOG = 10; // YARN complains when too many counters
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    * A Map only job that generates random linked list and stores them.
253    */
254   static class Generator extends Configured implements Tool {
255 
256     private static final Log LOG = LogFactory.getLog(Generator.class);
257 
258     /**
259      * Set this configuration if you want to test single-column family flush works. If set, we will
260      * add a big column family and a small column family on either side of the usual ITBLL 'meta'
261      * column family. When we write out the ITBLL, we will also add to the big column family a value
262      * bigger than that for ITBLL and for small, something way smaller. The idea is that when
263      * flush-by-column family rather than by region is enabled, we can see if ITBLL is broke in any
264      * way. Here is how you would pass it:
265      * <p>
266      * $ ./bin/hbase org.apache.hadoop.hbase.test.IntegrationTestBigLinkedList
267      * -Dgenerator.multiple.columnfamilies=true generator 1 10 g
268      */
269     public static final String MULTIPLE_UNEVEN_COLUMNFAMILIES_KEY =
270         "generator.multiple.columnfamilies";
271 
272     /**
273      * Set this configuration if you want to scale up the size of test data quickly.
274      * <p>
275      * $ ./bin/hbase org.apache.hadoop.hbase.test.IntegrationTestBigLinkedList
276      * -Dgenerator.big.family.value.size=1024 generator 1 10 output
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           // Use Random64 to avoid issue described in HBASE-21256.
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     /** Ensure output files from prev-job go to map inputs for current job */
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      * Some ASCII art time:
384      * <p>
385      * [ . . . ] represents one batch of random longs of length WIDTH
386      * <pre>
387      *                _________________________
388      *               |                  ______ |
389      *               |                 |      ||
390      *             .-+-----------------+-----.||
391      *             | |                 |     |||
392      * first   = [ . . . . . . . . . . . ]   |||
393      *             ^ ^ ^ ^ ^ ^ ^ ^ ^ ^ ^     |||
394      *             | | | | | | | | | | |     |||
395      * prev    = [ . . . . . . . . . . . ]   |||
396      *             ^ ^ ^ ^ ^ ^ ^ ^ ^ ^ ^     |||
397      *             | | | | | | | | | | |     |||
398      * current = [ . . . . . . . . . . . ]   |||
399      *                                       |||
400      * ...                                   |||
401      *                                       |||
402      * last    = [ . . . . . . . . . . . ]   |||
403      *             ^ ^ ^ ^ ^ ^ ^ ^ ^ ^ ^_____|||
404      *             |                 |________||
405      *             |___________________________|
406      * </pre>
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             // this block of code turns the 1 million linked list of length 25 into one giant
519             //circular linked list of 25 million
520             circularLeftShift(first);
521             persist(output, -1, prev, first, null);
522             // At this point the entire loop has been flushed so we can add one of its nodes to the
523             // concurrent walker
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             // Tickle progress every so often else maprunner will think us hung
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           // See if we are to write multiple columns.
568           if (this.multipleUnevenColumnFamilies) {
569             // Use any column name.
570             put.addColumn(TINY_FAMILY_NAME, TINY_FAMILY_NAME, this.tinyValue);
571             // Use any column name.
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             // no-op
599           }
600         }
601       }
602 
603       /**
604        * Randomly selects and walks a random flushed loop concurrently with the Generator Mapper by
605        * spawning ConcurrentWalker's with specified StartNodes. These ConcurrentWalker's are
606        * configured to only log erroneous nodes.
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           // If isSpecificStart is set, only walk one list from that particular node.
673           // Note that in case of circular (or P-shaped) list it will walk forever, as is
674           // the case in normal run without startKey.
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               // Increment for successful walk
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           // Always add these families. Just skip writing to them when we do not test per CF flush.
735           htd.addFamily(new HColumnDescriptor(BIG_FAMILY_NAME));
736           htd.addFamily(new HColumnDescriptor(TINY_FAMILY_NAME));
737 
738           // If we want to pre-split compute how many splits.
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             // Looks like we're just letting things play out.
758             // Create a table with on region by default.
759             // This will make the splitting work hard.
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); //identity mapper
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    * Tool to search missing rows in WALs and hfiles.
879    * Pass in file or dir of keys to search for. Key file must have been written by Verify step
880    * (we depend on the format it writes out. We'll read them in and then search in hbase
881    * WALs and oldWALs dirs (Some of this is TODO).
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      * WALPlayer override that searches for keys loaded in the setup.
908      */
909     public static class WALSearcher extends WALPlayer {
910       public WALSearcher(Configuration conf) {
911         super(conf);
912       }
913 
914       /**
915        * The actual searcher mapper.
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           // TODO: Can I do a better compare than this copying out key?
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       // Put in place the above WALMapperSearcher.
956       @Override
957       public Job createSubmittableJob(String[] args) throws IOException {
958         Job job = super.createSubmittableJob(args);
959         // Call my class instead.
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       // Now read all WALs. In two dirs. Presumes certain layout.
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           // Skip "_SUCCESS" file.
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       // Return entries that are flagged Counts.UNDEFINED in the value. Return the row. This is
1014       // what is missing.
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    * A Map Reduce job that verifies that the linked lists generated by
1038    * {@link Generator} do not have any holes.
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      * Don't change the order of these enums. Their ordinals are used as type flag when we emit
1085      * problems found from the reducer.
1086      */
1087     public static enum Counts {
1088       UNREFERENCED, UNDEFINED, REFERENCED, CORRUPT, EXTRAREFERENCES, EXTRA_UNDEF_REFERENCES,
1089       LOST_FAMILIES
1090     }
1091 
1092     /**
1093      * Per reducer, we output problem rows as byte arrasy so can be used as input for
1094      * subsequent investigative mapreduce jobs. Each emitted value is prefaced by a one byte flag
1095      * saying what sort of emission it is. Flag is the Count enum ordinal as a short.
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        * @param ordinal
1127        * @param r
1128        * @return Return new byte array that has <code>ordinal</code> as prefix on front taking up
1129        * Bytes.SIZEOF_SHORT bytes followed by <code>r</code>
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        * @param bs
1144        * @return Type from the Counts enum of this row. Reads prefix added by
1145        * {@link #addPrefixFlag(int, byte[])}
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        * @param bw
1154        * @return Row bytes minus the type flag.
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         // TODO check for more than one def, should not happen
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           // This is bad, found a node that is referenced but not defined. It must have been
1198           // lost, emit some info about this node for debugging purposes.
1199           // Write out a line per reference. If more than one, flag it.;
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             // Print out missing row; doing get on reference gives info on when the referencer
1214             // was added which can help a little debugging. This info is only available in mapper
1215             // output -- the 'Linked List error Key...' log message above. What we emit here is
1216             // useless for debugging.
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           // node is defined but not referenced
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             // Skip first reference.
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           // node is defined and referenced
1238           context.getCounter(Counts.REFERENCED).increment(1);
1239         }
1240       }
1241 
1242       /**
1243        * Dump out extra info around references if there are any. Helps debugging.
1244        * @return StringBuilder filled with references if any.
1245        * @throws IOException
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         // If a row is a reference but has no define, print the content of the row that has
1255         // this row as a 'prev'; it will help debug.  The missing row was written just before
1256         // the row we are dumping out here.
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                 // Key here probably has trailing zeros on it.
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           // We don't have access to the counters to know if we have "bad" counts
1356           return 0;
1357         }
1358 
1359         // If we find no unexpected values, the job didn't outright fail
1360         if (verifyUnexpectedValues(counters)) {
1361           // We didn't check referenced+unreferenced counts, leave that to visual inspection
1362           return 0;
1363         }
1364       }
1365 
1366       // We failed
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       // Run through each check, even if we fail one early
1383       boolean success = verifyExpectedValues(expectedReferenced, counters);
1384 
1385       if (!verifyUnexpectedValues(counters)) {
1386         // We found counter objects which imply failure
1387         success = false;
1388       }
1389 
1390       if (!success) {
1391         handleFailure(counters);
1392       }
1393       return success;
1394     }
1395 
1396     /**
1397      * Verify the values in the Counters against the expected number of entries written.
1398      *
1399      * @param expectedReferenced
1400      *          Expected number of referenced entrires
1401      * @param counters
1402      *          The Job's Counters object
1403      * @return True if the values match what's expected, false otherwise
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      * Verify that the Counters don't contain values which indicate an outright failure from the Reducers.
1429      *
1430      * @param counters
1431      *          The Job's counters
1432      * @return True if the "bad" counter objects are 0, false otherwise
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    * Executes Generate and Verify in a loop. Data is not cleaned between runs, so each iteration
1480    * adds more data.
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(); //create a random UUID.
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(); //create a random UUID.
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; //run indefinitely (kind of)
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    * A stand alone program that prints out portions of a list created by {@link Generator}
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    * A stand alone program that deletes a single node.
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    * A stand alone program that follows a linked list created by {@link Generator} and prints
1687    * timing info.
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       // If isSpecificStart is set, only walk one list from that particular node.
1729       // Note that in case of circular (or P-shaped) list it will walk forever, as is
1730       // the case in normal run without startKey.
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     //Loop <num iterations> <num mappers> <num nodes per mapper> <output dir> <num reducers>
1842     Configuration conf = getTestingUtil(getConf()).getConfiguration();
1843     if (isMultiUnevenColumnFamilies(getConf())) {
1844       // make sure per CF flush is on
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     //get the class, run with the conf
1880     if (args.length < 1) {
1881       printUsage(this.getClass().getSimpleName() +
1882         " <general options> COMMAND [<COMMAND options>]", "General options:", "");
1883       printCommands();
1884       // Have to throw an exception here to stop the processing. Looks ugly but gets message across.
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     // Make sure scanners log something useful to make debugging possible.
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 }