distributedlog-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From si...@apache.org
Subject [16/51] [partial] incubator-distributedlog git commit: DL-4: Repackage the source under apache namespace
Date Thu, 05 Jan 2017 00:51:21 GMT
http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/b44820b5/distributedlog-core/src/main/java/com/twitter/distributedlog/tools/DistributedLogTool.java
----------------------------------------------------------------------
diff --git a/distributedlog-core/src/main/java/com/twitter/distributedlog/tools/DistributedLogTool.java b/distributedlog-core/src/main/java/com/twitter/distributedlog/tools/DistributedLogTool.java
deleted file mode 100644
index 03d70bd..0000000
--- a/distributedlog-core/src/main/java/com/twitter/distributedlog/tools/DistributedLogTool.java
+++ /dev/null
@@ -1,2873 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package com.twitter.distributedlog.tools;
-
-import java.io.BufferedReader;
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileNotFoundException;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.InputStreamReader;
-import java.io.OutputStreamWriter;
-import java.io.PrintWriter;
-import java.net.MalformedURLException;
-import java.net.URI;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.Enumeration;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.SortedMap;
-import java.util.TreeMap;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicReference;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-
-import com.google.common.base.Preconditions;
-import com.twitter.distributedlog.BKDistributedLogNamespace;
-import com.twitter.distributedlog.Entry;
-import com.twitter.distributedlog.MetadataAccessor;
-import com.twitter.distributedlog.callback.NamespaceListener;
-import com.twitter.distributedlog.impl.BKNamespaceDriver;
-import com.twitter.distributedlog.logsegment.LogSegmentMetadataStore;
-import com.twitter.distributedlog.namespace.DistributedLogNamespace;
-import com.twitter.distributedlog.namespace.DistributedLogNamespaceBuilder;
-import com.twitter.distributedlog.namespace.NamespaceDriver;
-import com.twitter.distributedlog.util.Utils;
-import org.apache.bookkeeper.client.BKException;
-import org.apache.bookkeeper.client.BookKeeper;
-import org.apache.bookkeeper.client.BookKeeperAccessor;
-import org.apache.bookkeeper.client.BookKeeperAdmin;
-import org.apache.bookkeeper.client.LedgerEntry;
-import org.apache.bookkeeper.client.LedgerHandle;
-import org.apache.bookkeeper.client.LedgerMetadata;
-import org.apache.bookkeeper.client.LedgerReader;
-import org.apache.bookkeeper.net.BookieSocketAddress;
-import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks;
-import org.apache.bookkeeper.util.IOUtils;
-import org.apache.commons.cli.CommandLine;
-import org.apache.commons.cli.Options;
-import org.apache.commons.cli.ParseException;
-import org.apache.commons.codec.binary.Hex;
-import org.apache.commons.configuration.ConfigurationException;
-import org.apache.commons.lang3.tuple.Pair;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.collect.Lists;
-import com.google.common.collect.Sets;
-import com.google.common.util.concurrent.RateLimiter;
-import com.twitter.distributedlog.AsyncLogReader;
-import com.twitter.distributedlog.AsyncLogWriter;
-import com.twitter.distributedlog.BookKeeperClient;
-import com.twitter.distributedlog.BookKeeperClientBuilder;
-import com.twitter.distributedlog.DLSN;
-import com.twitter.distributedlog.DistributedLogConfiguration;
-import com.twitter.distributedlog.DistributedLogConstants;
-import com.twitter.distributedlog.DistributedLogManager;
-import com.twitter.distributedlog.exceptions.LogNotFoundException;
-import com.twitter.distributedlog.LogReader;
-import com.twitter.distributedlog.LogRecord;
-import com.twitter.distributedlog.LogRecordWithDLSN;
-import com.twitter.distributedlog.LogSegmentMetadata;
-import com.twitter.distributedlog.ZooKeeperClient;
-import com.twitter.distributedlog.ZooKeeperClientBuilder;
-import com.twitter.distributedlog.auditor.DLAuditor;
-import com.twitter.distributedlog.bk.LedgerAllocator;
-import com.twitter.distributedlog.bk.LedgerAllocatorUtils;
-import com.twitter.distributedlog.impl.metadata.BKDLConfig;
-import com.twitter.distributedlog.metadata.MetadataUpdater;
-import com.twitter.distributedlog.metadata.LogSegmentMetadataStoreUpdater;
-import com.twitter.distributedlog.util.SchedulerUtils;
-import com.twitter.util.Await;
-import com.twitter.util.FutureEventListener;
-
-import static com.google.common.base.Charsets.UTF_8;
-
-public class DistributedLogTool extends Tool {
-
-    static final Logger logger = LoggerFactory.getLogger(DistributedLogTool.class);
-
-    static final List<String> EMPTY_LIST = Lists.newArrayList();
-
-    static int compareByCompletionTime(long time1, long time2) {
-        return time1 > time2 ? 1 : (time1 < time2 ? -1 : 0);
-    }
-
-    static final Comparator<LogSegmentMetadata> LOGSEGMENT_COMPARATOR_BY_TIME = new Comparator<LogSegmentMetadata>() {
-        @Override
-        public int compare(LogSegmentMetadata o1, LogSegmentMetadata o2) {
-            if (o1.isInProgress() && o2.isInProgress()) {
-                return compareByCompletionTime(o1.getFirstTxId(), o2.getFirstTxId());
-            } else if (!o1.isInProgress() && !o2.isInProgress()) {
-                return compareByCompletionTime(o1.getCompletionTime(), o2.getCompletionTime());
-            } else if (o1.isInProgress() && !o2.isInProgress()) {
-                return compareByCompletionTime(o1.getFirstTxId(), o2.getCompletionTime());
-            } else {
-                return compareByCompletionTime(o1.getCompletionTime(), o2.getFirstTxId());
-            }
-        }
-    };
-
-    static DLSN parseDLSN(String dlsnStr) throws ParseException {
-        if (dlsnStr.equals("InitialDLSN")) {
-            return DLSN.InitialDLSN;
-        }
-        String[] parts = dlsnStr.split(",");
-        if (parts.length != 3) {
-            throw new ParseException("Invalid dlsn : " + dlsnStr);
-        }
-        try {
-            return new DLSN(Long.parseLong(parts[0]), Long.parseLong(parts[1]), Long.parseLong(parts[2]));
-        } catch (Exception nfe) {
-            throw new ParseException("Invalid dlsn : " + dlsnStr);
-        }
-    }
-
-    /**
-     * Per DL Command, which parses basic options. e.g. uri.
-     */
-    protected abstract static class PerDLCommand extends OptsCommand {
-
-        protected Options options = new Options();
-        protected final DistributedLogConfiguration dlConf;
-        protected URI uri;
-        protected String zkAclId = null;
-        protected boolean force = false;
-        protected DistributedLogNamespace namespace = null;
-
-        protected PerDLCommand(String name, String description) {
-            super(name, description);
-            dlConf = new DistributedLogConfiguration();
-            // Tools are allowed to read old metadata as long as they can interpret it
-            dlConf.setDLLedgerMetadataSkipMinVersionCheck(true);
-            options.addOption("u", "uri", true, "DistributedLog URI");
-            options.addOption("c", "conf", true, "DistributedLog Configuration File");
-            options.addOption("a", "zk-acl-id", true, "Zookeeper ACL ID");
-            options.addOption("f", "force", false, "Force command (no warnings or prompts)");
-        }
-
-        @Override
-        protected int runCmd(CommandLine commandLine) throws Exception {
-            try {
-                parseCommandLine(commandLine);
-            } catch (ParseException pe) {
-                System.err.println("ERROR: failed to parse commandline : '" + pe.getMessage() + "'");
-                printUsage();
-                return -1;
-            }
-            try {
-                return runCmd();
-            } finally {
-                if (null != namespace) {
-                    namespace.close();
-                }
-            }
-        }
-
-        protected abstract int runCmd() throws Exception;
-
-        @Override
-        protected Options getOptions() {
-            return options;
-        }
-
-        protected void parseCommandLine(CommandLine cmdline) throws ParseException {
-            if (!cmdline.hasOption("u")) {
-                throw new ParseException("No distributedlog uri provided.");
-            }
-            uri = URI.create(cmdline.getOptionValue("u"));
-            if (cmdline.hasOption("c")) {
-                String configFile = cmdline.getOptionValue("c");
-                try {
-                    dlConf.loadConf(new File(configFile).toURI().toURL());
-                } catch (ConfigurationException e) {
-                    throw new ParseException("Failed to load distributedlog configuration from " + configFile + ".");
-                } catch (MalformedURLException e) {
-                    throw new ParseException("Failed to load distributedlog configuration from " + configFile + ": malformed uri.");
-                }
-            }
-            if (cmdline.hasOption("a")) {
-                zkAclId = cmdline.getOptionValue("a");
-            }
-            if (cmdline.hasOption("f")) {
-                force = true;
-            }
-        }
-
-        protected DistributedLogConfiguration getConf() {
-            return dlConf;
-        }
-
-        protected URI getUri() {
-            return uri;
-        }
-
-        protected void setUri(URI uri) {
-            this.uri = uri;
-        }
-
-        protected String getZkAclId() {
-            return zkAclId;
-        }
-
-        protected void setZkAclId(String zkAclId) {
-            this.zkAclId = zkAclId;
-        }
-
-        protected boolean getForce() {
-            return force;
-        }
-
-        protected void setForce(boolean force) {
-            this.force = force;
-        }
-
-        protected DistributedLogNamespace getNamespace() throws IOException {
-            if (null == this.namespace) {
-                this.namespace = DistributedLogNamespaceBuilder.newBuilder()
-                        .uri(getUri())
-                        .conf(getConf())
-                        .build();
-            }
-            return this.namespace;
-        }
-
-        protected LogSegmentMetadataStore getLogSegmentMetadataStore() throws IOException {
-            return getNamespace()
-                    .getNamespaceDriver()
-                    .getLogStreamMetadataStore(NamespaceDriver.Role.READER)
-                    .getLogSegmentMetadataStore();
-        }
-
-        protected ZooKeeperClient getZooKeeperClient() throws IOException {
-            NamespaceDriver driver = getNamespace().getNamespaceDriver();
-            assert(driver instanceof BKNamespaceDriver);
-            return ((BKNamespaceDriver) driver).getWriterZKC();
-        }
-
-        protected BookKeeperClient getBookKeeperClient() throws IOException {
-            NamespaceDriver driver = getNamespace().getNamespaceDriver();
-            assert(driver instanceof BKNamespaceDriver);
-            return ((BKNamespaceDriver) driver).getReaderBKC();
-        }
-    }
-
-    /**
-     * Base class for simple command with no resource setup requirements.
-     */
-    public abstract static class SimpleCommand extends OptsCommand {
-
-        protected final Options options = new Options();
-
-        SimpleCommand(String name, String description) {
-            super(name, description);
-        }
-
-        @Override
-        protected int runCmd(CommandLine commandLine) throws Exception {
-            try {
-                parseCommandLine(commandLine);
-            } catch (ParseException pe) {
-                System.err.println("ERROR: failed to parse commandline : '" + pe.getMessage() + "'");
-                printUsage();
-                return -1;
-            }
-            return runSimpleCmd();
-        }
-
-        abstract protected int runSimpleCmd() throws Exception;
-
-        abstract protected void parseCommandLine(CommandLine cmdline) throws ParseException;
-
-        @Override
-        protected Options getOptions() {
-            return options;
-        }
-    }
-
-    /**
-     * Per Stream Command, which parse common options for per stream. e.g. stream name.
-     */
-    abstract static class PerStreamCommand extends PerDLCommand {
-
-        protected String streamName;
-
-        protected PerStreamCommand(String name, String description) {
-            super(name, description);
-            options.addOption("s", "stream", true, "Stream Name");
-        }
-
-        @Override
-        protected void parseCommandLine(CommandLine cmdline) throws ParseException {
-            super.parseCommandLine(cmdline);
-            if (!cmdline.hasOption("s")) {
-                throw new ParseException("No stream name provided.");
-            }
-            streamName = cmdline.getOptionValue("s");
-        }
-
-        protected String getStreamName() {
-            return streamName;
-        }
-
-        protected void setStreamName(String streamName) {
-            this.streamName = streamName;
-        }
-    }
-
-    /**
-     * NOTE: we might consider adding a command to 'delete' namespace. The implementation of the namespace
-     *       driver should implement the 'delete' operation.
-     */
-    protected static class DeleteAllocatorPoolCommand extends PerDLCommand {
-
-        int concurrency = 1;
-        String allocationPoolPath = DistributedLogConstants.ALLOCATION_POOL_NODE;
-
-        DeleteAllocatorPoolCommand() {
-            super("delete_allocator_pool", "Delete allocator pool for a given distributedlog instance");
-            options.addOption("t", "concurrency", true, "Concurrency on deleting allocator pool.");
-            options.addOption("ap", "allocation-pool-path", true, "Ledger Allocation Pool Path");
-        }
-
-        @Override
-        protected void parseCommandLine(CommandLine cmdline) throws ParseException {
-            super.parseCommandLine(cmdline);
-            if (cmdline.hasOption("t")) {
-                concurrency = Integer.parseInt(cmdline.getOptionValue("t"));
-                if (concurrency <= 0) {
-                    throw new ParseException("Invalid concurrency value : " + concurrency + ": it must be greater or equal to 0.");
-                }
-            }
-            if (cmdline.hasOption("ap")) {
-                allocationPoolPath = cmdline.getOptionValue("ap");
-                if (!allocationPoolPath.startsWith(".") || !allocationPoolPath.contains("allocation")) {
-                    throw new ParseException("Invalid allocation pool path : " + allocationPoolPath + ": it must starts with a '.' and must contains 'allocation'");
-                }
-            }
-        }
-
-        @Override
-        protected int runCmd() throws Exception {
-            String rootPath = getUri().getPath() + "/" + allocationPoolPath;
-            final ScheduledExecutorService allocationExecutor = Executors.newSingleThreadScheduledExecutor();
-            ExecutorService executorService = Executors.newFixedThreadPool(concurrency);
-            Preconditions.checkArgument(getNamespace() instanceof BKDistributedLogNamespace);
-            BKDistributedLogNamespace bkns = (BKDistributedLogNamespace) getNamespace();
-            final ZooKeeperClient zkc = ((BKNamespaceDriver) bkns.getNamespaceDriver()).getWriterZKC();
-            final BookKeeperClient bkc = ((BKNamespaceDriver) bkns.getNamespaceDriver()).getReaderBKC();
-            try {
-                List<String> pools = zkc.get().getChildren(rootPath, false);
-                final LinkedBlockingQueue<String> poolsToDelete = new LinkedBlockingQueue<String>();
-                if (getForce() || IOUtils.confirmPrompt("Are you sure you want to delete allocator pools : " + pools)) {
-                    for (String pool : pools) {
-                        poolsToDelete.add(rootPath + "/" + pool);
-                    }
-                    final CountDownLatch doneLatch = new CountDownLatch(concurrency);
-                    for (int i = 0; i < concurrency; i++) {
-                        final int tid = i;
-                        executorService.submit(new Runnable() {
-                            @Override
-                            public void run() {
-                                while (!poolsToDelete.isEmpty()) {
-                                    String poolPath = poolsToDelete.poll();
-                                    if (null == poolPath) {
-                                        break;
-                                    }
-                                    try {
-                                        LedgerAllocator allocator =
-                                                LedgerAllocatorUtils.createLedgerAllocatorPool(poolPath, 0, getConf(),
-                                                        zkc, bkc,
-                                                        allocationExecutor);
-                                        allocator.delete();
-                                        System.out.println("Deleted allocator pool : " + poolPath + " .");
-                                    } catch (IOException ioe) {
-                                        System.err.println("Failed to delete allocator pool " + poolPath + " : " + ioe.getMessage());
-                                    }
-                                }
-                                doneLatch.countDown();
-                                System.out.println("Thread " + tid + " is done.");
-                            }
-                        });
-                    }
-                    doneLatch.await();
-                }
-            } finally {
-                executorService.shutdown();
-                allocationExecutor.shutdown();
-            }
-            return 0;
-        }
-
-        @Override
-        protected String getUsage() {
-            return "delete_allocator_pool";
-        }
-    }
-
-    public static class ListCommand extends PerDLCommand {
-
-        boolean printMetadata = false;
-        boolean printHex = false;
-
-        ListCommand() {
-            super("list", "list streams of a given distributedlog instance");
-            options.addOption("m", "meta", false, "Print metadata associated with each stream");
-            options.addOption("x", "hex", false, "Print metadata in hex format");
-        }
-
-        @Override
-        protected void parseCommandLine(CommandLine cmdline) throws ParseException {
-            super.parseCommandLine(cmdline);
-            printMetadata = cmdline.hasOption("m");
-            printHex = cmdline.hasOption("x");
-        }
-
-        @Override
-        protected String getUsage() {
-            return "list [options]";
-        }
-
-        @Override
-        protected int runCmd() throws Exception {
-            printStreams(getNamespace());
-            return 0;
-        }
-
-        protected void printStreams(DistributedLogNamespace namespace) throws Exception {
-            Iterator<String> streams = namespace.getLogs();
-            System.out.println("Streams under " + getUri() + " : ");
-            System.out.println("--------------------------------");
-            while (streams.hasNext()) {
-                String streamName = streams.next();
-                System.out.println(streamName);
-                if (!printMetadata) {
-                    continue;
-                }
-                MetadataAccessor accessor =
-                        namespace.getNamespaceDriver().getMetadataAccessor(streamName);
-                byte[] metadata = accessor.getMetadata();
-                if (null == metadata || metadata.length == 0) {
-                    continue;
-                }
-                if (printHex) {
-                    System.out.println(Hex.encodeHexString(metadata));
-                } else {
-                    System.out.println(new String(metadata, UTF_8));
-                }
-                System.out.println("");
-            }
-            System.out.println("--------------------------------");
-        }
-    }
-
-    public static class WatchNamespaceCommand extends PerDLCommand implements NamespaceListener {
-        private Set<String> currentSet = Sets.<String>newHashSet();
-        private CountDownLatch doneLatch = new CountDownLatch(1);
-
-        WatchNamespaceCommand() {
-            super("watch", "watch and report changes for a dl namespace");
-        }
-
-        @Override
-        protected void parseCommandLine(CommandLine cmdline) throws ParseException {
-            super.parseCommandLine(cmdline);
-        }
-
-        @Override
-        protected String getUsage() {
-            return "watch [options]";
-        }
-
-        @Override
-        protected int runCmd() throws Exception {
-            watchAndReportChanges(getNamespace());
-            doneLatch.await();
-            return 0;
-        }
-
-        @Override
-        public synchronized void onStreamsChanged(Iterator<String> streams) {
-            Set<String> updatedSet = Sets.newHashSet(streams);
-            Set<String> oldStreams = Sets.difference(currentSet, updatedSet);
-            Set<String> newStreams = Sets.difference(updatedSet, currentSet);
-            currentSet = updatedSet;
-
-            System.out.println("Old streams : ");
-            for (String stream : oldStreams) {
-                System.out.println(stream);
-            }
-
-            System.out.println("New streams : ");
-            for (String stream : newStreams) {
-                System.out.println(stream);
-            }
-
-            System.out.println("");
-        }
-
-        protected void watchAndReportChanges(DistributedLogNamespace namespace) throws Exception {
-            namespace.registerNamespaceListener(this);
-        }
-    }
-
-    protected static class InspectCommand extends PerDLCommand {
-
-        int numThreads = 1;
-        String streamPrefix = null;
-        boolean printInprogressOnly = false;
-        boolean dumpEntries = false;
-        boolean orderByTime = false;
-        boolean printStreamsOnly = false;
-        boolean checkInprogressOnly = false;
-
-        InspectCommand() {
-            super("inspect", "Inspect streams under a given dl uri to find any potential corruptions");
-            options.addOption("t", "threads", true, "Number threads to do inspection.");
-            options.addOption("ft", "filter", true, "Stream filter by prefix");
-            options.addOption("i", "inprogress", false, "Print inprogress log segments only");
-            options.addOption("d", "dump", false, "Dump entries of inprogress log segments");
-            options.addOption("ot", "orderbytime", false, "Order the log segments by completion time");
-            options.addOption("pso", "print-stream-only", false, "Print streams only");
-            options.addOption("cio", "check-inprogress-only", false, "Check duplicated inprogress only");
-        }
-
-        @Override
-        protected void parseCommandLine(CommandLine cmdline) throws ParseException {
-            super.parseCommandLine(cmdline);
-            if (cmdline.hasOption("t")) {
-                numThreads = Integer.parseInt(cmdline.getOptionValue("t"));
-            }
-            if (cmdline.hasOption("ft")) {
-                streamPrefix = cmdline.getOptionValue("ft");
-            }
-            printInprogressOnly = cmdline.hasOption("i");
-            dumpEntries = cmdline.hasOption("d");
-            orderByTime = cmdline.hasOption("ot");
-            printStreamsOnly = cmdline.hasOption("pso");
-            checkInprogressOnly = cmdline.hasOption("cio");
-        }
-
-        @Override
-        protected int runCmd() throws Exception {
-            SortedMap<String, List<Pair<LogSegmentMetadata, List<String>>>> corruptedCandidates =
-                    new TreeMap<String, List<Pair<LogSegmentMetadata, List<String>>>>();
-            inspectStreams(corruptedCandidates);
-            System.out.println("Corrupted Candidates : ");
-            if (printStreamsOnly) {
-                System.out.println(corruptedCandidates.keySet());
-                return 0;
-            }
-            for (Map.Entry<String, List<Pair<LogSegmentMetadata, List<String>>>> entry : corruptedCandidates.entrySet()) {
-                System.out.println(entry.getKey() + " : \n");
-                for (Pair<LogSegmentMetadata, List<String>> pair : entry.getValue()) {
-                    System.out.println("\t - " + pair.getLeft());
-                    if (printInprogressOnly && dumpEntries) {
-                        int i = 0;
-                        for (String entryData : pair.getRight()) {
-                            System.out.println("\t" + i + "\t: " + entryData);
-                            ++i;
-                        }
-                    }
-                }
-                System.out.println();
-            }
-            return 0;
-        }
-
-        private void inspectStreams(final SortedMap<String, List<Pair<LogSegmentMetadata, List<String>>>> corruptedCandidates)
-                throws Exception {
-            Iterator<String> streamCollection = getNamespace().getLogs();
-            final List<String> streams = new ArrayList<String>();
-            while (streamCollection.hasNext()) {
-                String s = streamCollection.next();
-                if (null != streamPrefix) {
-                    if (s.startsWith(streamPrefix)) {
-                        streams.add(s);
-                    }
-                } else {
-                    streams.add(s);
-                }
-            }
-            if (0 == streams.size()) {
-                return;
-            }
-            println("Streams : " + streams);
-            if (!getForce() && !IOUtils.confirmPrompt("Are you sure you want to inspect " + streams.size() + " streams")) {
-                return;
-            }
-            numThreads = Math.min(streams.size(), numThreads);
-            final int numStreamsPerThreads = streams.size() / numThreads;
-            Thread[] threads = new Thread[numThreads];
-            for (int i = 0; i < numThreads; i++) {
-                final int tid = i;
-                threads[i] = new Thread("Inspect-" + i) {
-                    @Override
-                    public void run() {
-                        try {
-                            inspectStreams(streams, tid, numStreamsPerThreads, corruptedCandidates);
-                            System.out.println("Thread " + tid + " finished.");
-                        } catch (Exception e) {
-                            System.err.println("Thread " + tid + " quits with exception : " + e.getMessage());
-                        }
-                    }
-                };
-                threads[i].start();
-            }
-            for (int i = 0; i < numThreads; i++) {
-                threads[i].join();
-            }
-        }
-
-        private void inspectStreams(List<String> streams,
-                                    int tid,
-                                    int numStreamsPerThreads,
-                                    SortedMap<String, List<Pair<LogSegmentMetadata, List<String>>>> corruptedCandidates)
-                throws Exception {
-            int startIdx = tid * numStreamsPerThreads;
-            int endIdx = Math.min(streams.size(), (tid + 1) * numStreamsPerThreads);
-            for (int i = startIdx; i < endIdx; i++) {
-                String s = streams.get(i);
-                BookKeeperClient bkc = getBookKeeperClient();
-                DistributedLogManager dlm = getNamespace().openLog(s);
-                try {
-                    List<LogSegmentMetadata> segments = dlm.getLogSegments();
-                    if (segments.size() <= 1) {
-                        continue;
-                    }
-                    boolean isCandidate = false;
-                    if (checkInprogressOnly) {
-                        Set<Long> inprogressSeqNos = new HashSet<Long>();
-                        for (LogSegmentMetadata segment : segments) {
-                            if (segment.isInProgress()) {
-                                inprogressSeqNos.add(segment.getLogSegmentSequenceNumber());
-                            }
-                        }
-                        for (LogSegmentMetadata segment : segments) {
-                            if (!segment.isInProgress() && inprogressSeqNos.contains(segment.getLogSegmentSequenceNumber())) {
-                                isCandidate = true;
-                            }
-                        }
-                    } else {
-                        LogSegmentMetadata firstSegment = segments.get(0);
-                        long lastSeqNo = firstSegment.getLogSegmentSequenceNumber();
-
-                        for (int j = 1; j < segments.size(); j++) {
-                            LogSegmentMetadata nextSegment = segments.get(j);
-                            if (lastSeqNo + 1 != nextSegment.getLogSegmentSequenceNumber()) {
-                                isCandidate = true;
-                                break;
-                            }
-                            ++lastSeqNo;
-                        }
-                    }
-                    if (isCandidate) {
-                        if (orderByTime) {
-                            Collections.sort(segments, LOGSEGMENT_COMPARATOR_BY_TIME);
-                        }
-                        List<Pair<LogSegmentMetadata, List<String>>> ledgers =
-                                new ArrayList<Pair<LogSegmentMetadata, List<String>>>();
-                        for (LogSegmentMetadata seg : segments) {
-                            LogSegmentMetadata segment = seg;
-                            List<String> dumpedEntries = new ArrayList<String>();
-                            if (segment.isInProgress()) {
-                                LedgerHandle lh = bkc.get().openLedgerNoRecovery(segment.getLogSegmentId(), BookKeeper.DigestType.CRC32,
-                                                                                 dlConf.getBKDigestPW().getBytes(UTF_8));
-                                try {
-                                    long lac = lh.readLastConfirmed();
-                                    segment = segment.mutator().setLastEntryId(lac).build();
-                                    if (printInprogressOnly && dumpEntries && lac >= 0) {
-                                        Enumeration<LedgerEntry> entries = lh.readEntries(0L, lac);
-                                        while (entries.hasMoreElements()) {
-                                            LedgerEntry entry = entries.nextElement();
-                                            dumpedEntries.add(new String(entry.getEntry(), UTF_8));
-                                        }
-                                    }
-                                } finally {
-                                    lh.close();
-                                }
-                            }
-                            if (printInprogressOnly) {
-                                if (segment.isInProgress()) {
-                                    ledgers.add(Pair.of(segment, dumpedEntries));
-                                }
-                            } else {
-                                ledgers.add(Pair.of(segment, EMPTY_LIST));
-                            }
-                        }
-                        synchronized (corruptedCandidates) {
-                            corruptedCandidates.put(s, ledgers);
-                        }
-                    }
-                } finally {
-                    dlm.close();
-                }
-            }
-        }
-
-        @Override
-        protected String getUsage() {
-            return "inspect [options]";
-        }
-    }
-
-    protected static class TruncateCommand extends PerDLCommand {
-
-        int numThreads = 1;
-        String streamPrefix = null;
-        boolean deleteStream = false;
-
-        TruncateCommand() {
-            super("truncate", "truncate streams under a given dl uri");
-            options.addOption("t", "threads", true, "Number threads to do truncation");
-            options.addOption("ft", "filter", true, "Stream filter by prefix");
-            options.addOption("d", "delete", false, "Delete Stream");
-        }
-
-        @Override
-        protected void parseCommandLine(CommandLine cmdline) throws ParseException {
-            super.parseCommandLine(cmdline);
-            if (cmdline.hasOption("t")) {
-                numThreads = Integer.parseInt(cmdline.getOptionValue("t"));
-            }
-            if (cmdline.hasOption("ft")) {
-                streamPrefix = cmdline.getOptionValue("ft");
-            }
-            if (cmdline.hasOption("d")) {
-                deleteStream = true;
-            }
-        }
-
-        @Override
-        protected String getUsage() {
-            return "truncate [options]";
-        }
-
-        protected void setFilter(String filter) {
-            this.streamPrefix = filter;
-        }
-
-        @Override
-        protected int runCmd() throws Exception {
-            getConf().setZkAclId(getZkAclId());
-            return truncateStreams(getNamespace());
-        }
-
-        private int truncateStreams(final DistributedLogNamespace namespace) throws Exception {
-            Iterator<String> streamCollection = namespace.getLogs();
-            final List<String> streams = new ArrayList<String>();
-            while (streamCollection.hasNext()) {
-                String s = streamCollection.next();
-                if (null != streamPrefix) {
-                    if (s.startsWith(streamPrefix)) {
-                        streams.add(s);
-                    }
-                } else {
-                    streams.add(s);
-                }
-            }
-            if (0 == streams.size()) {
-                return 0;
-            }
-            System.out.println("Streams : " + streams);
-            if (!getForce() && !IOUtils.confirmPrompt("Do you want to truncate " + streams.size() + " streams ?")) {
-                return 0;
-            }
-            numThreads = Math.min(streams.size(), numThreads);
-            final int numStreamsPerThreads = streams.size() / numThreads + 1;
-            Thread[] threads = new Thread[numThreads];
-            for (int i = 0; i < numThreads; i++) {
-                final int tid = i;
-                threads[i] = new Thread("Truncate-" + i) {
-                    @Override
-                    public void run() {
-                        try {
-                            truncateStreams(namespace, streams, tid, numStreamsPerThreads);
-                            System.out.println("Thread " + tid + " finished.");
-                        } catch (IOException e) {
-                            System.err.println("Thread " + tid + " quits with exception : " + e.getMessage());
-                        }
-                    }
-                };
-                threads[i].start();
-            }
-            for (int i = 0; i < numThreads; i++) {
-                threads[i].join();
-            }
-            return 0;
-        }
-
-        private void truncateStreams(DistributedLogNamespace namespace, List<String> streams,
-                                     int tid, int numStreamsPerThreads) throws IOException {
-            int startIdx = tid * numStreamsPerThreads;
-            int endIdx = Math.min(streams.size(), (tid + 1) * numStreamsPerThreads);
-            for (int i = startIdx; i < endIdx; i++) {
-                String s = streams.get(i);
-                DistributedLogManager dlm = namespace.openLog(s);
-                try {
-                    if (deleteStream) {
-                        dlm.delete();
-                    } else {
-                        dlm.purgeLogsOlderThan(Long.MAX_VALUE);
-                    }
-                } finally {
-                    dlm.close();
-                }
-            }
-        }
-    }
-
-    public static class SimpleBookKeeperClient {
-        BookKeeperClient bkc;
-        ZooKeeperClient zkc;
-
-        public SimpleBookKeeperClient(DistributedLogConfiguration conf, URI uri) {
-            try {
-                zkc = ZooKeeperClientBuilder.newBuilder()
-                    .sessionTimeoutMs(conf.getZKSessionTimeoutMilliseconds())
-                    .zkAclId(conf.getZkAclId())
-                    .uri(uri)
-                    .build();
-                BKDLConfig bkdlConfig = BKDLConfig.resolveDLConfig(zkc, uri);
-                BKDLConfig.propagateConfiguration(bkdlConfig, conf);
-                bkc = BookKeeperClientBuilder.newBuilder()
-                        .zkc(zkc)
-                        .dlConfig(conf)
-                        .ledgersPath(bkdlConfig.getBkLedgersPath())
-                        .name("dlog")
-                        .build();
-            } catch (Exception e) {
-                close();
-            }
-        }
-        public BookKeeperClient client() {
-            return bkc;
-        }
-        public void close() {
-            if (null != bkc) {
-                bkc.close();
-            }
-            if (null != zkc) {
-                zkc.close();
-            }
-        }
-    }
-
-    protected static class ShowCommand extends PerStreamCommand {
-
-        SimpleBookKeeperClient bkc = null;
-        boolean listSegments = true;
-        boolean listEppStats = false;
-        long firstLid = 0;
-        long lastLid = -1;
-
-        ShowCommand() {
-            super("show", "show metadata of a given stream and list segments");
-            options.addOption("ns", "no-log-segments", false, "Do not list log segment metadata");
-            options.addOption("lp", "placement-stats", false, "Show ensemble placement stats");
-            options.addOption("fl", "first-ledger", true, "First log sement no");
-            options.addOption("ll", "last-ledger", true, "Last log sement no");
-        }
-
-        @Override
-        protected void parseCommandLine(CommandLine cmdline) throws ParseException {
-            super.parseCommandLine(cmdline);
-            if (cmdline.hasOption("fl")) {
-                try {
-                    firstLid = Long.parseLong(cmdline.getOptionValue("fl"));
-                } catch (NumberFormatException nfe) {
-                    throw new ParseException("Invalid ledger id " + cmdline.getOptionValue("fl"));
-                }
-            }
-            if (firstLid < 0) {
-                throw new IllegalArgumentException("Invalid ledger id " + firstLid);
-            }
-            if (cmdline.hasOption("ll")) {
-                try {
-                    lastLid = Long.parseLong(cmdline.getOptionValue("ll"));
-                } catch (NumberFormatException nfe) {
-                    throw new ParseException("Invalid ledger id " + cmdline.getOptionValue("ll"));
-                }
-            }
-            if (lastLid != -1 && firstLid > lastLid) {
-                throw new IllegalArgumentException("Invalid ledger ids " + firstLid + " " + lastLid);
-            }
-            listSegments = !cmdline.hasOption("ns");
-            listEppStats = cmdline.hasOption("lp");
-        }
-
-        @Override
-        protected int runCmd() throws Exception {
-            DistributedLogManager dlm = getNamespace().openLog(getStreamName());
-            try {
-                if (listEppStats) {
-                    bkc = new SimpleBookKeeperClient(getConf(), getUri());
-                }
-                printMetadata(dlm);
-            } finally {
-                dlm.close();
-                if (null != bkc) {
-                    bkc.close();
-                }
-            }
-            return 0;
-        }
-
-        private void printMetadata(DistributedLogManager dlm) throws Exception {
-            printHeader(dlm);
-            if (listSegments) {
-                System.out.println("Ledgers : ");
-                List<LogSegmentMetadata> segments = dlm.getLogSegments();
-                for (LogSegmentMetadata segment : segments) {
-                    if (include(segment)) {
-                        printLedgerRow(segment);
-                    }
-                }
-            }
-        }
-
-        private void printHeader(DistributedLogManager dlm) throws Exception {
-            DLSN firstDlsn = Await.result(dlm.getFirstDLSNAsync());
-            boolean endOfStreamMarked = dlm.isEndOfStreamMarked();
-            DLSN lastDlsn = dlm.getLastDLSN();
-            long firstTxnId = dlm.getFirstTxId();
-            long lastTxnId = dlm.getLastTxId();
-            long recordCount = dlm.getLogRecordCount();
-            String result = String.format("Stream : (firstTxId=%d, lastTxid=%d, firstDlsn=%s, lastDlsn=%s, endOfStreamMarked=%b, recordCount=%d)",
-                firstTxnId, lastTxnId, getDlsnName(firstDlsn), getDlsnName(lastDlsn), endOfStreamMarked, recordCount);
-            System.out.println(result);
-            if (listEppStats) {
-                printEppStatsHeader(dlm);
-            }
-        }
-
-        boolean include(LogSegmentMetadata segment) {
-            return (firstLid <= segment.getLogSegmentSequenceNumber() && (lastLid == -1 || lastLid >= segment.getLogSegmentSequenceNumber()));
-        }
-
-        private void printEppStatsHeader(DistributedLogManager dlm) throws Exception {
-            String label = "Ledger Placement :";
-            System.out.println(label);
-            Map<BookieSocketAddress, Integer> totals = new HashMap<BookieSocketAddress, Integer>();
-            List<LogSegmentMetadata> segments = dlm.getLogSegments();
-            for (LogSegmentMetadata segment : segments) {
-                if (include(segment)) {
-                    merge(totals, getBookieStats(segment));
-                }
-            }
-            List<Map.Entry<BookieSocketAddress, Integer>> entries = new ArrayList<Map.Entry<BookieSocketAddress, Integer>>(totals.entrySet());
-            Collections.sort(entries, new Comparator<Map.Entry<BookieSocketAddress, Integer>>() {
-                @Override
-                public int compare(Map.Entry<BookieSocketAddress, Integer> o1, Map.Entry<BookieSocketAddress, Integer> o2) {
-                    return o2.getValue() - o1.getValue();
-                }
-            });
-            int width = 0;
-            int totalEntries = 0;
-            for (Map.Entry<BookieSocketAddress, Integer> entry : entries) {
-                width = Math.max(width, label.length() + 1 + entry.getKey().toString().length());
-                totalEntries += entry.getValue();
-            }
-            for (Map.Entry<BookieSocketAddress, Integer> entry : entries) {
-                System.out.println(String.format("%"+width+"s\t%6.2f%%\t\t%d", entry.getKey(), entry.getValue()*1.0/totalEntries, entry.getValue()));
-            }
-        }
-
-        private void printLedgerRow(LogSegmentMetadata segment) throws Exception {
-            System.out.println(segment.getLogSegmentSequenceNumber() + "\t: " + segment);
-        }
-
-        private Map<BookieSocketAddress, Integer> getBookieStats(LogSegmentMetadata segment) throws Exception {
-            Map<BookieSocketAddress, Integer> stats = new HashMap<BookieSocketAddress, Integer>();
-            LedgerHandle lh = bkc.client().get().openLedgerNoRecovery(segment.getLogSegmentId(), BookKeeper.DigestType.CRC32,
-                    getConf().getBKDigestPW().getBytes(UTF_8));
-            long eidFirst = 0;
-            for (SortedMap.Entry<Long, ArrayList<BookieSocketAddress>> entry : LedgerReader.bookiesForLedger(lh).entrySet()) {
-                long eidLast = entry.getKey().longValue();
-                long count = eidLast - eidFirst + 1;
-                for (BookieSocketAddress bookie : entry.getValue()) {
-                    merge(stats, bookie, (int) count);
-                }
-                eidFirst = eidLast;
-            }
-            return stats;
-        }
-
-        void merge(Map<BookieSocketAddress, Integer> m, BookieSocketAddress bookie, Integer count) {
-            if (m.containsKey(bookie)) {
-                m.put(bookie, count + m.get(bookie).intValue());
-            } else {
-                m.put(bookie, count);
-            }
-        }
-
-        void merge(Map<BookieSocketAddress, Integer> m1, Map<BookieSocketAddress, Integer> m2) {
-            for (Map.Entry<BookieSocketAddress, Integer> entry : m2.entrySet()) {
-                merge(m1, entry.getKey(), entry.getValue());
-            }
-        }
-
-        String getDlsnName(DLSN dlsn) {
-            if (dlsn.equals(DLSN.InvalidDLSN)) {
-                return "InvalidDLSN";
-            }
-            return dlsn.toString();
-        }
-
-        @Override
-        protected String getUsage() {
-            return "show [options]";
-        }
-    }
-
-    static class CountCommand extends PerStreamCommand {
-
-        DLSN startDLSN = null;
-        DLSN endDLSN = null;
-
-        protected CountCommand() {
-            super("count", "count number records between dlsns");
-        }
-
-        @Override
-        protected void parseCommandLine(CommandLine cmdline) throws ParseException {
-            super.parseCommandLine(cmdline);
-            String[] args = cmdline.getArgs();
-            if (args.length < 1) {
-                throw new ParseException("Must specify at least start dlsn.");
-            }
-            if (args.length >= 1) {
-                startDLSN = parseDLSN(args[0]);
-            }
-            if (args.length >= 2) {
-                endDLSN = parseDLSN(args[1]);
-            }
-        }
-
-        @Override
-        protected int runCmd() throws Exception {
-            DistributedLogManager dlm = getNamespace().openLog(getStreamName());
-            try {
-                long count = 0;
-                if (null == endDLSN) {
-                    count = countToLastRecord(dlm);
-                } else {
-                    count = countFromStartToEnd(dlm);
-                }
-                System.out.println("total is " + count + " records.");
-                return 0;
-            } finally {
-                dlm.close();
-            }
-        }
-
-        int countFromStartToEnd(DistributedLogManager dlm) throws Exception {
-            int count = 0;
-            try {
-                LogReader reader = dlm.getInputStream(startDLSN);
-                try {
-                    LogRecordWithDLSN record = reader.readNext(false);
-                    LogRecordWithDLSN preRecord = record;
-                    System.out.println("first record : " + record);
-                    while (null != record) {
-                        if (record.getDlsn().compareTo(endDLSN) > 0) {
-                            break;
-                        }
-                        ++count;
-                        if (count % 1000 == 0) {
-                            logger.info("read {} records from {}...", count, getStreamName());
-                        }
-                        preRecord = record;
-                        record = reader.readNext(false);
-                    }
-                    System.out.println("last record : " + preRecord);
-                } finally {
-                    reader.close();
-                }
-            } finally {
-                dlm.close();
-            }
-            return count;
-        }
-
-        long countToLastRecord(DistributedLogManager dlm) throws Exception {
-            return Await.result(dlm.getLogRecordCountAsync(startDLSN)).longValue();
-        }
-
-        @Override
-        protected String getUsage() {
-            return "count <start> <end>";
-        }
-    }
-
-    public static class DeleteCommand extends PerStreamCommand {
-
-        protected DeleteCommand() {
-            super("delete", "delete a given stream");
-        }
-
-        @Override
-        protected int runCmd() throws Exception {
-            getConf().setZkAclId(getZkAclId());
-            DistributedLogManager dlm = getNamespace().openLog(getStreamName());
-            try {
-                dlm.delete();
-            } finally {
-                dlm.close();
-            }
-            return 0;
-        }
-
-        @Override
-        protected String getUsage() {
-            return "delete";
-        }
-    }
-
-    public static class DeleteLedgersCommand extends PerDLCommand {
-
-        private final List<Long> ledgers = new ArrayList<Long>();
-
-        int numThreads = 1;
-
-        protected DeleteLedgersCommand() {
-            super("delete_ledgers", "delete given ledgers");
-            options.addOption("l", "ledgers", true, "List of ledgers, separated by comma");
-            options.addOption("lf", "ledgers-file", true, "File of list of ledgers, each line has a ledger id");
-            options.addOption("t", "concurrency", true, "Number of threads to run deletions");
-        }
-
-        @Override
-        protected void parseCommandLine(CommandLine cmdline) throws ParseException {
-            super.parseCommandLine(cmdline);
-            if (cmdline.hasOption("l") && cmdline.hasOption("lf")) {
-                throw new ParseException("Please specify ledgers: either use list or use file only.");
-            }
-            if (!cmdline.hasOption("l") && !cmdline.hasOption("lf")) {
-                throw new ParseException("No ledgers specified. Please specify ledgers either use list or use file only.");
-            }
-            if (cmdline.hasOption("l")) {
-                String ledgersStr = cmdline.getOptionValue("l");
-                String[] ledgerStrs = ledgersStr.split(",");
-                for (String ledgerStr : ledgerStrs) {
-                    ledgers.add(Long.parseLong(ledgerStr));
-                }
-            }
-            if (cmdline.hasOption("lf")) {
-                BufferedReader br = null;
-                try {
-
-                    br = new BufferedReader(new InputStreamReader(
-                            new FileInputStream(new File(cmdline.getOptionValue("lf"))), UTF_8.name()));
-                    String line;
-                    while ((line = br.readLine()) != null) {
-                        ledgers.add(Long.parseLong(line));
-                    }
-                } catch (FileNotFoundException e) {
-                    throw new ParseException("No ledgers file " + cmdline.getOptionValue("lf") + " found.");
-                } catch (IOException e) {
-                    throw new ParseException("Invalid ledgers file " + cmdline.getOptionValue("lf") + " found.");
-                } finally {
-                    if (null != br) {
-                        try {
-                            br.close();
-                        } catch (IOException e) {
-                            // no-op
-                        }
-                    }
-                }
-            }
-            if (cmdline.hasOption("t")) {
-                numThreads = Integer.parseInt(cmdline.getOptionValue("t"));
-            }
-        }
-
-        @Override
-        protected String getUsage() {
-            return "delete_ledgers [options]";
-        }
-
-        @Override
-        protected int runCmd() throws Exception {
-            ExecutorService executorService = Executors.newFixedThreadPool(numThreads);
-            try {
-                final AtomicInteger numLedgers = new AtomicInteger(0);
-                final CountDownLatch doneLatch = new CountDownLatch(numThreads);
-                final AtomicInteger numFailures = new AtomicInteger(0);
-                final LinkedBlockingQueue<Long> ledgerQueue =
-                        new LinkedBlockingQueue<Long>();
-                ledgerQueue.addAll(ledgers);
-                for (int i = 0; i < numThreads; i++) {
-                    final int tid = i;
-                    executorService.submit(new Runnable() {
-                        @Override
-                        public void run() {
-                            while (true) {
-                                Long ledger = ledgerQueue.poll();
-                                if (null == ledger) {
-                                    break;
-                                }
-                                try {
-                                    getBookKeeperClient().get().deleteLedger(ledger);
-                                    int numLedgersDeleted = numLedgers.incrementAndGet();
-                                    if (numLedgersDeleted % 1000 == 0) {
-                                        System.out.println("Deleted " + numLedgersDeleted + " ledgers.");
-                                    }
-                                } catch (BKException.BKNoSuchLedgerExistsException e) {
-                                    int numLedgersDeleted = numLedgers.incrementAndGet();
-                                    if (numLedgersDeleted % 1000 == 0) {
-                                        System.out.println("Deleted " + numLedgersDeleted + " ledgers.");
-                                    }
-                                } catch (Exception e) {
-                                    numFailures.incrementAndGet();
-                                    break;
-                                }
-                            }
-                            doneLatch.countDown();
-                            System.out.println("Thread " + tid + " quits");
-                        }
-                    });
-                }
-                doneLatch.await();
-                if (numFailures.get() > 0) {
-                    throw new IOException("Encounter " + numFailures.get() + " failures during deleting ledgers");
-                }
-            } finally {
-                executorService.shutdown();
-            }
-            return 0;
-        }
-    }
-
-    public static class CreateCommand extends PerDLCommand {
-
-        final List<String> streams = new ArrayList<String>();
-
-        String streamPrefix = null;
-        String streamExpression = null;
-
-        CreateCommand() {
-            super("create", "create streams under a given namespace");
-            options.addOption("r", "prefix", true, "Prefix of stream name. E.g. 'QuantumLeapTest-'.");
-            options.addOption("e", "expression", true, "Expression to generate stream suffix. " +
-                              "Currently we support range 'x-y', list 'x,y,z' and name 'xyz'");
-        }
-
-        @Override
-        protected void parseCommandLine(CommandLine cmdline) throws ParseException {
-            super.parseCommandLine(cmdline);
-            if (cmdline.hasOption("r")) {
-                streamPrefix = cmdline.getOptionValue("r");
-            }
-            if (cmdline.hasOption("e")) {
-                streamExpression = cmdline.getOptionValue("e");
-            }
-            if (null == streamPrefix || null == streamExpression) {
-                throw new ParseException("Please specify stream prefix & expression.");
-            }
-        }
-
-        protected void generateStreams(String streamPrefix, String streamExpression) throws ParseException {
-            // parse the stream expression
-            if (streamExpression.contains("-")) {
-                // a range expression
-                String[] parts = streamExpression.split("-");
-                if (parts.length != 2) {
-                    throw new ParseException("Invalid stream index range : " + streamExpression);
-                }
-                try {
-                    int start = Integer.parseInt(parts[0]);
-                    int end = Integer.parseInt(parts[1]);
-                    if (start > end) {
-                        throw new ParseException("Invalid stream index range : " + streamExpression);
-                    }
-                    for (int i = start; i <= end; i++) {
-                        streams.add(streamPrefix + i);
-                    }
-                } catch (NumberFormatException nfe) {
-                    throw new ParseException("Invalid stream index range : " + streamExpression);
-                }
-            } else if (streamExpression.contains(",")) {
-                // a list expression
-                String[] parts = streamExpression.split(",");
-                try {
-                    for (String part : parts) {
-                        int idx = Integer.parseInt(part);
-                        streams.add(streamPrefix + idx);
-                    }
-                } catch (NumberFormatException nfe) {
-                    throw new ParseException("Invalid stream suffix list : " + streamExpression);
-                }
-            } else {
-                streams.add(streamPrefix + streamExpression);
-            }
-        }
-
-        @Override
-        protected int runCmd() throws Exception {
-            generateStreams(streamPrefix, streamExpression);
-            if (streams.isEmpty()) {
-                System.out.println("Nothing to create.");
-                return 0;
-            }
-            if (!getForce() && !IOUtils.confirmPrompt("You are going to create streams : " + streams)) {
-                return 0;
-            }
-            getConf().setZkAclId(getZkAclId());
-            for (String stream : streams) {
-                getNamespace().createLog(stream);
-            }
-            return 0;
-        }
-
-        @Override
-        protected String getUsage() {
-            return "create [options]";
-        }
-
-        protected void setPrefix(String prefix) {
-            this.streamPrefix = prefix;
-        }
-
-        protected void setExpression(String expression) {
-            this.streamExpression = expression;
-        }
-    }
-
-    protected static class DumpCommand extends PerStreamCommand {
-
-        boolean printHex = false;
-        boolean skipPayload = false;
-        Long fromTxnId = null;
-        DLSN fromDLSN = null;
-        int count = 100;
-
-        DumpCommand() {
-            super("dump", "dump records of a given stream");
-            options.addOption("x", "hex", false, "Print record in hex format");
-            options.addOption("sp", "skip-payload", false, "Skip printing the payload of the record");
-            options.addOption("o", "offset", true, "Txn ID to start dumping.");
-            options.addOption("n", "seqno", true, "Sequence Number to start dumping");
-            options.addOption("e", "eid", true, "Entry ID to start dumping");
-            options.addOption("t", "slot", true, "Slot to start dumping");
-            options.addOption("l", "limit", true, "Number of entries to dump. Default is 100.");
-        }
-
-        @Override
-        protected void parseCommandLine(CommandLine cmdline) throws ParseException {
-            super.parseCommandLine(cmdline);
-            printHex = cmdline.hasOption("x");
-            skipPayload = cmdline.hasOption("sp");
-            if (cmdline.hasOption("o")) {
-                try {
-                    fromTxnId = Long.parseLong(cmdline.getOptionValue("o"));
-                } catch (NumberFormatException nfe) {
-                    throw new ParseException("Invalid txn id " + cmdline.getOptionValue("o"));
-                }
-            }
-            if (cmdline.hasOption("l")) {
-                try {
-                    count = Integer.parseInt(cmdline.getOptionValue("l"));
-                } catch (NumberFormatException nfe) {
-                    throw new ParseException("Invalid count " + cmdline.getOptionValue("l"));
-                }
-                if (count <= 0) {
-                    throw new ParseException("Negative count found : " + count);
-                }
-            }
-            if (cmdline.hasOption("n")) {
-                long seqno;
-                try {
-                    seqno = Long.parseLong(cmdline.getOptionValue("n"));
-                } catch (NumberFormatException nfe) {
-                    throw new ParseException("Invalid sequence number " + cmdline.getOptionValue("n"));
-                }
-                long eid;
-                if (cmdline.hasOption("e")) {
-                    eid = Long.parseLong(cmdline.getOptionValue("e"));
-                } else {
-                    eid = 0;
-                }
-                long slot;
-                if (cmdline.hasOption("t")) {
-                    slot = Long.parseLong(cmdline.getOptionValue("t"));
-                } else {
-                    slot = 0;
-                }
-                fromDLSN = new DLSN(seqno, eid, slot);
-            }
-            if (null == fromTxnId && null == fromDLSN) {
-                throw new ParseException("No start Txn/DLSN is specified.");
-            }
-        }
-
-        @Override
-        protected int runCmd() throws Exception {
-            DistributedLogManager dlm = getNamespace().openLog(getStreamName());
-            long totalCount = dlm.getLogRecordCount();
-            try {
-                AsyncLogReader reader;
-                Object startOffset;
-                try {
-                    DLSN lastDLSN = Await.result(dlm.getLastDLSNAsync());
-                    System.out.println("Last DLSN : " + lastDLSN);
-                    if (null == fromDLSN) {
-                        reader = dlm.getAsyncLogReader(fromTxnId);
-                        startOffset = fromTxnId;
-                    } else {
-                        reader = dlm.getAsyncLogReader(fromDLSN);
-                        startOffset = fromDLSN;
-                    }
-                } catch (LogNotFoundException lee) {
-                    System.out.println("No stream found to dump records.");
-                    return 0;
-                }
-                try {
-                    System.out.println(String.format("Dump records for %s (from = %s, dump count = %d, total records = %d)",
-                            getStreamName(), startOffset, count, totalCount));
-
-                    dumpRecords(reader);
-                } finally {
-                    Utils.close(reader);
-                }
-            } finally {
-                dlm.close();
-            }
-            return 0;
-        }
-
-        private void dumpRecords(AsyncLogReader reader) throws Exception {
-            int numRead = 0;
-            LogRecord record = Await.result(reader.readNext());
-            while (record != null) {
-                // dump the record
-                dumpRecord(record);
-                ++numRead;
-                if (numRead >= count) {
-                    break;
-                }
-                record = Await.result(reader.readNext());
-            }
-            if (numRead == 0) {
-                System.out.println("No records.");
-            } else {
-                System.out.println("------------------------------------------------");
-            }
-        }
-
-        private void dumpRecord(LogRecord record) {
-            System.out.println("------------------------------------------------");
-            if (record instanceof LogRecordWithDLSN) {
-                System.out.println("Record (txn = " + record.getTransactionId() + ", bytes = "
-                        + record.getPayload().length + ", dlsn = "
-                        + ((LogRecordWithDLSN) record).getDlsn() + ", sequence id = "
-                        + ((LogRecordWithDLSN) record).getSequenceId() + ")");
-            } else {
-                System.out.println("Record (txn = " + record.getTransactionId() + ", bytes = "
-                        + record.getPayload().length + ")");
-            }
-            System.out.println("");
-
-            if (skipPayload) {
-                return;
-            }
-
-            if (printHex) {
-                System.out.println(Hex.encodeHexString(record.getPayload()));
-            } else {
-                System.out.println(new String(record.getPayload(), UTF_8));
-            }
-        }
-
-        @Override
-        protected String getUsage() {
-            return "dump [options]";
-        }
-
-        protected void setFromTxnId(Long fromTxnId) {
-            this.fromTxnId = fromTxnId;
-        }
-    }
-
-    /**
-     * TODO: refactor inspect & inspectstream
-     * TODO: support force
-     *
-     * inspectstream -lac -gap (different options for different operations for a single stream)
-     * inspect -lac -gap (inspect the namespace, which will use inspect stream)
-     */
-    static class InspectStreamCommand extends PerStreamCommand {
-
-        InspectStreamCommand() {
-            super("inspectstream", "Inspect a given stream to identify any metadata corruptions");
-        }
-
-        @Override
-        protected int runCmd() throws Exception {
-            DistributedLogManager dlm = getNamespace().openLog(getStreamName());
-            try {
-                return inspectAndRepair(dlm.getLogSegments());
-            } finally {
-                dlm.close();
-            }
-        }
-
-        protected int inspectAndRepair(List<LogSegmentMetadata> segments) throws Exception {
-            LogSegmentMetadataStore metadataStore = getLogSegmentMetadataStore();
-            ZooKeeperClient zkc = getZooKeeperClient();
-            BKDLConfig bkdlConfig = BKDLConfig.resolveDLConfig(zkc, getUri());
-            BKDLConfig.propagateConfiguration(bkdlConfig, getConf());
-            BookKeeperClient bkc = BookKeeperClientBuilder.newBuilder()
-                    .dlConfig(getConf())
-                    .zkServers(bkdlConfig.getBkZkServersForReader())
-                    .ledgersPath(bkdlConfig.getBkLedgersPath())
-                    .name("dlog")
-                    .build();
-            try {
-                List<LogSegmentMetadata> segmentsToRepair = inspectLogSegments(bkc, segments);
-                if (segmentsToRepair.isEmpty()) {
-                    System.out.println("The stream is good. No log segments to repair.");
-                    return 0;
-                }
-                System.out.println(segmentsToRepair.size() + " segments to repair : ");
-                System.out.println(segmentsToRepair);
-                System.out.println();
-                if (!IOUtils.confirmPrompt("Do you want to repair them (Y/N): ")) {
-                    return 0;
-                }
-                repairLogSegments(metadataStore, bkc, segmentsToRepair);
-                return 0;
-            } finally {
-                bkc.close();
-            }
-        }
-
-        protected List<LogSegmentMetadata> inspectLogSegments(
-                BookKeeperClient bkc, List<LogSegmentMetadata> segments) throws Exception {
-            List<LogSegmentMetadata> segmentsToRepair = new ArrayList<LogSegmentMetadata>();
-            for (LogSegmentMetadata segment : segments) {
-                if (!segment.isInProgress() && !inspectLogSegment(bkc, segment)) {
-                    segmentsToRepair.add(segment);
-                }
-            }
-            return segmentsToRepair;
-        }
-
-        /**
-         * Inspect a given log segment.
-         *
-         * @param bkc
-         *          bookkeeper client
-         * @param metadata
-         *          metadata of the log segment to
-         * @return true if it is a good stream, false if the stream has inconsistent metadata.
-         * @throws Exception
-         */
-        protected boolean inspectLogSegment(BookKeeperClient bkc,
-                                            LogSegmentMetadata metadata) throws Exception {
-            if (metadata.isInProgress()) {
-                System.out.println("Skip inprogress log segment " + metadata);
-                return true;
-            }
-            long ledgerId = metadata.getLogSegmentId();
-            LedgerHandle lh = bkc.get().openLedger(ledgerId, BookKeeper.DigestType.CRC32,
-                    getConf().getBKDigestPW().getBytes(UTF_8));
-            LedgerHandle readLh = bkc.get().openLedger(ledgerId, BookKeeper.DigestType.CRC32,
-                    getConf().getBKDigestPW().getBytes(UTF_8));
-            LedgerReader lr = new LedgerReader(bkc.get());
-            final AtomicReference<List<LedgerEntry>> entriesHolder = new AtomicReference<List<LedgerEntry>>(null);
-            final AtomicInteger rcHolder = new AtomicInteger(-1234);
-            final CountDownLatch doneLatch = new CountDownLatch(1);
-            try {
-                lr.forwardReadEntriesFromLastConfirmed(readLh, new BookkeeperInternalCallbacks.GenericCallback<List<LedgerEntry>>() {
-                    @Override
-                    public void operationComplete(int rc, List<LedgerEntry> entries) {
-                        rcHolder.set(rc);
-                        entriesHolder.set(entries);
-                        doneLatch.countDown();
-                    }
-                });
-                doneLatch.await();
-                if (BKException.Code.OK != rcHolder.get()) {
-                    throw BKException.create(rcHolder.get());
-                }
-                List<LedgerEntry> entries = entriesHolder.get();
-                long lastEntryId;
-                if (entries.isEmpty()) {
-                    lastEntryId = LedgerHandle.INVALID_ENTRY_ID;
-                } else {
-                    LedgerEntry lastEntry = entries.get(entries.size() - 1);
-                    lastEntryId = lastEntry.getEntryId();
-                }
-                if (lastEntryId != lh.getLastAddConfirmed()) {
-                    System.out.println("Inconsistent Last Add Confirmed Found for LogSegment " + metadata.getLogSegmentSequenceNumber() + ": ");
-                    System.out.println("\t metadata: " + metadata);
-                    System.out.println("\t lac in ledger metadata is " + lh.getLastAddConfirmed() + ", but lac in bookies is " + lastEntryId);
-                    return false;
-                } else {
-                    return true;
-                }
-            } finally {
-                lh.close();
-                readLh.close();
-            }
-        }
-
-        protected void repairLogSegments(LogSegmentMetadataStore metadataStore,
-                                         BookKeeperClient bkc,
-                                         List<LogSegmentMetadata> segments) throws Exception {
-            BookKeeperAdmin bkAdmin = new BookKeeperAdmin(bkc.get());
-            try {
-                MetadataUpdater metadataUpdater = LogSegmentMetadataStoreUpdater.createMetadataUpdater(
-                        getConf(), metadataStore);
-                for (LogSegmentMetadata segment : segments) {
-                    repairLogSegment(bkAdmin, metadataUpdater, segment);
-                }
-            } finally {
-                bkAdmin.close();
-            }
-        }
-
-        protected void repairLogSegment(BookKeeperAdmin bkAdmin,
-                                        MetadataUpdater metadataUpdater,
-                                        LogSegmentMetadata segment) throws Exception {
-            if (segment.isInProgress()) {
-                System.out.println("Skip inprogress log segment " + segment);
-                return;
-            }
-            LedgerHandle lh = bkAdmin.openLedger(segment.getLogSegmentId(), true);
-            long lac = lh.getLastAddConfirmed();
-            Enumeration<LedgerEntry> entries = lh.readEntries(lac, lac);
-            if (!entries.hasMoreElements()) {
-                throw new IOException("Entry " + lac + " isn't found for " + segment);
-            }
-            LedgerEntry lastEntry = entries.nextElement();
-            Entry.Reader reader = Entry.newBuilder()
-                    .setLogSegmentInfo(segment.getLogSegmentSequenceNumber(), segment.getStartSequenceId())
-                    .setEntryId(lastEntry.getEntryId())
-                    .setEnvelopeEntry(LogSegmentMetadata.supportsEnvelopedEntries(segment.getVersion()))
-                    .setInputStream(lastEntry.getEntryInputStream())
-                    .buildReader();
-            LogRecordWithDLSN record = reader.nextRecord();
-            LogRecordWithDLSN lastRecord = null;
-            while (null != record) {
-                lastRecord = record;
-                record = reader.nextRecord();
-            }
-            if (null == lastRecord) {
-                throw new IOException("No record found in entry " + lac + " for " + segment);
-            }
-            System.out.println("Updating last record for " + segment + " to " + lastRecord);
-            if (!IOUtils.confirmPrompt("Do you want to make this change (Y/N): ")) {
-                return;
-            }
-            metadataUpdater.updateLastRecord(segment, lastRecord);
-        }
-
-        @Override
-        protected String getUsage() {
-            return "inspectstream [options]";
-        }
-    }
-
-    static interface BKCommandRunner {
-        int run(ZooKeeperClient zkc, BookKeeperClient bkc) throws Exception;
-    }
-
-    abstract static class PerBKCommand extends PerDLCommand {
-
-        protected PerBKCommand(String name, String description) {
-            super(name, description);
-        }
-
-        @Override
-        protected int runCmd() throws Exception {
-            return runBKCommand(new BKCommandRunner() {
-                @Override
-                public int run(ZooKeeperClient zkc, BookKeeperClient bkc) throws Exception {
-                    return runBKCmd(zkc, bkc);
-                }
-            });
-        }
-
-        protected int runBKCommand(BKCommandRunner runner) throws Exception {
-            return runner.run(getZooKeeperClient(), getBookKeeperClient());
-        }
-
-        abstract protected int runBKCmd(ZooKeeperClient zkc, BookKeeperClient bkc) throws Exception;
-    }
-
-    static class RecoverCommand extends PerBKCommand {
-
-        final List<Long> ledgers = new ArrayList<Long>();
-        boolean query = false;
-        boolean dryrun = false;
-        boolean skipOpenLedgers = false;
-        boolean fenceOnly = false;
-        int fenceRate = 1;
-        int concurrency = 1;
-        final Set<BookieSocketAddress> bookiesSrc = new HashSet<BookieSocketAddress>();
-        int partition = 0;
-        int numPartitions = 0;
-
-        RecoverCommand() {
-            super("recover", "Recover the ledger data that stored on failed bookies");
-            options.addOption("l", "ledger", true, "Specific ledger to recover");
-            options.addOption("lf", "ledgerfile", true, "File contains ledgers list");
-            options.addOption("q", "query", false, "Query the ledgers that contain given bookies");
-            options.addOption("d", "dryrun", false, "Print the recovery plan w/o actually recovering");
-            options.addOption("cy", "concurrency", true, "Number of ledgers could be recovered in parallel");
-            options.addOption("sk", "skipOpenLedgers", false, "Skip recovering open ledgers");
-            options.addOption("p", "partition", true, "partition");
-            options.addOption("n", "num-partitions", true, "num partitions");
-            options.addOption("fo", "fence-only", true, "fence the ledgers only w/o re-replicating entries");
-            options.addOption("fr", "fence-rate", true, "rate on fencing ledgers");
-        }
-
-        @Override
-        protected void parseCommandLine(CommandLine cmdline) throws ParseException {
-            super.parseCommandLine(cmdline);
-            query = cmdline.hasOption("q");
-            force = cmdline.hasOption("f");
-            dryrun = cmdline.hasOption("d");
-            skipOpenLedgers = cmdline.hasOption("sk");
-            fenceOnly = cmdline.hasOption("fo");
-            if (cmdline.hasOption("l")) {
-                String[] lidStrs = cmdline.getOptionValue("l").split(",");
-                try {
-                    for (String lidStr : lidStrs) {
-                        ledgers.add(Long.parseLong(lidStr));
-                    }
-                } catch (NumberFormatException nfe) {
-                    throw new ParseException("Invalid ledger id provided : " + cmdline.getOptionValue("l"));
-                }
-            }
-            if (cmdline.hasOption("lf")) {
-                String file = cmdline.getOptionValue("lf");
-                try {
-                    BufferedReader br = new BufferedReader(
-                            new InputStreamReader(new FileInputStream(file), UTF_8.name()));
-                    try {
-                        String line = br.readLine();
-
-                        while (line != null) {
-                            ledgers.add(Long.parseLong(line));
-                            line = br.readLine();
-                        }
-                    } finally {
-                        br.close();
-                    }
-                } catch (IOException e) {
-                    throw new ParseException("Invalid ledgers file provided : " + file);
-                }
-            }
-            if (cmdline.hasOption("cy")) {
-                try {
-                    concurrency = Integer.parseInt(cmdline.getOptionValue("cy"));
-                } catch (NumberFormatException nfe) {
-                    throw new ParseException("Invalid concurrency provided : " + cmdline.getOptionValue("cy"));
-                }
-            }
-            if (cmdline.hasOption("p")) {
-                partition = Integer.parseInt(cmdline.getOptionValue("p"));
-            }
-            if (cmdline.hasOption("n")) {
-                numPartitions = Integer.parseInt(cmdline.getOptionValue("n"));
-            }
-            if (cmdline.hasOption("fr")) {
-                fenceRate = Integer.parseInt(cmdline.getOptionValue("fr"));
-            }
-            // Get bookies list to recover
-            String[] args = cmdline.getArgs();
-            final String[] bookieStrs = args[0].split(",");
-            for (String bookieStr : bookieStrs) {
-                final String bookieStrParts[] = bookieStr.split(":");
-                if (bookieStrParts.length != 2) {
-                    throw new ParseException("BookieSrcs has invalid bookie address format (host:port expected) : "
-                            + bookieStr);
-                }
-                try {
-                    bookiesSrc.add(new BookieSocketAddress(bookieStrParts[0],
-                            Integer.parseInt(bookieStrParts[1])));
-                } catch (NumberFormatException nfe) {
-                    throw new ParseException("Invalid ledger id provided : " + cmdline.getOptionValue("l"));
-                }
-            }
-        }
-
-        @Override
-        protected int runBKCmd(ZooKeeperClient zkc, BookKeeperClient bkc) throws Exception {
-            BookKeeperAdmin bkAdmin = new BookKeeperAdmin(bkc.get());
-            try {
-                if (query) {
-                    return bkQuery(bkAdmin, bookiesSrc);
-                }
-                if (fenceOnly) {
-                    return bkFence(bkc, ledgers, fenceRate);
-                }
-                if (!force) {
-                    System.out.println("Bookies : " + bookiesSrc);
-                    if (!IOUtils.confirmPrompt("Do you want to recover them: (Y/N)")) {
-                        return -1;
-                    }
-                }
-                if (!ledgers.isEmpty()) {
-                    System.out.println("Ledgers : " + ledgers);
-                    long numProcessed = 0;
-                    Iterator<Long> ledgersIter = ledgers.iterator();
-                    LinkedBlockingQueue<Long> ledgersToProcess = new LinkedBlockingQueue<Long>();
-                    while (ledgersIter.hasNext()) {
-                        long lid = ledgersIter.next();
-                        if (numPartitions <=0 || (numPartitions > 0 && lid % numPartitions == partition)) {
-                            ledgersToProcess.add(lid);
-                            ++numProcessed;
-                        }
-                        if (ledgersToProcess.size() == 10000) {
-                            System.out.println("Processing " + numProcessed + " ledgers");
-                            bkRecovery(ledgersToProcess, bookiesSrc, dryrun, skipOpenLedgers);
-                            ledgersToProcess.clear();
-                            System.out.println("Processed " + numProcessed + " ledgers");
-                        }
-                    }
-                    if (!ledgersToProcess.isEmpty()) {
-                        System.out.println("Processing " + numProcessed + " ledgers");
-                        bkRecovery(ledgersToProcess, bookiesSrc, dryrun, skipOpenLedgers);
-                        System.out.println("Processed " + numProcessed + " ledgers");
-                    }
-                    System.out.println("Done.");
-                    CountDownLatch latch = new CountDownLatch(1);
-                    latch.await();
-                    return 0;
-                }
-                return bkRecovery(bkAdmin, bookiesSrc, dryrun, skipOpenLedgers);
-            } finally {
-                bkAdmin.close();
-            }
-        }
-
-        private int bkFence(final BookKeeperClient bkc, List<Long> ledgers, int fenceRate) throws Exception {
-            if (ledgers.isEmpty()) {
-                System.out.println("Nothing to fence. Done.");
-                return 0;
-            }
-            ExecutorService executorService = Executors.newCachedThreadPool();
-            final RateLimiter rateLimiter = RateLimiter.create(fenceRate);
-            final byte[] passwd = getConf().getBKDigestPW().getBytes(UTF_8);
-            final CountDownLatch latch = new CountDownLatch(ledgers.size());
-            final AtomicInteger numPendings = new AtomicInteger(ledgers.size());
-            final LinkedBlockingQueue<Long> ledgersQueue = new LinkedBlockingQueue<Long>();
-            ledgersQueue.addAll(ledgers);
-
-            for (int i = 0; i < concurrency; i++) {
-                executorService.submit(new Runnable() {
-                    @Override
-                    public void run() {
-                        while (!ledgersQueue.isEmpty()) {
-                            rateLimiter.acquire();
-                            Long lid = ledgersQueue.poll();
-                            if (null == lid) {
-                                break;
-                            }
-                            System.out.println("Fencing ledger " + lid);
-                            int numRetries = 3;
-                            while (numRetries > 0) {
-                                try {
-                                    LedgerHandle lh = bkc.get().openLedger(lid, BookKeeper.DigestType.CRC32, passwd);
-                                    lh.close();
-                                    System.out.println("Fenced ledger " + lid + ", " + numPendings.decrementAndGet() + " left.");
-                                    latch.countDown();
-                                } catch (BKException.BKNoSuchLedgerExistsException bke) {
-                                    System.out.println("Skipped fence non-exist ledger " + lid + ", " + numPendings.decrementAndGet() + " left.");
-                                    latch.countDown();
-                                } catch (BKException.BKLedgerRecoveryException lre) {
-                                    --numRetries;
-                                    continue;
-                                } catch (Exception e) {
-                                    e.printStackTrace();
-                                    break;
-                                }
-                                numRetries = 0;
-                            }
-                        }
-                        System.out.println("Thread exits");
-                    }
-                });
-            }
-            latch.await();
-            SchedulerUtils.shutdownScheduler(executorService, 2, TimeUnit.MINUTES);
-            return 0;
-        }
-
-        private int bkQuery(BookKeeperAdmin bkAdmin, Set<BookieSocketAddress> bookieAddrs)
-                throws InterruptedException, BKException {
-            SortedMap<Long, LedgerMetadata> ledgersContainBookies =
-                    bkAdmin.getLedgersContainBookies(bookieAddrs);
-            System.err.println("NOTE: Bookies in inspection list are marked with '*'.");
-            for (Map.Entry<Long, LedgerMetadata> ledger : ledgersContainBookies.entrySet()) {
-                System.out.println("ledger " + ledger.getKey() + " : " + ledger.getValue().getState());
-                Map<Long, Integer> numBookiesToReplacePerEnsemble =
-                        inspectLedger(ledger.getValue(), bookieAddrs);
-                System.out.print("summary: [");
-                for (Map.Entry<Long, Integer> entry : numBookiesToReplacePerEnsemble.entrySet()) {
-                    System.out.print(entry.getKey() + "=" + entry.getValue() + ", ");
-                }
-                System.out.println("]");
-                System.out.println();
-            }
-            System.out.println("Done");
-            return 0;
-        }
-
-        private Map<Long, Integer> inspectLedger(LedgerMetadata metadata, Set<BookieSocketAddress> bookiesToInspect) {
-            Map<Long, Integer> numBookiesToReplacePerEnsemble = new TreeMap<Long, Integer>();
-            for (Map.Entry<Long, ArrayList<BookieSocketAddress>> ensemble : metadata.getEnsembles().entrySet()) {
-                ArrayList<BookieSocketAddress> bookieList = ensemble.getValue();
-                System.out.print(ensemble.getKey() + ":\t");
-                int numBookiesToReplace = 0;
-                for (BookieSocketAddress bookie: bookieList) {
-                    System.out.print(bookie.toString());
-                    if (bookiesToInspect.contains(bookie)) {
-                        System.out.print("*");
-                        ++numBookiesToReplace;
-                    } else {
-                        System.out.print(" ");
-                    }
-                    System.out.print(" ");
-                }
-                System.out.println();
-                numBookiesToReplacePerEnsemble.put(ensemble.getKey(), numBookiesToReplace);
-            }
-            return numBookiesToReplacePerEnsemble;
-        }
-
-        private int bkRecovery(final LinkedBlockingQueue<Long> ledgers, final Set<BookieSocketAddress> bookieAddrs,
-                               final boolean dryrun, final boolean skipOpenLedgers)
-                throws Exception {
-            return runBKCommand(new BKCommandRunner() {
-                @Override
-                public int run(ZooKeeperClient zkc, BookKeeperClient bkc) throws Exception {
-                    BookKeeperAdmin bkAdmin = new BookKeeperAdmin(bkc.get());
-                    try {
-                        bkRecovery(bkAdmin, ledgers, bookieAddrs, dryrun, skipOpenLedgers);
-                        return 0;
-                    } finally {
-                        bkAdmin.close();
-                    }
-                }
-            });
-        }
-
-        private int bkRecovery(final BookKeeperAdmin bkAdmin, final LinkedBlockingQueue<Long> ledgers,
-                               final Set<BookieSocketAddress> bookieAddrs,
-                               final boolean dryrun, final boolean skipOpenLedgers)
-                throws InterruptedException, BKException {
-            final AtomicInteger numPendings = new AtomicInteger(ledgers.size());
-            final ExecutorService executorService = Executors.newCachedThreadPool();
-            final CountDownLatch doneLatch = new CountDownLatch(concurrency);
-            Runnable r = new Runnable() {
-                @Override
-                public void run() {
-                    while (!ledgers.isEmpty()) {
-                        long lid = -1L;
-      

<TRUNCATED>


Mime
View raw message