/**
 * 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.Collection;
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.twitter.distributedlog.BKDistributedLogNamespace;
import com.twitter.distributedlog.Entry;
import com.twitter.distributedlog.logsegment.LogSegmentMetadataStore;
import com.twitter.distributedlog.namespace.DistributedLogNamespace;
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.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 static com.google.common.base.Charsets.UTF_8;

@SuppressWarnings("deprecation")
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 com.twitter.distributedlog.DistributedLogManagerFactory factory = 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 {
                synchronized (this) {
                    if (null != factory) {
                        factory.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 synchronized com.twitter.distributedlog.DistributedLogManagerFactory getFactory() throws IOException {
            if (null == this.factory) {
                this.factory = new com.twitter.distributedlog.DistributedLogManagerFactory(getConf(), getUri());
                logger.info("Construct DLM : uri = {}", getUri());
            }
            return this.factory;
        }

        protected LogSegmentMetadataStore getLogSegmentMetadataStore() throws IOException {
            DistributedLogNamespace namespace = getFactory().getNamespace();
            assert(namespace instanceof BKDistributedLogNamespace);
            return ((BKDistributedLogNamespace) namespace).getWriterSegmentMetadataStore();
        }

        protected ZooKeeperClient getZooKeeperClient() throws IOException {
            DistributedLogNamespace namespace = getFactory().getNamespace();
            assert(namespace instanceof BKDistributedLogNamespace);
            return ((BKDistributedLogNamespace) namespace).getSharedWriterZKCForDL();
        }

        protected BookKeeperClient getBookKeeperClient() throws IOException {
            DistributedLogNamespace namespace = getFactory().getNamespace();
            assert(namespace instanceof BKDistributedLogNamespace);
            return ((BKDistributedLogNamespace) namespace).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;
        }
    }

    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);
            try {
                List<String> pools = getZooKeeperClient().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(),
                                                        getZooKeeperClient(), getBookKeeperClient(),
                                                        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 {
            if (printMetadata) {
                printStreamsWithMetadata(getFactory());
            } else {
                printStreams(getFactory());
            }
            return 0;
        }

        protected void printStreamsWithMetadata(com.twitter.distributedlog.DistributedLogManagerFactory factory)
                throws Exception {
            Map<String, byte[]> streams = factory.enumerateLogsWithMetadataInNamespace();
            System.out.println("Streams under " + getUri() + " : ");
            System.out.println("--------------------------------");
            for (Map.Entry<String, byte[]> entry : streams.entrySet()) {
                println(entry.getKey());
                if (null == entry.getValue() || entry.getValue().length == 0) {
                    continue;
                }
                if (printHex) {
                    System.out.println(Hex.encodeHexString(entry.getValue()));
                } else {
                    System.out.println(new String(entry.getValue(), UTF_8));
                }
                System.out.println("");
            }
            System.out.println("--------------------------------");
        }

        protected void printStreams(com.twitter.distributedlog.DistributedLogManagerFactory factory) throws Exception {
            Collection<String> streams = factory.enumerateAllLogsInNamespace();
            System.out.println("Streams under " + getUri() + " : ");
            System.out.println("--------------------------------");
            for (String stream : streams) {
                System.out.println(stream);
            }
            System.out.println("--------------------------------");
        }
    }

    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 {
            Collection<String> streamCollection = getFactory().enumerateAllLogsInNamespace();
            final List<String> streams = new ArrayList<String>();
            if (null != streamPrefix) {
                for (String s : streamCollection) {
                    if (s.startsWith(streamPrefix)) {
                        streams.add(s);
                    }
                }
            } else {
                streams.addAll(streamCollection);
            }
            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 =
                        getFactory().createDistributedLogManagerWithSharedClients(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.getLedgerId(), 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(getFactory());
        }

        private int truncateStreams(final com.twitter.distributedlog.DistributedLogManagerFactory factory) throws Exception {
            Collection<String> streamCollection = factory.enumerateAllLogsInNamespace();
            final List<String> streams = new ArrayList<String>();
            if (null != streamPrefix) {
                for (String s : streamCollection) {
                    if (s.startsWith(streamPrefix)) {
                        streams.add(s);
                    }
                }
            } else {
                streams.addAll(streamCollection);
            }
            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;
            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 {
