distributedlog-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From si...@apache.org
Subject [26/51] [partial] incubator-distributedlog git commit: DL-4: Repackage the source under apache namespace
Date Thu, 05 Jan 2017 00:51:31 GMT
http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/b44820b5/distributedlog-core/src/main/java/com/twitter/distributedlog/auditor/DLAuditor.java
----------------------------------------------------------------------
diff --git a/distributedlog-core/src/main/java/com/twitter/distributedlog/auditor/DLAuditor.java b/distributedlog-core/src/main/java/com/twitter/distributedlog/auditor/DLAuditor.java
deleted file mode 100644
index a081606..0000000
--- a/distributedlog-core/src/main/java/com/twitter/distributedlog/auditor/DLAuditor.java
+++ /dev/null
@@ -1,630 +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.auditor;
-
-import com.google.common.base.Objects;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
-import com.google.common.util.concurrent.SettableFuture;
-import com.twitter.distributedlog.BookKeeperClient;
-import com.twitter.distributedlog.BookKeeperClientBuilder;
-import com.twitter.distributedlog.DistributedLogConfiguration;
-import com.twitter.distributedlog.DistributedLogManager;
-import com.twitter.distributedlog.LogSegmentMetadata;
-import com.twitter.distributedlog.impl.BKNamespaceDriver;
-import com.twitter.distributedlog.namespace.DistributedLogNamespace;
-import com.twitter.distributedlog.ZooKeeperClient;
-import com.twitter.distributedlog.ZooKeeperClientBuilder;
-import com.twitter.distributedlog.exceptions.DLInterruptedException;
-import com.twitter.distributedlog.exceptions.ZKException;
-import com.twitter.distributedlog.impl.metadata.BKDLConfig;
-import com.twitter.distributedlog.namespace.DistributedLogNamespaceBuilder;
-import com.twitter.distributedlog.namespace.NamespaceDriver;
-import com.twitter.distributedlog.util.DLUtils;
-import org.apache.bookkeeper.client.BKException;
-import org.apache.bookkeeper.client.BookKeeper;
-import org.apache.bookkeeper.client.BookKeeperAccessor;
-import org.apache.bookkeeper.client.LedgerHandle;
-import org.apache.bookkeeper.meta.LedgerManager;
-import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks;
-import org.apache.bookkeeper.zookeeper.BoundExponentialBackoffRetryPolicy;
-import org.apache.bookkeeper.zookeeper.RetryPolicy;
-import org.apache.commons.lang3.tuple.Pair;
-import org.apache.zookeeper.AsyncCallback;
-import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.data.Stat;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.net.URI;
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.TreeSet;
-import java.util.concurrent.ConcurrentSkipListMap;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
-
-import static com.google.common.base.Charsets.UTF_8;
-
-/**
- * DL Auditor will audit DL namespace, e.g. find leaked ledger, report disk usage by streams.
- */
-public class DLAuditor {
-
-    private static final Logger logger = LoggerFactory.getLogger(DLAuditor.class);
-
-    private final DistributedLogConfiguration conf;
-
-    public DLAuditor(DistributedLogConfiguration conf) {
-        this.conf = conf;
-    }
-
-    private ZooKeeperClient getZooKeeperClient(DistributedLogNamespace namespace) {
-        NamespaceDriver driver = namespace.getNamespaceDriver();
-        assert(driver instanceof BKNamespaceDriver);
-        return ((BKNamespaceDriver) driver).getWriterZKC();
-    }
-
-    private BookKeeperClient getBookKeeperClient(DistributedLogNamespace namespace) {
-        NamespaceDriver driver = namespace.getNamespaceDriver();
-        assert(driver instanceof BKNamespaceDriver);
-        return ((BKNamespaceDriver) driver).getReaderBKC();
-    }
-
-    private String validateAndGetZKServers(List<URI> uris) {
-        URI firstURI = uris.get(0);
-        String zkServers = BKNamespaceDriver.getZKServersFromDLUri(firstURI);
-        for (URI uri : uris) {
-            if (!zkServers.equalsIgnoreCase(BKNamespaceDriver.getZKServersFromDLUri(uri))) {
-                throw new IllegalArgumentException("Uris don't belong to same zookeeper cluster");
-            }
-        }
-        return zkServers;
-    }
-
-    private BKDLConfig resolveBKDLConfig(ZooKeeperClient zkc, List<URI> uris) throws IOException {
-        URI firstURI = uris.get(0);
-        BKDLConfig bkdlConfig = BKDLConfig.resolveDLConfig(zkc, firstURI);
-        for (URI uri : uris) {
-            BKDLConfig anotherConfig = BKDLConfig.resolveDLConfig(zkc, uri);
-            if (!(Objects.equal(bkdlConfig.getBkLedgersPath(), anotherConfig.getBkLedgersPath())
-                    && Objects.equal(bkdlConfig.getBkZkServersForWriter(), anotherConfig.getBkZkServersForWriter()))) {
-                throw new IllegalArgumentException("Uris don't use same bookkeeper cluster");
-            }
-        }
-        return bkdlConfig;
-    }
-
-    public Pair<Set<Long>, Set<Long>> collectLedgers(List<URI> uris, List<List<String>> allocationPaths)
-            throws IOException {
-        Preconditions.checkArgument(uris.size() > 0, "No uri provided to audit");
-
-        String zkServers = validateAndGetZKServers(uris);
-        RetryPolicy retryPolicy = new BoundExponentialBackoffRetryPolicy(
-                conf.getZKRetryBackoffStartMillis(),
-                conf.getZKRetryBackoffMaxMillis(),
-                Integer.MAX_VALUE);
-        ZooKeeperClient zkc = ZooKeeperClientBuilder.newBuilder()
-                .name("DLAuditor-ZK")
-                .zkServers(zkServers)
-                .sessionTimeoutMs(conf.getZKSessionTimeoutMilliseconds())
-                .retryPolicy(retryPolicy)
-                .zkAclId(conf.getZkAclId())
-                .build();
-        ExecutorService executorService = Executors.newCachedThreadPool();
-        try {
-            BKDLConfig bkdlConfig = resolveBKDLConfig(zkc, uris);
-            logger.info("Resolved bookkeeper config : {}", bkdlConfig);
-
-            BookKeeperClient bkc = BookKeeperClientBuilder.newBuilder()
-                    .name("DLAuditor-BK")
-                    .dlConfig(conf)
-                    .zkServers(bkdlConfig.getBkZkServersForWriter())
-                    .ledgersPath(bkdlConfig.getBkLedgersPath())
-                    .build();
-            try {
-                Set<Long> bkLedgers = collectLedgersFromBK(bkc, executorService);
-                Set<Long> dlLedgers = collectLedgersFromDL(uris, allocationPaths);
-                return Pair.of(bkLedgers, dlLedgers);
-            } finally {
-                bkc.close();
-            }
-        } finally {
-            zkc.close();
-            executorService.shutdown();
-        }
-    }
-
-    /**
-     * Find leak ledgers phase 1: collect ledgers set.
-     */
-    private Set<Long> collectLedgersFromBK(BookKeeperClient bkc,
-                                           final ExecutorService executorService)
-            throws IOException {
-        LedgerManager lm = BookKeeperAccessor.getLedgerManager(bkc.get());
-
-        final Set<Long> ledgers = new HashSet<Long>();
-        final SettableFuture<Void> doneFuture = SettableFuture.create();
-
-        BookkeeperInternalCallbacks.Processor<Long> collector =
-                new BookkeeperInternalCallbacks.Processor<Long>() {
-            @Override
-            public void process(Long lid,
-                                final AsyncCallback.VoidCallback cb) {
-                synchronized (ledgers) {
-                    ledgers.add(lid);
-                    if (0 == ledgers.size() % 1000) {
-                        logger.info("Collected {} ledgers", ledgers.size());
-                    }
-                }
-                executorService.submit(new Runnable() {
-                    @Override
-                    public void run() {
-                        cb.processResult(BKException.Code.OK, null, null);
-                    }
-                });
-
-            }
-        };
-        AsyncCallback.VoidCallback finalCb = new AsyncCallback.VoidCallback() {
-            @Override
-            public void processResult(int rc, String path, Object ctx) {
-                if (BKException.Code.OK == rc) {
-                    doneFuture.set(null);
-                } else {
-                    doneFuture.setException(BKException.create(rc));
-                }
-            }
-        };
-        lm.asyncProcessLedgers(collector, finalCb, null, BKException.Code.OK,
-                BKException.Code.ZKException);
-        try {
-            doneFuture.get();
-            logger.info("Collected total {} ledgers", ledgers.size());
-        } catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
-            throw new DLInterruptedException("Interrupted on collecting ledgers : ", e);
-        } catch (ExecutionException e) {
-            if (e.getCause() instanceof IOException) {
-                throw (IOException)(e.getCause());
-            } else {
-                throw new IOException("Failed to collect ledgers : ", e.getCause());
-            }
-        }
-        return ledgers;
-    }
-
-    /**
-     * Find leak ledgers phase 2: collect ledgers from uris.
-     */
-    private Set<Long> collectLedgersFromDL(List<URI> uris, List<List<String>> allocationPaths)
-            throws IOException {
-        final Set<Long> ledgers = new TreeSet<Long>();
-        List<DistributedLogNamespace> namespaces =
-                new ArrayList<DistributedLogNamespace>(uris.size());
-        try {
-            for (URI uri : uris) {
-                namespaces.add(
-                        DistributedLogNamespaceBuilder.newBuilder()
-                                .conf(conf)
-                                .uri(uri)
-                                .build());
-            }
-            final CountDownLatch doneLatch = new CountDownLatch(uris.size());
-            final AtomicInteger numFailures = new AtomicInteger(0);
-            ExecutorService executor = Executors.newFixedThreadPool(uris.size());
-            try {
-                int i = 0;
-                for (final DistributedLogNamespace namespace : namespaces) {
-                    final DistributedLogNamespace dlNamespace = namespace;
-                    final URI uri = uris.get(i);
-                    final List<String> aps = allocationPaths.get(i);
-                    i++;
-                    executor.submit(new Runnable() {
-                        @Override
-                        public void run() {
-                            try {
-                                logger.info("Collecting ledgers from {} : {}", uri, aps);
-                                collectLedgersFromAllocator(uri, namespace, aps, ledgers);
-                                synchronized (ledgers) {
-                                    logger.info("Collected {} ledgers from allocators for {} : {} ",
-                                            new Object[]{ledgers.size(), uri, ledgers});
-                                }
-                                collectLedgersFromDL(uri, namespace, ledgers);
-                            } catch (IOException e) {
-                                numFailures.incrementAndGet();
-                                logger.info("Error to collect ledgers from DL : ", e);
-                            }
-                            doneLatch.countDown();
-                        }
-                    });
-                }
-                try {
-                    doneLatch.await();
-                    if (numFailures.get() > 0) {
-                        throw new IOException(numFailures.get() + " errors to collect ledgers from DL");
-                    }
-                } catch (InterruptedException e) {
-                    Thread.currentThread().interrupt();
-                    logger.warn("Interrupted on collecting ledgers from DL : ", e);
-                    throw new DLInterruptedException("Interrupted on collecting ledgers from DL : ", e);
-                }
-            } finally {
-                executor.shutdown();
-            }
-        } finally {
-            for (DistributedLogNamespace namespace : namespaces) {
-                namespace.close();
-            }
-        }
-        return ledgers;
-    }
-
-    private void collectLedgersFromAllocator(final URI uri,
-                                             final DistributedLogNamespace namespace,
-                                             final List<String> allocationPaths,
-                                             final Set<Long> ledgers) throws IOException {
-        final LinkedBlockingQueue<String> poolQueue =
-                new LinkedBlockingQueue<String>();
-        for (String allocationPath : allocationPaths) {
-            String rootPath = uri.getPath() + "/" + allocationPath;
-            try {
-                List<String> pools = getZooKeeperClient(namespace).get().getChildren(rootPath, false);
-                for (String pool : pools) {
-                    poolQueue.add(rootPath + "/" + pool);
-                }
-            } catch (KeeperException e) {
-                throw new ZKException("Failed to get list of pools from " + rootPath, e);
-            } catch (InterruptedException e) {
-                Thread.currentThread().interrupt();
-                throw new DLInterruptedException("Interrupted on getting list of pools from " + rootPath, e);
-            }
-        }
-
-
-        logger.info("Collecting ledgers from allocators for {} : {}", uri, poolQueue);
-
-        executeAction(poolQueue, 10, new Action<String>() {
-            @Override
-            public void execute(String poolPath) throws IOException {
-                try {
-                    collectLedgersFromPool(poolPath);
-                } catch (InterruptedException e) {
-                    throw new DLInterruptedException("Interrupted on collecting ledgers from allocation pool " + poolPath, e);
-                } catch (KeeperException e) {
-                    throw new ZKException("Failed to collect ledgers from allocation pool " + poolPath, e.code());
-                }
-            }
-
-            private void collectLedgersFromPool(String poolPath)
-                    throws InterruptedException, ZooKeeperClient.ZooKeeperConnectionException, KeeperException {
-                List<String> allocators = getZooKeeperClient(namespace).get()
-                                        .getChildren(poolPath, false);
-                for (String allocator : allocators) {
-                    String allocatorPath = poolPath + "/" + allocator;
-                    byte[] data = getZooKeeperClient(namespace).get().getData(allocatorPath, false, new Stat());
-                    if (null != data && data.length > 0) {
-                        try {
-                            long ledgerId = DLUtils.bytes2LogSegmentId(data);
-                            synchronized (ledgers) {
-                                ledgers.add(ledgerId);
-                            }
-                        } catch (NumberFormatException nfe) {
-                            logger.warn("Invalid ledger found in allocator path {} : ", allocatorPath, nfe);
-                        }
-                    }
-                }
-            }
-        });
-
-        logger.info("Collected ledgers from allocators for {}.", uri);
-    }
-
-    private void collectLedgersFromDL(final URI uri,
-                                      final DistributedLogNamespace namespace,
-                                      final Set<Long> ledgers) throws IOException {
-        logger.info("Enumerating {} to collect streams.", uri);
-        Iterator<String> streams = namespace.getLogs();
-        final LinkedBlockingQueue<String> streamQueue = new LinkedBlockingQueue<String>();
-        while (streams.hasNext()) {
-            streamQueue.add(streams.next());
-        }
-
-        logger.info("Collected {} streams from uri {} : {}",
-                    new Object[] { streamQueue.size(), uri, streams });
-
-        executeAction(streamQueue, 10, new Action<String>() {
-            @Override
-            public void execute(String stream) throws IOException {
-                collectLedgersFromStream(namespace, stream, ledgers);
-            }
-        });
-    }
-
-    private List<Long> collectLedgersFromStream(DistributedLogNamespace namespace,
-                                                String stream,
-                                                Set<Long> ledgers)
-            throws IOException {
-        DistributedLogManager dlm = namespace.openLog(stream);
-        try {
-            List<LogSegmentMetadata> segments = dlm.getLogSegments();
-            List<Long> sLedgers = new ArrayList<Long>();
-            for (LogSegmentMetadata segment : segments) {
-                synchronized (ledgers) {
-                    ledgers.add(segment.getLogSegmentId());
-                }
-                sLedgers.add(segment.getLogSegmentId());
-            }
-            return sLedgers;
-        } finally {
-            dlm.close();
-        }
-    }
-
-    /**
-     * Calculating stream space usage from given <i>uri</i>.
-     *
-     * @param uri dl uri
-     * @throws IOException
-     */
-    public Map<String, Long> calculateStreamSpaceUsage(final URI uri) throws IOException {
-        logger.info("Collecting stream space usage for {}.", uri);
-        DistributedLogNamespace namespace = DistributedLogNamespaceBuilder.newBuilder()
-                .conf(conf)
-                .uri(uri)
-                .build();
-        try {
-            return calculateStreamSpaceUsage(uri, namespace);
-        } finally {
-            namespace.close();
-        }
-    }
-
-    private Map<String, Long> calculateStreamSpaceUsage(
-            final URI uri, final DistributedLogNamespace namespace)
-        throws IOException {
-        Iterator<String> streams = namespace.getLogs();
-        final LinkedBlockingQueue<String> streamQueue = new LinkedBlockingQueue<String>();
-        while (streams.hasNext()) {
-            streamQueue.add(streams.next());
-        }
-
-        final Map<String, Long> streamSpaceUsageMap =
-                new ConcurrentSkipListMap<String, Long>();
-        final AtomicInteger numStreamsCollected = new AtomicInteger(0);
-
-        executeAction(streamQueue, 10, new Action<String>() {
-            @Override
-            public void execute(String stream) throws IOException {
-                streamSpaceUsageMap.put(stream,
-                        calculateStreamSpaceUsage(namespace, stream));
-                if (numStreamsCollected.incrementAndGet() % 1000 == 0) {
-                    logger.info("Calculated {} streams from uri {}.", numStreamsCollected.get(), uri);
-                }
-            }
-        });
-
-        return streamSpaceUsageMap;
-    }
-
-    private long calculateStreamSpaceUsage(final DistributedLogNamespace namespace,
-                                           final String stream) throws IOException {
-        DistributedLogManager dlm = namespace.openLog(stream);
-        long totalBytes = 0;
-        try {
-            List<LogSegmentMetadata> segments = dlm.getLogSegments();
-            for (LogSegmentMetadata segment : segments) {
-                try {
-                    LedgerHandle lh = getBookKeeperClient(namespace).get().openLedgerNoRecovery(segment.getLogSegmentId(),
-                            BookKeeper.DigestType.CRC32, conf.getBKDigestPW().getBytes(UTF_8));
-                    totalBytes += lh.getLength();
-                    lh.close();
-                } catch (BKException e) {
-                    logger.error("Failed to open ledger {} : ", segment.getLogSegmentId(), e);
-                    throw new IOException("Failed to open ledger " + segment.getLogSegmentId(), e);
-                } catch (InterruptedException e) {
-                    logger.warn("Interrupted on opening ledger {} : ", segment.getLogSegmentId(), e);
-                    Thread.currentThread().interrupt();
-                    throw new DLInterruptedException("Interrupted on opening ledger " + segment.getLogSegmentId(), e);
-                }
-            }
-        } finally {
-            dlm.close();
-        }
-        return totalBytes;
-    }
-
-    public long calculateLedgerSpaceUsage(URI uri) throws IOException {
-        List<URI> uris = Lists.newArrayList(uri);
-        String zkServers = validateAndGetZKServers(uris);
-        RetryPolicy retryPolicy = new BoundExponentialBackoffRetryPolicy(
-                conf.getZKRetryBackoffStartMillis(),
-                conf.getZKRetryBackoffMaxMillis(),
-                Integer.MAX_VALUE);
-        ZooKeeperClient zkc = ZooKeeperClientBuilder.newBuilder()
-                .name("DLAuditor-ZK")
-                .zkServers(zkServers)
-                .sessionTimeoutMs(conf.getZKSessionTimeoutMilliseconds())
-                .retryPolicy(retryPolicy)
-                .zkAclId(conf.getZkAclId())
-                .build();
-        ExecutorService executorService = Executors.newCachedThreadPool();
-        try {
-            BKDLConfig bkdlConfig = resolveBKDLConfig(zkc, uris);
-            logger.info("Resolved bookkeeper config : {}", bkdlConfig);
-
-            BookKeeperClient bkc = BookKeeperClientBuilder.newBuilder()
-                    .name("DLAuditor-BK")
-                    .dlConfig(conf)
-                    .zkServers(bkdlConfig.getBkZkServersForWriter())
-                    .ledgersPath(bkdlConfig.getBkLedgersPath())
-                    .build();
-            try {
-                return calculateLedgerSpaceUsage(bkc, executorService);
-            } finally {
-                bkc.close();
-            }
-        } finally {
-            zkc.close();
-            executorService.shutdown();
-        }
-    }
-
-    private long calculateLedgerSpaceUsage(BookKeeperClient bkc,
-                                           final ExecutorService executorService)
-        throws IOException {
-        final AtomicLong totalBytes = new AtomicLong(0);
-        final AtomicLong totalEntries = new AtomicLong(0);
-        final AtomicLong numLedgers = new AtomicLong(0);
-
-        LedgerManager lm = BookKeeperAccessor.getLedgerManager(bkc.get());
-
-        final SettableFuture<Void> doneFuture = SettableFuture.create();
-        final BookKeeper bk = bkc.get();
-
-        BookkeeperInternalCallbacks.Processor<Long> collector =
-                new BookkeeperInternalCallbacks.Processor<Long>() {
-            @Override
-            public void process(final Long lid,
-                                final AsyncCallback.VoidCallback cb) {
-                numLedgers.incrementAndGet();
-                executorService.submit(new Runnable() {
-                    @Override
-                    public void run() {
-                        bk.asyncOpenLedgerNoRecovery(lid, BookKeeper.DigestType.CRC32, conf.getBKDigestPW().getBytes(UTF_8),
-                                new org.apache.bookkeeper.client.AsyncCallback.OpenCallback() {
-                            @Override
-                            public void openComplete(int rc, LedgerHandle lh, Object ctx) {
-                                final int cbRc;
-                                if (BKException.Code.OK == rc) {
-                                    totalBytes.addAndGet(lh.getLength());
-                                    totalEntries.addAndGet(lh.getLastAddConfirmed() + 1);
-                                    cbRc = rc;
-                                } else {
-                                    cbRc = BKException.Code.ZKException;
-                                }
-                                executorService.submit(new Runnable() {
-                                    @Override
-                                    public void run() {
-                                        cb.processResult(cbRc, null, null);
-                                    }
-                                });
-                            }
-                        }, null);
-                    }
-                });
-            }
-        };
-        AsyncCallback.VoidCallback finalCb = new AsyncCallback.VoidCallback() {
-            @Override
-            public void processResult(int rc, String path, Object ctx) {
-                if (BKException.Code.OK == rc) {
-                    doneFuture.set(null);
-                } else {
-                    doneFuture.setException(BKException.create(rc));
-                }
-            }
-        };
-        lm.asyncProcessLedgers(collector, finalCb, null, BKException.Code.OK, BKException.Code.ZKException);
-        try {
-            doneFuture.get();
-            logger.info("calculated {} ledgers\n\ttotal bytes = {}\n\ttotal entries = {}",
-                    new Object[] { numLedgers.get(), totalBytes.get(), totalEntries.get() });
-        } catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
-            throw new DLInterruptedException("Interrupted on calculating ledger space : ", e);
-        } catch (ExecutionException e) {
-            if (e.getCause() instanceof IOException) {
-                throw (IOException)(e.getCause());
-            } else {
-                throw new IOException("Failed to calculate ledger space : ", e.getCause());
-            }
-        }
-        return totalBytes.get();
-    }
-
-    public void close() {
-        // no-op
-    }
-
-    static interface Action<T> {
-        void execute(T item) throws IOException ;
-    }
-
-    static <T> void executeAction(final LinkedBlockingQueue<T> queue,
-                                  final int numThreads,
-                                  final Action<T> action) throws IOException {
-        final CountDownLatch failureLatch = new CountDownLatch(1);
-        final CountDownLatch doneLatch = new CountDownLatch(queue.size());
-        final AtomicInteger numFailures = new AtomicInteger(0);
-        final AtomicInteger completedThreads = new AtomicInteger(0);
-
-        ExecutorService executorService = Executors.newFixedThreadPool(numThreads);
-        try {
-            for (int i = 0 ; i < numThreads; i++) {
-                executorService.submit(new Runnable() {
-                    @Override
-                    public void run() {
-                        while (true) {
-                            T item = queue.poll();
-                            if (null == item) {
-                                break;
-                            }
-                            try {
-                                action.execute(item);
-                            } catch (IOException ioe) {
-                                logger.error("Failed to execute action on item '{}'", item, ioe);
-                                numFailures.incrementAndGet();
-                                failureLatch.countDown();
-                                break;
-                            }
-                            doneLatch.countDown();
-                        }
-                        if (numFailures.get() == 0 && completedThreads.incrementAndGet() == numThreads) {
-                            failureLatch.countDown();
-                        }
-                    }
-                });
-            }
-            try {
-                failureLatch.await();
-                if (numFailures.get() > 0) {
-                    throw new IOException("Encountered " + numFailures.get() + " failures on executing action.");
-                }
-                doneLatch.await();
-            } catch (InterruptedException ie) {
-                Thread.currentThread().interrupt();
-                logger.warn("Interrupted on executing action", ie);
-                throw new DLInterruptedException("Interrupted on executing action", ie);
-            }
-        } finally {
-            executorService.shutdown();
-        }
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/b44820b5/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/DynamicQuorumConfigProvider.java
----------------------------------------------------------------------
diff --git a/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/DynamicQuorumConfigProvider.java b/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/DynamicQuorumConfigProvider.java
deleted file mode 100644
index 871997f..0000000
--- a/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/DynamicQuorumConfigProvider.java
+++ /dev/null
@@ -1,37 +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.bk;
-
-import com.twitter.distributedlog.config.DynamicDistributedLogConfiguration;
-
-/**
- * Provider returns quorum configs based on dynamic configuration.
- */
-public class DynamicQuorumConfigProvider implements QuorumConfigProvider {
-
-    private final DynamicDistributedLogConfiguration conf;
-
-    public DynamicQuorumConfigProvider(DynamicDistributedLogConfiguration conf) {
-        this.conf = conf;
-    }
-
-    @Override
-    public QuorumConfig getQuorumConfig() {
-        return conf.getQuorumConfig();
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/b44820b5/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/ImmutableQuorumConfigProvider.java
----------------------------------------------------------------------
diff --git a/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/ImmutableQuorumConfigProvider.java b/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/ImmutableQuorumConfigProvider.java
deleted file mode 100644
index 6c3f06e..0000000
--- a/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/ImmutableQuorumConfigProvider.java
+++ /dev/null
@@ -1,35 +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.bk;
-
-/**
- * Provider that returns an immutable quorum config.
- */
-public class ImmutableQuorumConfigProvider implements QuorumConfigProvider {
-
-    private final QuorumConfig quorumConfig;
-
-    public ImmutableQuorumConfigProvider(QuorumConfig quorumConfig) {
-        this.quorumConfig = quorumConfig;
-    }
-
-    @Override
-    public QuorumConfig getQuorumConfig() {
-        return quorumConfig;
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/b44820b5/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/LedgerAllocator.java
----------------------------------------------------------------------
diff --git a/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/LedgerAllocator.java b/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/LedgerAllocator.java
deleted file mode 100644
index c14f374..0000000
--- a/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/LedgerAllocator.java
+++ /dev/null
@@ -1,32 +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.bk;
-
-import com.twitter.distributedlog.util.Allocator;
-import org.apache.bookkeeper.client.LedgerHandle;
-
-import java.io.IOException;
-
-public interface LedgerAllocator extends Allocator<LedgerHandle, Object> {
-
-    /**
-     * Start the ledger allocator. The implementaion should not be blocking call.
-     */
-    void start() throws IOException;
-
-}

http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/b44820b5/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/LedgerAllocatorDelegator.java
----------------------------------------------------------------------
diff --git a/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/LedgerAllocatorDelegator.java b/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/LedgerAllocatorDelegator.java
deleted file mode 100644
index b76d03a..0000000
--- a/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/LedgerAllocatorDelegator.java
+++ /dev/null
@@ -1,83 +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.bk;
-
-import com.twitter.distributedlog.util.Transaction;
-import com.twitter.distributedlog.util.Transaction.OpListener;
-import com.twitter.util.Future;
-import org.apache.bookkeeper.client.LedgerHandle;
-
-import java.io.IOException;
-
-/**
- * Delegator of the underlying allocator. If it owns the allocator, it takes
- * the responsibility of start the allocator and close the allocator.
- */
-public class LedgerAllocatorDelegator implements LedgerAllocator {
-
-    private final LedgerAllocator allocator;
-    private final boolean ownAllocator;
-
-    /**
-     * Create an allocator's delegator.
-     *
-     * @param allocator
-     *          the underlying allocator
-     * @param ownAllocator
-     *          whether to own the allocator
-     */
-    public LedgerAllocatorDelegator(LedgerAllocator allocator,
-                                    boolean ownAllocator)
-            throws IOException {
-        this.allocator = allocator;
-        this.ownAllocator = ownAllocator;
-        if (this.ownAllocator) {
-            this.allocator.start();
-        }
-    }
-
-    @Override
-    public void start() throws IOException {
-        // no-op
-    }
-
-    @Override
-    public Future<Void> delete() {
-        return Future.exception(new UnsupportedOperationException("Can't delete an allocator by delegator"));
-    }
-
-    @Override
-    public void allocate() throws IOException {
-        this.allocator.allocate();
-    }
-
-    @Override
-    public Future<LedgerHandle> tryObtain(Transaction<Object> txn,
-                                          OpListener<LedgerHandle> listener) {
-        return this.allocator.tryObtain(txn, listener);
-    }
-
-    @Override
-    public Future<Void> asyncClose() {
-        if (ownAllocator) {
-            return this.allocator.asyncClose();
-        } else {
-            return Future.value(null);
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/b44820b5/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/LedgerAllocatorPool.java
----------------------------------------------------------------------
diff --git a/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/LedgerAllocatorPool.java b/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/LedgerAllocatorPool.java
deleted file mode 100644
index dd0894e..0000000
--- a/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/LedgerAllocatorPool.java
+++ /dev/null
@@ -1,458 +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.bk;
-
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.Lists;
-import com.twitter.distributedlog.BookKeeperClient;
-import com.twitter.distributedlog.DistributedLogConfiguration;
-import com.twitter.distributedlog.ZooKeeperClient;
-import com.twitter.distributedlog.exceptions.DLInterruptedException;
-import com.twitter.distributedlog.util.FutureUtils;
-import com.twitter.distributedlog.util.Transaction;
-import com.twitter.distributedlog.util.Utils;
-import com.twitter.util.Function;
-import com.twitter.util.Future;
-import com.twitter.util.FutureEventListener;
-import com.twitter.util.Promise;
-import org.apache.bookkeeper.client.LedgerHandle;
-import org.apache.bookkeeper.meta.ZkVersion;
-import org.apache.bookkeeper.util.ZkUtils;
-import org.apache.bookkeeper.versioning.Versioned;
-import org.apache.zookeeper.AsyncCallback;
-import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.data.Stat;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import scala.runtime.AbstractFunction1;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-
-public class LedgerAllocatorPool implements LedgerAllocator {
-
-    static final Logger logger = LoggerFactory.getLogger(LedgerAllocatorPool.class);
-
-    private final DistributedLogConfiguration conf;
-    private final QuorumConfigProvider quorumConfigProvider;
-    private final BookKeeperClient bkc;
-    private final ZooKeeperClient zkc;
-    private final ScheduledExecutorService scheduledExecutorService;
-    private final String poolPath;
-    private final int corePoolSize;
-
-    private final LinkedList<SimpleLedgerAllocator> pendingList =
-            new LinkedList<SimpleLedgerAllocator>();
-    private final LinkedList<SimpleLedgerAllocator> allocatingList =
-            new LinkedList<SimpleLedgerAllocator>();
-    private final Map<String, SimpleLedgerAllocator> rescueMap =
-            new HashMap<String, SimpleLedgerAllocator>();
-    private final Map<LedgerHandle, SimpleLedgerAllocator> obtainMap =
-            new HashMap<LedgerHandle, SimpleLedgerAllocator>();
-    private final Map<SimpleLedgerAllocator, LedgerHandle> reverseObtainMap =
-            new HashMap<SimpleLedgerAllocator, LedgerHandle>();
-
-    public LedgerAllocatorPool(String poolPath, int corePoolSize,
-                               DistributedLogConfiguration conf,
-                               ZooKeeperClient zkc,
-                               BookKeeperClient bkc,
-                               ScheduledExecutorService scheduledExecutorService) throws IOException {
-        this.poolPath = poolPath;
-        this.corePoolSize = corePoolSize;
-        this.conf = conf;
-        this.quorumConfigProvider =
-                new ImmutableQuorumConfigProvider(conf.getQuorumConfig());
-        this.zkc = zkc;
-        this.bkc = bkc;
-        this.scheduledExecutorService = scheduledExecutorService;
-        initializePool();
-    }
-
-    @Override
-    public void start() throws IOException {
-        for (LedgerAllocator allocator : pendingList) {
-            // issue allocating requests during initialize
-            allocator.allocate();
-        }
-    }
-
-    @VisibleForTesting
-    synchronized int pendingListSize() {
-        return pendingList.size();
-    }
-
-    @VisibleForTesting
-    synchronized int allocatingListSize() {
-        return allocatingList.size();
-    }
-
-    @VisibleForTesting
-    public synchronized int obtainMapSize() {
-        return obtainMap.size();
-    }
-
-    @VisibleForTesting
-    synchronized int rescueSize() {
-        return rescueMap.size();
-    }
-
-    @VisibleForTesting
-    synchronized SimpleLedgerAllocator getLedgerAllocator(LedgerHandle lh) {
-        return obtainMap.get(lh);
-    }
-
-    private void initializePool() throws IOException {
-        try {
-            List<String> allocators;
-            try {
-                allocators = zkc.get().getChildren(poolPath, false);
-            } catch (KeeperException.NoNodeException e) {
-                logger.info("Allocator Pool {} doesn't exist. Creating it.", poolPath);
-                ZkUtils.createFullPathOptimistic(zkc.get(), poolPath, new byte[0], zkc.getDefaultACL(),
-                        CreateMode.PERSISTENT);
-                allocators = zkc.get().getChildren(poolPath, false);
-            }
-            if (null == allocators) {
-                allocators = new ArrayList<String>();
-            }
-            if (allocators.size() < corePoolSize) {
-                createAllocators(corePoolSize - allocators.size());
-                allocators = zkc.get().getChildren(poolPath, false);
-            }
-            initializeAllocators(allocators);
-        } catch (InterruptedException ie) {
-            throw new DLInterruptedException("Interrupted when ensuring " + poolPath + " created : ", ie);
-        } catch (KeeperException ke) {
-            throw new IOException("Encountered zookeeper exception when initializing pool " + poolPath + " : ", ke);
-        }
-    }
-
-    private void createAllocators(int numAllocators) throws InterruptedException, IOException {
-        final AtomicInteger numPendings = new AtomicInteger(numAllocators);
-        final AtomicInteger numFailures = new AtomicInteger(0);
-        final CountDownLatch latch = new CountDownLatch(1);
-        AsyncCallback.StringCallback createCallback = new AsyncCallback.StringCallback() {
-            @Override
-            public void processResult(int rc, String path, Object ctx, String name) {
-                if (KeeperException.Code.OK.intValue() != rc) {
-                    numFailures.incrementAndGet();
-                    latch.countDown();
-                    return;
-                }
-                if (numPendings.decrementAndGet() == 0 && numFailures.get() == 0) {
-                    latch.countDown();
-                }
-            }
-        };
-        for (int i = 0; i < numAllocators; i++) {
-            zkc.get().create(poolPath + "/A", new byte[0],
-                             zkc.getDefaultACL(),
-                             CreateMode.PERSISTENT_SEQUENTIAL,
-                             createCallback, null);
-        }
-        latch.await();
-        if (numFailures.get() > 0) {
-            throw new IOException("Failed to create " + numAllocators + " allocators.");
-        }
-    }
-
-    /**
-     * Initialize simple allocators with given list of allocator names <i>allocators</i>.
-     * It initializes a simple allocator with its simple allocator path.
-     */
-    private void initializeAllocators(List<String> allocators) throws IOException, InterruptedException {
-        final AtomicInteger numPendings = new AtomicInteger(allocators.size());
-        final AtomicInteger numFailures = new AtomicInteger(0);
-        final CountDownLatch latch = new CountDownLatch(numPendings.get() > 0 ? 1 : 0);
-        AsyncCallback.DataCallback dataCallback = new AsyncCallback.DataCallback() {
-            @Override
-            public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
-                if (KeeperException.Code.OK.intValue() != rc) {
-                    numFailures.incrementAndGet();
-                    latch.countDown();
-                    return;
-                }
-                Versioned<byte[]> allocatorData =
-                        new Versioned<byte[]>(data, new ZkVersion(stat.getVersion()));
-                SimpleLedgerAllocator allocator =
-                        new SimpleLedgerAllocator(path, allocatorData, quorumConfigProvider, zkc, bkc);
-                allocator.start();
-                pendingList.add(allocator);
-                if (numPendings.decrementAndGet() == 0 && numFailures.get() == 0) {
-                    latch.countDown();
-                }
-            }
-        };
-        for (String name : allocators) {
-            String path = poolPath + "/" + name;
-            zkc.get().getData(path, false, dataCallback, null);
-        }
-        latch.await();
-        if (numFailures.get() > 0) {
-            throw new IOException("Failed to initialize allocators : " + allocators);
-        }
-    }
-
-    private void scheduleAllocatorRescue(final SimpleLedgerAllocator ledgerAllocator) {
-        try {
-            scheduledExecutorService.schedule(new Runnable() {
-                @Override
-                public void run() {
-                    try {
-                        rescueAllocator(ledgerAllocator);
-                    } catch (DLInterruptedException dle) {
-                        Thread.currentThread().interrupt();
-                    }
-                }
-            }, conf.getZKRetryBackoffStartMillis(), TimeUnit.MILLISECONDS);
-        } catch (RejectedExecutionException ree) {
-            logger.warn("Failed to schedule rescuing ledger allocator {} : ", ledgerAllocator.allocatePath, ree);
-        }
-    }
-
-    /**
-     * Rescue a ledger allocator from an ERROR state
-     * @param ledgerAllocator
-     *          ledger allocator to rescue
-     */
-    private void rescueAllocator(final SimpleLedgerAllocator ledgerAllocator) throws DLInterruptedException {
-        SimpleLedgerAllocator oldAllocator;
-        synchronized (this) {
-            oldAllocator = rescueMap.put(ledgerAllocator.allocatePath, ledgerAllocator);
-        }
-        if (oldAllocator != null) {
-            logger.info("ledger allocator {} is being rescued.", ledgerAllocator.allocatePath);
-            return;
-        }
-        try {
-            zkc.get().getData(ledgerAllocator.allocatePath, false, new AsyncCallback.DataCallback() {
-                @Override
-                public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
-                    boolean retry = false;
-                    SimpleLedgerAllocator newAllocator = null;
-                    if (KeeperException.Code.OK.intValue() == rc) {
-                        Versioned<byte[]> allocatorData =
-                                new Versioned<byte[]>(data, new ZkVersion(stat.getVersion()));
-                        logger.info("Rescuing ledger allocator {}.", path);
-                        newAllocator = new SimpleLedgerAllocator(path, allocatorData, quorumConfigProvider, zkc, bkc);
-                        newAllocator.start();
-                        logger.info("Rescued ledger allocator {}.", path);
-                    } else if (KeeperException.Code.NONODE.intValue() == rc) {
-                        logger.info("Ledger allocator {} doesn't exist, skip rescuing it.", path);
-                    } else {
-                        retry = true;
-                    }
-                    synchronized (LedgerAllocatorPool.this) {
-                        rescueMap.remove(ledgerAllocator.allocatePath);
-                        if (null != newAllocator) {
-                            pendingList.addLast(newAllocator);
-                        }
-                    }
-                    if (retry) {
-                        scheduleAllocatorRescue(ledgerAllocator);
-                    }
-                }
-            }, null);
-        } catch (InterruptedException ie) {
-            logger.warn("Interrupted on rescuing ledger allocator {} : ", ledgerAllocator.allocatePath, ie);
-            synchronized (LedgerAllocatorPool.this) {
-                rescueMap.remove(ledgerAllocator.allocatePath);
-            }
-            throw new DLInterruptedException("Interrupted on rescuing ledger allocator " + ledgerAllocator.allocatePath, ie);
-        } catch (IOException ioe) {
-            logger.warn("Failed to rescue ledger allocator {}, retry rescuing it later : ", ledgerAllocator.allocatePath, ioe);
-            synchronized (LedgerAllocatorPool.this) {
-                rescueMap.remove(ledgerAllocator.allocatePath);
-            }
-            scheduleAllocatorRescue(ledgerAllocator);
-        }
-    }
-
-    @Override
-    public void allocate() throws IOException {
-        SimpleLedgerAllocator allocator;
-        synchronized (this) {
-            if (pendingList.isEmpty()) {
-                // if no ledger allocator available, we should fail it immediately, which the request will be redirected to other
-                // proxies
-                throw new IOException("No ledger allocator available under " + poolPath + ".");
-            } else {
-                allocator = pendingList.removeFirst();
-            }
-        }
-        boolean success = false;
-        try {
-            allocator.allocate();
-            synchronized (this) {
-                allocatingList.addLast(allocator);
-            }
-            success = true;
-        } finally {
-            if (!success) {
-                rescueAllocator(allocator);
-            }
-        }
-    }
-
-    @Override
-    public Future<LedgerHandle> tryObtain(final Transaction<Object> txn,
-                                          final Transaction.OpListener<LedgerHandle> listener) {
-        final SimpleLedgerAllocator allocator;
-        synchronized (this) {
-            if (allocatingList.isEmpty()) {
-                return Future.exception(new IOException("No ledger allocator available under " + poolPath + "."));
-            } else {
-                allocator = allocatingList.removeFirst();
-            }
-        }
-
-        final Promise<LedgerHandle> tryObtainPromise = new Promise<LedgerHandle>();
-        final FutureEventListener<LedgerHandle> tryObtainListener = new FutureEventListener<LedgerHandle>() {
-            @Override
-            public void onSuccess(LedgerHandle lh) {
-                synchronized (LedgerAllocatorPool.this) {
-                    obtainMap.put(lh, allocator);
-                    reverseObtainMap.put(allocator, lh);
-                    tryObtainPromise.setValue(lh);
-                }
-            }
-
-            @Override
-            public void onFailure(Throwable cause) {
-                try {
-                    rescueAllocator(allocator);
-                } catch (IOException ioe) {
-                    logger.info("Failed to rescue allocator {}", allocator.allocatePath, ioe);
-                }
-                tryObtainPromise.setException(cause);
-            }
-        };
-
-        allocator.tryObtain(txn, new Transaction.OpListener<LedgerHandle>() {
-            @Override
-            public void onCommit(LedgerHandle lh) {
-                confirmObtain(allocator);
-                listener.onCommit(lh);
-            }
-
-            @Override
-            public void onAbort(Throwable t) {
-                abortObtain(allocator);
-                listener.onAbort(t);
-            }
-        }).addEventListener(tryObtainListener);
-        return tryObtainPromise;
-    }
-
-    void confirmObtain(SimpleLedgerAllocator allocator) {
-        synchronized (this) {
-            LedgerHandle lh = reverseObtainMap.remove(allocator);
-            if (null != lh) {
-                obtainMap.remove(lh);
-            }
-        }
-        synchronized (this) {
-            pendingList.addLast(allocator);
-        }
-    }
-
-    void abortObtain(SimpleLedgerAllocator allocator) {
-        synchronized (this) {
-            LedgerHandle lh = reverseObtainMap.remove(allocator);
-            if (null != lh) {
-                obtainMap.remove(lh);
-            }
-        }
-        // if a ledger allocator is aborted, it is better to rescue it. since the ledger allocator might
-        // already encounter BadVersion exception.
-        try {
-            rescueAllocator(allocator);
-        } catch (DLInterruptedException e) {
-            logger.warn("Interrupted on rescuing ledger allocator pool {} : ", poolPath, e);
-            Thread.currentThread().interrupt();
-        }
-    }
-
-    @Override
-    public Future<Void> asyncClose() {
-        List<LedgerAllocator> allocatorsToClose;
-        synchronized (this) {
-            allocatorsToClose = Lists.newArrayListWithExpectedSize(
-                    pendingList.size() + allocatingList.size() + obtainMap.size());
-            for (LedgerAllocator allocator : pendingList) {
-                allocatorsToClose.add(allocator);
-            }
-            for (LedgerAllocator allocator : allocatingList) {
-                allocatorsToClose.add(allocator);
-            }
-            for (LedgerAllocator allocator : obtainMap.values()) {
-                allocatorsToClose.add(allocator);
-            }
-        }
-        return FutureUtils.processList(allocatorsToClose, new Function<LedgerAllocator, Future<Void>>() {
-            @Override
-            public Future<Void> apply(LedgerAllocator allocator) {
-                return allocator.asyncClose();
-            }
-        }, scheduledExecutorService).map(new AbstractFunction1<List<Void>, Void>() {
-            @Override
-            public Void apply(List<Void> values) {
-                return null;
-            }
-        });
-    }
-
-    @Override
-    public Future<Void> delete() {
-        List<LedgerAllocator> allocatorsToDelete;
-        synchronized (this) {
-            allocatorsToDelete = Lists.newArrayListWithExpectedSize(
-                    pendingList.size() + allocatingList.size() + obtainMap.size());
-            for (LedgerAllocator allocator : pendingList) {
-                allocatorsToDelete.add(allocator);
-            }
-            for (LedgerAllocator allocator : allocatingList) {
-                allocatorsToDelete.add(allocator);
-            }
-            for (LedgerAllocator allocator : obtainMap.values()) {
-                allocatorsToDelete.add(allocator);
-            }
-        }
-        return FutureUtils.processList(allocatorsToDelete, new Function<LedgerAllocator, Future<Void>>() {
-            @Override
-            public Future<Void> apply(LedgerAllocator allocator) {
-                return allocator.delete();
-            }
-        }, scheduledExecutorService).flatMap(new AbstractFunction1<List<Void>, Future<Void>>() {
-            @Override
-            public Future<Void> apply(List<Void> values) {
-                return Utils.zkDelete(zkc, poolPath, new ZkVersion(-1));
-            }
-        });
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/b44820b5/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/LedgerAllocatorUtils.java
----------------------------------------------------------------------
diff --git a/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/LedgerAllocatorUtils.java b/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/LedgerAllocatorUtils.java
deleted file mode 100644
index 0db6d74..0000000
--- a/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/LedgerAllocatorUtils.java
+++ /dev/null
@@ -1,54 +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.bk;
-
-import com.twitter.distributedlog.BookKeeperClient;
-import com.twitter.distributedlog.DistributedLogConfiguration;
-import com.twitter.distributedlog.ZooKeeperClient;
-
-import java.io.IOException;
-import java.util.concurrent.ScheduledExecutorService;
-
-public class LedgerAllocatorUtils {
-
-    /**
-     * Create ledger allocator pool.
-     *
-     * @param poolPath
-     *          ledger allocator pool path.
-     * @param corePoolSize
-     *          ledger allocator pool core size.
-     * @param conf
-     *          distributedlog configuration.
-     * @param zkc
-     *          zookeeper client
-     * @param bkc
-     *          bookkeeper client
-     * @return ledger allocator
-     * @throws IOException
-     */
-    public static LedgerAllocator createLedgerAllocatorPool(
-            String poolPath,
-            int corePoolSize,
-            DistributedLogConfiguration conf,
-            ZooKeeperClient zkc,
-            BookKeeperClient bkc,
-            ScheduledExecutorService scheduledExecutorService) throws IOException {
-        return new LedgerAllocatorPool(poolPath, corePoolSize, conf, zkc, bkc, scheduledExecutorService);
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/b44820b5/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/QuorumConfig.java
----------------------------------------------------------------------
diff --git a/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/QuorumConfig.java b/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/QuorumConfig.java
deleted file mode 100644
index a9cc16c..0000000
--- a/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/QuorumConfig.java
+++ /dev/null
@@ -1,92 +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.bk;
-
-import com.google.common.base.Objects;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Configuration for quorums
- */
-public class QuorumConfig {
-
-    private static final Logger logger = LoggerFactory.getLogger(QuorumConfig.class);
-
-    private final int ensembleSize;
-    private final int writeQuorumSize;
-    private final int ackQuorumSize;
-
-    public QuorumConfig(int ensembleSize,
-                        int writeQuorumSize,
-                        int ackQuorumSize) {
-        this.ensembleSize = ensembleSize;
-        if (this.ensembleSize < writeQuorumSize) {
-            this.writeQuorumSize = this.ensembleSize;
-            logger.warn("Setting write quorum size {} greater than ensemble size {}",
-                    writeQuorumSize, this.ensembleSize);
-        } else {
-            this.writeQuorumSize = writeQuorumSize;
-        }
-        if (this.writeQuorumSize < ackQuorumSize) {
-            this.ackQuorumSize = this.writeQuorumSize;
-            logger.warn("Setting write ack quorum size {} greater than write quorum size {}",
-                    ackQuorumSize, this.writeQuorumSize);
-        } else {
-            this.ackQuorumSize = ackQuorumSize;
-        }
-    }
-
-    public int getEnsembleSize() {
-        return ensembleSize;
-    }
-
-    public int getWriteQuorumSize() {
-        return writeQuorumSize;
-    }
-
-    public int getAckQuorumSize() {
-        return ackQuorumSize;
-    }
-
-    @Override
-    public int hashCode() {
-        return Objects.hashCode(ensembleSize, writeQuorumSize, ackQuorumSize);
-    }
-
-    @Override
-    public boolean equals(Object obj) {
-        if (!(obj instanceof QuorumConfig)) {
-            return false;
-        }
-        QuorumConfig other = (QuorumConfig) obj;
-        return ensembleSize == other.ensembleSize
-                && writeQuorumSize == other.writeQuorumSize
-                && ackQuorumSize == other.ackQuorumSize;
-    }
-
-    @Override
-    public String toString() {
-        StringBuilder sb = new StringBuilder();
-        sb.append("QuorumConfig[ensemble=")
-          .append(ensembleSize).append(", write quorum=")
-          .append(writeQuorumSize).append(", ack quorum=")
-          .append(ackQuorumSize).append("]");
-        return sb.toString();
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/b44820b5/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/QuorumConfigProvider.java
----------------------------------------------------------------------
diff --git a/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/QuorumConfigProvider.java b/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/QuorumConfigProvider.java
deleted file mode 100644
index 2f65427..0000000
--- a/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/QuorumConfigProvider.java
+++ /dev/null
@@ -1,32 +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.bk;
-
-/**
- * Provider to provide quorum config
- */
-public interface QuorumConfigProvider {
-
-    /**
-     * Get the quorum config for a given log stream.
-     *
-     * @return quorum config
-     */
-    QuorumConfig getQuorumConfig();
-
-}

http://git-wip-us.apache.org/repos/asf/incubator-distributedlog/blob/b44820b5/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/SimpleLedgerAllocator.java
----------------------------------------------------------------------
diff --git a/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/SimpleLedgerAllocator.java b/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/SimpleLedgerAllocator.java
deleted file mode 100644
index ab5976e..0000000
--- a/distributedlog-core/src/main/java/com/twitter/distributedlog/bk/SimpleLedgerAllocator.java
+++ /dev/null
@@ -1,536 +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.bk;
-
-import com.google.common.collect.Lists;
-import com.twitter.distributedlog.BookKeeperClient;
-import com.twitter.distributedlog.DistributedLogConstants;
-import com.twitter.distributedlog.util.DLUtils;
-import com.twitter.distributedlog.util.Transaction;
-import com.twitter.distributedlog.util.Transaction.OpListener;
-import com.twitter.distributedlog.ZooKeeperClient;
-import com.twitter.distributedlog.util.FutureUtils;
-import com.twitter.distributedlog.util.Utils;
-import com.twitter.distributedlog.zk.ZKTransaction;
-import com.twitter.distributedlog.zk.ZKVersionedSetOp;
-import com.twitter.util.Future;
-import com.twitter.util.FutureEventListener;
-import com.twitter.util.Promise;
-import org.apache.bookkeeper.client.LedgerHandle;
-import org.apache.bookkeeper.meta.ZkVersion;
-import org.apache.bookkeeper.versioning.Version;
-import org.apache.bookkeeper.versioning.Versioned;
-import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.data.Stat;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import scala.runtime.AbstractFunction0;
-import scala.runtime.AbstractFunction1;
-import scala.runtime.BoxedUnit;
-
-import java.io.IOException;
-import java.util.LinkedList;
-import java.util.List;
-
-/**
- * Allocator to allocate ledgers.
- */
-public class SimpleLedgerAllocator implements LedgerAllocator, FutureEventListener<LedgerHandle>, OpListener<Version> {
-
-    static final Logger LOG = LoggerFactory.getLogger(SimpleLedgerAllocator.class);
-
-    static enum Phase {
-        ALLOCATING, ALLOCATED, HANDING_OVER, HANDED_OVER, ERROR
-    }
-
-    static class AllocationException extends IOException {
-
-        private static final long serialVersionUID = -1111397872059426882L;
-
-        private final Phase phase;
-
-        public AllocationException(Phase phase, String msg) {
-            super(msg);
-            this.phase = phase;
-        }
-
-        public Phase getPhase() {
-            return this.phase;
-        }
-
-    }
-
-    static class ConcurrentObtainException extends AllocationException {
-
-        private static final long serialVersionUID = -8532471098537176913L;
-
-        public ConcurrentObtainException(Phase phase, String msg) {
-            super(phase, msg);
-        }
-    }
-
-    // zookeeper client
-    final ZooKeeperClient zkc;
-    // bookkeeper client
-    final BookKeeperClient bkc;
-    // znode path
-    final String allocatePath;
-    // allocation phase
-    Phase phase = Phase.HANDED_OVER;
-    // version
-    ZkVersion version = new ZkVersion(-1);
-    // outstanding allocation
-    Promise<LedgerHandle> allocatePromise;
-    // outstanding tryObtain transaction
-    Transaction<Object> tryObtainTxn = null;
-    OpListener<LedgerHandle> tryObtainListener = null;
-    // ledger id left from previous allocation
-    Long ledgerIdLeftFromPrevAllocation = null;
-    // Allocated Ledger
-    LedgerHandle allocatedLh = null;
-
-    Future<Void> closeFuture = null;
-    final LinkedList<Future<Void>> ledgerDeletions =
-            new LinkedList<Future<Void>>();
-
-    // Ledger configuration
-    private final QuorumConfigProvider quorumConfigProvider;
-
-    static Future<Versioned<byte[]>> getAndCreateAllocationData(final String allocatePath,
-                                                                final ZooKeeperClient zkc) {
-        return Utils.zkGetData(zkc, allocatePath, false)
-                .flatMap(new AbstractFunction1<Versioned<byte[]>, Future<Versioned<byte[]>>>() {
-            @Override
-            public Future<Versioned<byte[]>> apply(Versioned<byte[]> result) {
-                if (null != result && null != result.getVersion() && null != result.getValue()) {
-                    return Future.value(result);
-                }
-                return createAllocationData(allocatePath, zkc);
-            }
-        });
-    }
-
-    private static Future<Versioned<byte[]>> createAllocationData(final String allocatePath,
-                                                                  final ZooKeeperClient zkc) {
-        try {
-            final Promise<Versioned<byte[]>> promise = new Promise<Versioned<byte[]>>();
-            zkc.get().create(allocatePath, DistributedLogConstants.EMPTY_BYTES,
-                    zkc.getDefaultACL(), CreateMode.PERSISTENT,
-                    new org.apache.zookeeper.AsyncCallback.Create2Callback() {
-                        @Override
-                        public void processResult(int rc, String path, Object ctx, String name, Stat stat) {
-                            if (KeeperException.Code.OK.intValue() == rc) {
-                                promise.setValue(new Versioned<byte[]>(DistributedLogConstants.EMPTY_BYTES,
-                                        new ZkVersion(stat.getVersion())));
-                            } else if (KeeperException.Code.NODEEXISTS.intValue() == rc) {
-                                Utils.zkGetData(zkc, allocatePath, false).proxyTo(promise);
-                            } else {
-                                promise.setException(FutureUtils.zkException(
-                                        KeeperException.create(KeeperException.Code.get(rc)), allocatePath));
-                            }
-                        }
-                    }, null);
-            return promise;
-        } catch (ZooKeeperClient.ZooKeeperConnectionException e) {
-            return Future.exception(FutureUtils.zkException(e, allocatePath));
-        } catch (InterruptedException e) {
-            return Future.exception(FutureUtils.zkException(e, allocatePath));
-        }
-    }
-
-    public static Future<SimpleLedgerAllocator> of(final String allocatePath,
-                                                   final Versioned<byte[]> allocationData,
-                                                   final QuorumConfigProvider quorumConfigProvider,
-                                                   final ZooKeeperClient zkc,
-                                                   final BookKeeperClient bkc) {
-        if (null != allocationData && null != allocationData.getValue()
-                && null != allocationData.getVersion()) {
-            return Future.value(new SimpleLedgerAllocator(allocatePath, allocationData,
-                    quorumConfigProvider, zkc, bkc));
-        }
-        return getAndCreateAllocationData(allocatePath, zkc)
-                .map(new AbstractFunction1<Versioned<byte[]>, SimpleLedgerAllocator>() {
-            @Override
-            public SimpleLedgerAllocator apply(Versioned<byte[]> allocationData) {
-                return new SimpleLedgerAllocator(allocatePath, allocationData,
-                        quorumConfigProvider, zkc, bkc);
-            }
-        });
-    }
-
-    /**
-     * Construct a ledger allocator.
-     *
-     * @param allocatePath
-     *          znode path to store the allocated ledger.
-     * @param allocationData
-     *          allocation data.
-     * @param quorumConfigProvider
-     *          Quorum configuration provider.
-     * @param zkc
-     *          zookeeper client.
-     * @param bkc
-     *          bookkeeper client.
-     */
-    public SimpleLedgerAllocator(String allocatePath,
-                                 Versioned<byte[]> allocationData,
-                                 QuorumConfigProvider quorumConfigProvider,
-                                 ZooKeeperClient zkc,
-                                 BookKeeperClient bkc) {
-        this.zkc = zkc;
-        this.bkc = bkc;
-        this.allocatePath = allocatePath;
-        this.quorumConfigProvider = quorumConfigProvider;
-        initialize(allocationData);
-    }
-
-    /**
-     * Initialize the allocator.
-     *
-     * @param allocationData
-     *          Allocation Data.
-     */
-    private void initialize(Versioned<byte[]> allocationData) {
-        setVersion((ZkVersion) allocationData.getVersion());
-        byte[] data = allocationData.getValue();
-        if (null != data && data.length > 0) {
-            // delete the allocated ledger since this is left by last allocation.
-            try {
-                ledgerIdLeftFromPrevAllocation = DLUtils.bytes2LogSegmentId(data);
-            } catch (NumberFormatException nfe) {
-                LOG.warn("Invalid data found in allocator path {} : ", allocatePath, nfe);
-            }
-        }
-
-    }
-
-    private synchronized void deleteLedgerLeftFromPreviousAllocationIfNecessary() {
-        if (null != ledgerIdLeftFromPrevAllocation) {
-            LOG.info("Deleting allocated-but-unused ledger left from previous allocation {}.", ledgerIdLeftFromPrevAllocation);
-            deleteLedger(ledgerIdLeftFromPrevAllocation);
-            ledgerIdLeftFromPrevAllocation = null;
-        }
-    }
-
-    @Override
-    public synchronized void allocate() throws IOException {
-        if (Phase.ERROR == phase) {
-            throw new AllocationException(Phase.ERROR, "Error on ledger allocator for " + allocatePath);
-        }
-        if (Phase.HANDED_OVER == phase) {
-            // issue an allocate request when ledger is already handed over.
-            allocateLedger();
-        }
-    }
-
-    @Override
-    public synchronized Future<LedgerHandle> tryObtain(final Transaction<Object> txn,
-                                                       final OpListener<LedgerHandle> listener) {
-        if (Phase.ERROR == phase) {
-            return Future.exception(new AllocationException(Phase.ERROR,
-                    "Error on allocating ledger under " + allocatePath));
-        }
-        if (Phase.HANDING_OVER == phase || Phase.HANDED_OVER == phase || null != tryObtainTxn) {
-            return Future.exception(new ConcurrentObtainException(phase,
-                    "Ledger handle is handling over to another thread : " + phase));
-        }
-        tryObtainTxn = txn;
-        tryObtainListener = listener;
-        if (null != allocatedLh) {
-            completeAllocation(allocatedLh);
-        }
-        return allocatePromise;
-    }
-
-    @Override
-    public void onCommit(Version r) {
-        confirmObtain((ZkVersion) r);
-    }
-
-    private void confirmObtain(ZkVersion zkVersion) {
-        boolean shouldAllocate = false;
-        OpListener<LedgerHandle> listenerToNotify = null;
-        LedgerHandle lhToNotify = null;
-        synchronized (this) {
-            if (Phase.HANDING_OVER == phase) {
-                setPhase(Phase.HANDED_OVER);
-                setVersion(zkVersion);
-                listenerToNotify = tryObtainListener;
-                lhToNotify = allocatedLh;
-                // reset the state
-                allocatedLh = null;
-                allocatePromise = null;
-                tryObtainTxn = null;
-                tryObtainListener = null;
-                // mark flag to issue an allocation request
-                shouldAllocate = true;
-            }
-        }
-        if (null != listenerToNotify && null != lhToNotify) {
-            // notify the listener
-            listenerToNotify.onCommit(lhToNotify);
-        }
-        if (shouldAllocate) {
-            // issue an allocation request
-            allocateLedger();
-        }
-    }
-
-    @Override
-    public void onAbort(Throwable t) {
-        OpListener<LedgerHandle> listenerToNotify;
-        synchronized (this) {
-            listenerToNotify = tryObtainListener;
-            if (t instanceof KeeperException &&
-                    ((KeeperException) t).code() == KeeperException.Code.BADVERSION) {
-                LOG.info("Set ledger allocator {} to ERROR state after hit bad version : version = {}",
-                        allocatePath, getVersion());
-                setPhase(Phase.ERROR);
-            } else {
-                if (Phase.HANDING_OVER == phase) {
-                    setPhase(Phase.ALLOCATED);
-                    tryObtainTxn = null;
-                    tryObtainListener = null;
-                }
-            }
-        }
-        if (null != listenerToNotify) {
-            listenerToNotify.onAbort(t);
-        }
-    }
-
-    private synchronized void setPhase(Phase phase) {
-        this.phase = phase;
-        LOG.info("Ledger allocator {} moved to phase {} : version = {}.",
-                new Object[] { allocatePath, phase, version });
-    }
-
-    private synchronized void allocateLedger() {
-        // make sure previous allocation is already handed over.
-        if (Phase.HANDED_OVER != phase) {
-            LOG.error("Trying allocate ledger for {} in phase {}, giving up.", allocatePath, phase);
-            return;
-        }
-        setPhase(Phase.ALLOCATING);
-        allocatePromise = new Promise<LedgerHandle>();
-        QuorumConfig quorumConfig = quorumConfigProvider.getQuorumConfig();
-        bkc.createLedger(
-                quorumConfig.getEnsembleSize(),
-                quorumConfig.getWriteQuorumSize(),
-                quorumConfig.getAckQuorumSize()
-        ).addEventListener(this);
-    }
-
-    private synchronized void completeAllocation(LedgerHandle lh) {
-        allocatedLh = lh;
-        if (null == tryObtainTxn) {
-            return;
-        }
-        org.apache.zookeeper.Op zkSetDataOp = org.apache.zookeeper.Op.setData(
-                allocatePath, DistributedLogConstants.EMPTY_BYTES, version.getZnodeVersion());
-        ZKVersionedSetOp commitOp = new ZKVersionedSetOp(zkSetDataOp, this);
-        tryObtainTxn.addOp(commitOp);
-        setPhase(Phase.HANDING_OVER);
-        FutureUtils.setValue(allocatePromise, lh);
-    }
-
-    private synchronized void failAllocation(Throwable cause) {
-        FutureUtils.setException(allocatePromise, cause);
-    }
-
-    @Override
-    public void onSuccess(LedgerHandle lh) {
-        // a ledger is created, update the ledger to allocation path before handling it over for usage.
-        markAsAllocated(lh);
-    }
-
-    @Override
-    public void onFailure(Throwable cause) {
-        LOG.error("Error creating ledger for allocating {} : ", allocatePath, cause);
-        setPhase(Phase.ERROR);
-        failAllocation(cause);
-    }
-
-    private synchronized ZkVersion getVersion() {
-        return version;
-    }
-
-    private synchronized void setVersion(ZkVersion newVersion) {
-        Version.Occurred occurred = newVersion.compare(version);
-        if (occurred == Version.Occurred.AFTER) {
-            LOG.info("Ledger allocator for {} moved version from {} to {}.",
-                    new Object[] { allocatePath, version, newVersion });
-            version = newVersion;
-        } else {
-            LOG.warn("Ledger allocator for {} received an old version {}, current version is {}.",
-                    new Object[] { allocatePath, newVersion , version });
-        }
-    }
-
-    private void markAsAllocated(final LedgerHandle lh) {
-        byte[] data = DLUtils.logSegmentId2Bytes(lh.getId());
-        Utils.zkSetData(zkc, allocatePath, data, getVersion())
-            .addEventListener(new FutureEventListener<ZkVersion>() {
-                @Override
-                public void onSuccess(ZkVersion version) {
-                    // we only issue deleting ledger left from previous allocation when we could allocate first ledger
-                    // as zookeeper version could prevent us doing stupid things.
-                    deleteLedgerLeftFromPreviousAllocationIfNecessary();
-                    setVersion(version);
-                    setPhase(Phase.ALLOCATED);
-                    // complete the allocation after it is marked as allocated
-                    completeAllocation(lh);
-                }
-
-                @Override
-                public void onFailure(Throwable cause) {
-                    setPhase(Phase.ERROR);
-                    deleteLedger(lh.getId());
-                    LOG.error("Fail mark ledger {} as allocated under {} : ",
-                            new Object[] { lh.getId(), allocatePath, cause });
-                    // fail the allocation since failed to mark it as allocated
-                    failAllocation(cause);
-                }
-            });
-    }
-
-    void deleteLedger(final long ledgerId) {
-        final Future<Void> deleteFuture = bkc.deleteLedger(ledgerId, true);
-        synchronized (ledgerDeletions) {
-            ledgerDeletions.add(deleteFuture);
-        }
-        deleteFuture.onFailure(new AbstractFunction1<Throwable, BoxedUnit>() {
-            @Override
-            public BoxedUnit apply(Throwable cause) {
-                LOG.error("Error deleting ledger {} for ledger allocator {}, retrying : ",
-                        new Object[] { ledgerId, allocatePath, cause });
-                if (!isClosing()) {
-                    deleteLedger(ledgerId);
-                }
-                return BoxedUnit.UNIT;
-            }
-        }).ensure(new AbstractFunction0<BoxedUnit>() {
-            @Override
-            public BoxedUnit apply() {
-                synchronized (ledgerDeletions) {
-                    ledgerDeletions.remove(deleteFuture);
-                }
-                return BoxedUnit.UNIT;
-            }
-        });
-    }
-
-    private synchronized boolean isClosing() {
-        return closeFuture != null;
-    }
-
-    private Future<Void> closeInternal(boolean cleanup) {
-        Promise<Void> closePromise;
-        synchronized (this) {
-            if (null != closeFuture) {
-                return closeFuture;
-            }
-            closePromise = new Promise<Void>();
-            closeFuture = closePromise;
-        }
-        if (!cleanup) {
-            LOG.info("Abort ledger allocator without cleaning up on {}.", allocatePath);
-            FutureUtils.setValue(closePromise, null);
-            return closePromise;
-        }
-        cleanupAndClose(closePromise);
-        return closePromise;
-    }
-
-    private void cleanupAndClose(final Promise<Void> closePromise) {
-        LOG.info("Closing ledger allocator on {}.", allocatePath);
-        final ZKTransaction txn = new ZKTransaction(zkc);
-        // try obtain ledger handle
-        tryObtain(txn, new OpListener<LedgerHandle>() {
-            @Override
-            public void onCommit(LedgerHandle r) {
-                // no-op
-                complete();
-            }
-
-            @Override
-            public void onAbort(Throwable t) {
-                // no-op
-                complete();
-            }
-
-            private void complete() {
-                FutureUtils.setValue(closePromise, null);
-                LOG.info("Closed ledger allocator on {}.", allocatePath);
-            }
-        }).addEventListener(new FutureEventListener<LedgerHandle>() {
-            @Override
-            public void onSuccess(LedgerHandle lh) {
-                // try obtain succeed
-                // if we could obtain the ledger handle, we have the responsibility to close it
-                deleteLedger(lh.getId());
-                // wait for deletion to be completed
-                List<Future<Void>> outstandingDeletions;
-                synchronized (ledgerDeletions) {
-                    outstandingDeletions = Lists.newArrayList(ledgerDeletions);
-                }
-                Future.collect(outstandingDeletions).addEventListener(new FutureEventListener<List<Void>>() {
-                    @Override
-                    public void onSuccess(List<Void> values) {
-                        txn.execute();
-                    }
-
-                    @Override
-                    public void onFailure(Throwable cause) {
-                        LOG.debug("Fail to obtain the allocated ledger handle when closing the allocator : ", cause);
-                        FutureUtils.setValue(closePromise, null);
-                    }
-                });
-            }
-
-            @Override
-            public void onFailure(Throwable cause) {
-                LOG.debug("Fail to obtain the allocated ledger handle when closing the allocator : ", cause);
-                FutureUtils.setValue(closePromise, null);
-            }
-        });
-
-    }
-
-    @Override
-    public void start() {
-        // nop
-    }
-
-    @Override
-    public Future<Void> asyncClose() {
-        return closeInternal(false);
-    }
-
-    @Override
-    public Future<Void> delete() {
-        return closeInternal(true).flatMap(new AbstractFunction1<Void, Future<Void>>() {
-            @Override
-            public Future<Void> apply(Void value) {
-                return Utils.zkDelete(zkc, allocatePath, getVersion());
-            }
-        });
-    }
-
-}



Mime
View raw message