Actually invoke the BadKeyOrphanReader. Also committing a bunch of

TODOs for later cleanup.

src/java/voldemort/client/protocol/admin/AdminClient.java

@@ -1606,6 +1606,9 @@ public Iterator<Pair<ByteArray, Versioned<byte[]>>> fetchEntries(int nodeId,
             return fetchEntries(nodeId, storeName, partitionList, filter, fetchMasterEntries, 0);
         }
 
+        // TODO: The use of "Pair" in the return for a fundamental type is
+        // awkward. We should have a core KeyValue type that effectively wraps
+        // up a ByteArray and a Versioned<byte[]>.
         /**
          * Fetch key/value tuples belonging to this map of replica type to
          * partition list

src/java/voldemort/utils/ConsistencyFix.java

@@ -46,6 +46,9 @@
 import voldemort.versioning.VectorClock;
 import voldemort.versioning.Versioned;
 
+// TODO: Move to new directory voldemort/tools. Also move ConsistencyCheck,
+// Rebalance, and possibly other tools (shells and so on). This would reduce the
+// amount of different stuff in the utils directory.
 public class ConsistencyFix {
 
     private static final Logger logger = Logger.getLogger(ConsistencyFix.class);
@@ -135,6 +138,10 @@ public String execute(int parallelism,
         ExecutorService badKeyWriterService;
         ExecutorService consistencyFixWorkers;
 
+        // TODO: Add ThreadFactory usage to ExecutorService usage so that
+        // threads have sane names. Figure out if any parameters currently
+        // passed from object-to-object could be given directly to factories.
+
         // Create BadKeyWriter thread
         BlockingQueue<BadKeyResult> badKeyQOut = new ArrayBlockingQueue<BadKeyResult>(parallelism * 10);
         badKeyWriterService = Executors.newSingleThreadExecutor();
@@ -155,11 +162,19 @@ public String execute(int parallelism,
         // Create BadKeyReader thread
         CountDownLatch allBadKeysReadLatch = new CountDownLatch(1);
         badKeyReaderService = Executors.newSingleThreadExecutor();
-        badKeyReaderService.submit(new BadKeyReader(allBadKeysReadLatch,
-                                                    badKeyFileIn,
-                                                    this,
-                                                    consistencyFixWorkers,
-                                                    badKeyQOut));
+        if(!orphanFormat) {
+            badKeyReaderService.submit(new BadKeyReader(allBadKeysReadLatch,
+                                                        badKeyFileIn,
+                                                        this,
+                                                        consistencyFixWorkers,
+                                                        badKeyQOut));
+        } else {
+            badKeyReaderService.submit(new BadKeyOrphanReader(allBadKeysReadLatch,
+                                                              badKeyFileIn,
+                                                              this,
+                                                              consistencyFixWorkers,
+                                                              badKeyQOut));
+        }
         logger.info("Created badKeyReader.");
 
         try {

src/java/voldemort/utils/ConsistencyFixCLI.java

@@ -38,6 +38,7 @@ public static void printUsage(String errMessage) {
     private static class Options {
 
         public final static int defaultParallelism = 8;
+        // TODO: change name to progressPeriodMs (or S, as case may be)
         public final static long defaultProgressBar = 1000;
         public final static long defaultPerServerIOPSLimit = 100;