/**
 * 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 org.apache.bookkeeper.meta;

import static org.apache.bookkeeper.metastore.MetastoreTable.ALL_FIELDS;
import static org.apache.bookkeeper.metastore.MetastoreTable.NON_FIELDS;

import java.io.IOException;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.LedgerMetadata;
import org.apache.bookkeeper.conf.AbstractConfiguration;
import org.apache.bookkeeper.metastore.MSException;
import org.apache.bookkeeper.metastore.MSWatchedEvent;
import org.apache.bookkeeper.metastore.MetaStore;
import org.apache.bookkeeper.metastore.MetastoreCallback;
import org.apache.bookkeeper.metastore.MetastoreCursor;
import org.apache.bookkeeper.metastore.MetastoreCursor.ReadEntriesCallback;
import org.apache.bookkeeper.metastore.MetastoreException;
import org.apache.bookkeeper.metastore.MetastoreFactory;
import org.apache.bookkeeper.metastore.MetastoreScannableTable;
import org.apache.bookkeeper.metastore.MetastoreTable;
import org.apache.bookkeeper.metastore.MetastoreTableItem;
import org.apache.bookkeeper.metastore.MetastoreUtils;
import org.apache.bookkeeper.metastore.MetastoreWatcher;
import org.apache.bookkeeper.metastore.Value;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.LedgerMetadataListener;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.Processor;
import org.apache.bookkeeper.replication.ReplicationException;
import org.apache.bookkeeper.util.StringUtils;
import org.apache.bookkeeper.versioning.Version;
import org.apache.bookkeeper.versioning.Versioned;
import org.apache.zookeeper.AsyncCallback;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.util.concurrent.ThreadFactoryBuilder;

/**
 * MetaStore Based Ledger Manager Factory
 */
public class MSLedgerManagerFactory extends LedgerManagerFactory {

    private final static Logger LOG = LoggerFactory.getLogger(MSLedgerManagerFactory.class);

    static int MS_CONNECT_BACKOFF_MS = 200;

    public static final int CUR_VERSION = 1;

    public static final String TABLE_NAME = "LEDGER";
    public static final String META_FIELD = ".META";

    AbstractConfiguration conf;
    ZooKeeper zk;
    MetaStore metastore;

    @Override
    public int getCurrentVersion() {
        return CUR_VERSION;
    }

    @Override
    public LedgerManagerFactory initialize(final AbstractConfiguration conf, final ZooKeeper zk,
            final int factoryVersion) throws IOException {
        if (CUR_VERSION != factoryVersion) {
            throw new IOException("Incompatible layout version found : " + factoryVersion);
        }
        this.conf = conf;
        this.zk = zk;

        // load metadata store
        String msName = conf.getMetastoreImplClass();
        try {
            metastore = MetastoreFactory.createMetaStore(msName);

            // TODO: should record version in somewhere. e.g. ZooKeeper
            int msVersion = metastore.getVersion();
            metastore.init(conf, msVersion);
        } catch (Throwable t) {
            throw new IOException("Failed to initialize metastore " + msName + " : ", t);
        }

        return this;
    }

    @Override
    public void uninitialize() throws IOException {
        metastore.close();
    }

    static Long key2LedgerId(String key) {
        return null == key ? null : Long.parseLong(key, 10);
    }

    static String ledgerId2Key(Long lid) {
        return null == lid ? null : StringUtils.getZKStringId(lid);
    }

    static String rangeToString(Long firstLedger, boolean firstInclusive, Long lastLedger, boolean lastInclusive) {
        StringBuilder sb = new StringBuilder();
        sb.append(firstInclusive ? "[ " : "( ").append(firstLedger).append(" ~ ").append(lastLedger)
                .append(lastInclusive ? " ]" : " )");
        return sb.toString();
    }

    static SortedSet<Long> entries2Ledgers(Iterator<MetastoreTableItem> entries) {
        SortedSet<Long> ledgers = new TreeSet<Long>();
        while (entries.hasNext()) {
            MetastoreTableItem item = entries.next();
            try {
                ledgers.add(key2LedgerId(item.getKey()));
            } catch (NumberFormatException nfe) {
                LOG.warn("Found invalid ledger key {}", item.getKey());
            }
        }
        return ledgers;
    }

    static class SyncResult<T> {
        T value;
        int rc;
        boolean finished = false;

        public synchronized void complete(int rc, T value) {
            this.rc = rc;
            this.value = value;
            finished = true;

            notify();
        }

        public synchronized void block() {
            try {
                while (!finished) {
                    wait();
                }
            } catch (InterruptedException ie) {
            }
        }

        public synchronized int getRetCode() {
            return rc;
        }

        public synchronized T getResult() {
            return value;
        }
    }

    @Override
    public LedgerIdGenerator newLedgerIdGenerator() {
        return new ZkLedgerIdGenerator(zk, conf.getZkLedgersRootPath(), MsLedgerManager.IDGEN_ZNODE);
    }

    static class MsLedgerManager implements LedgerManager, MetastoreWatcher {
        final ZooKeeper zk;
        final AbstractConfiguration conf;

        final MetaStore metastore;
        final MetastoreScannableTable ledgerTable;
        final int maxEntriesPerScan;

        static final String IDGEN_ZNODE = "ms-idgen";

        // ledger metadata listeners
        protected final ConcurrentMap<Long, Set<LedgerMetadataListener>> listeners =
                new ConcurrentHashMap<Long, Set<LedgerMetadataListener>>();

        // we use this to prevent long stack chains from building up in
        // callbacks
        ScheduledExecutorService scheduler;

        protected class ReadLedgerMetadataTask implements Runnable, GenericCallback<LedgerMetadata> {

            final long ledgerId;

            ReadLedgerMetadataTask(long ledgerId) {
                this.ledgerId = ledgerId;
            }

            @Override
            public void run() {
                if (null != listeners.get(ledgerId)) {
                    LOG.debug("Re-read ledger metadata for {}.", ledgerId);
                    readLedgerMetadata(ledgerId, ReadLedgerMetadataTask.this);
                } else {
                    LOG.debug("Ledger metadata listener for ledger {} is already removed.", ledgerId);
                }
            }

            @Override
            public void operationComplete(int rc, final LedgerMetadata result) {
                if (BKException.Code.OK == rc) {
                    final Set<LedgerMetadataListener> listenerSet = listeners.get(ledgerId);
                    if (null != listenerSet) {
                        LOG.debug("Ledger metadata is changed for {} : {}.", ledgerId, result);
                        scheduler.submit(new Runnable() {
                            @Override
                            public void run() {
                                synchronized(listenerSet) {
                                    for (LedgerMetadataListener listener : listenerSet) {
                                        listener.onChanged(ledgerId, result);
                                    }
                                }
                            }
                        });
                    }
                } else if (BKException.Code.NoSuchLedgerExistsException == rc) {
                    // the ledger is removed, do nothing
                    Set<LedgerMetadataListener> listenerSet = listeners.remove(ledgerId);
                    if (null != listenerSet) {
                        LOG.debug("Removed ledger metadata listener set on ledger {} as its ledger is deleted : {}",
                                ledgerId, listenerSet.size());
                    }
                } else {
                    LOG.warn("Failed on read ledger metadata of ledger {} : {}", ledgerId, rc);
                    scheduler.schedule(this, MS_CONNECT_BACKOFF_MS, TimeUnit.MILLISECONDS);
                }
            }
        }

        MsLedgerManager(final AbstractConfiguration conf, final ZooKeeper zk, final MetaStore metastore) {
            this.conf = conf;
            this.zk = zk;
            this.metastore = metastore;

            try {
                ledgerTable = metastore.createScannableTable(TABLE_NAME);
            } catch (MetastoreException mse) {
                LOG.error("Failed to instantiate table " + TABLE_NAME + " in metastore " + metastore.getName());
                throw new RuntimeException("Failed to instantiate table " + TABLE_NAME + " in metastore "
                        + metastore.getName());
            }
            // configuration settings
            maxEntriesPerScan = conf.getMetastoreMaxEntriesPerScan();

            ThreadFactoryBuilder tfb = new ThreadFactoryBuilder()
                    .setNameFormat("MSLedgerManagerScheduler-%d");
            this.scheduler = Executors.newSingleThreadScheduledExecutor(tfb
                    .build());
        }

        @Override
        public void process(MSWatchedEvent e){
            long ledgerId = key2LedgerId(e.getKey());
            switch(e.getType()) {
            case CHANGED:
                new ReadLedgerMetadataTask(key2LedgerId(e.getKey())).run();

                break;
            case REMOVED:
                Set<LedgerMetadataListener> listenerSet = listeners.get(ledgerId);
                if (listenerSet != null) {
                    synchronized (listenerSet) {
                        for(LedgerMetadataListener l : listenerSet){
                            unregisterLedgerMetadataListener(ledgerId, l);
                            l.onChanged( ledgerId, null );
                        }
                    }
                }

                break;
            default:
                LOG.warn("Unknown type: {}", e.getType());
                break;
            }
        }

        @Override
        public void registerLedgerMetadataListener(long ledgerId, LedgerMetadataListener listener) {
            if (null != listener) {
                LOG.info("Registered ledger metadata listener {} on ledger {}.", listener, ledgerId);
                Set<LedgerMetadataListener> listenerSet = listeners.get(ledgerId);
                if (listenerSet == null) {
                    Set<LedgerMetadataListener> newListenerSet = new HashSet<LedgerMetadataListener>();
                    Set<LedgerMetadataListener> oldListenerSet = listeners.putIfAbsent(ledgerId, newListenerSet);
                    if (null != oldListenerSet) {
                        listenerSet = oldListenerSet;
                    } else {
                        listenerSet = newListenerSet;
                    }
                }
                synchronized (listenerSet) {
                    listenerSet.add(listener);
                }
                new ReadLedgerMetadataTask(ledgerId).run();
            }
        }

        @Override
        public void unregisterLedgerMetadataListener(long ledgerId, LedgerMetadataListener listener) {
            Set<LedgerMetadataListener> listenerSet = listeners.get(ledgerId);
            if (listenerSet != null) {
                synchronized (listenerSet) {
                    if (listenerSet.remove(listener)) {
                        LOG.info("Unregistered ledger metadata listener {} on ledger {}.", listener, ledgerId);
                    }
                    if (listenerSet.isEmpty()) {
                        listeners.remove(ledgerId, listenerSet);
                    }
                }
            }
        }

        @Override
        public void close() {
            try {
                scheduler.shutdown();
            } catch (Exception e) {
                LOG.warn("Error when closing MsLedgerManager : ", e);
            }
            ledgerTable.close();
        }

        @Override
        public void createLedgerMetadata(final long lid, final LedgerMetadata metadata,
                                         final GenericCallback<Void> ledgerCb) {
            MetastoreCallback<Version> msCallback = new MetastoreCallback<Version>() {
                @Override
                public void complete(int rc, Version version, Object ctx) {
                    if (MSException.Code.BadVersion.getCode() == rc) {
                        ledgerCb.operationComplete(BKException.Code.MetadataVersionException, null);
                        return;
                    }
                    if (MSException.Code.OK.getCode() != rc) {
                        ledgerCb.operationComplete(BKException.Code.MetaStoreException, null);
                        return;
                    }
                    LOG.debug("Create ledger {} with version {} successfully.", lid, version);
                    // update version
                    metadata.setVersion(version);
                    ledgerCb.operationComplete(BKException.Code.OK, null);
                }
            };

            ledgerTable.put(ledgerId2Key(lid), new Value().setField(META_FIELD, metadata.serialize()),
                    Version.NEW, msCallback, null);
        }

        @Override
        public void removeLedgerMetadata(final long ledgerId, final Version version,
                                         final GenericCallback<Void> cb) {
            MetastoreCallback<Void> msCallback = new MetastoreCallback<Void>() {
                @Override
                public void complete(int rc, Void value, Object ctx) {
                    int bkRc;
                    if (MSException.Code.NoKey.getCode() == rc) {
                        LOG.warn("Ledger entry does not exist in meta table: ledgerId={}", ledgerId);
                        bkRc = BKException.Code.NoSuchLedgerExistsException;
                    } else if (MSException.Code.OK.getCode() == rc) {
                        bkRc = BKException.Code.OK;
                    } else {
                        bkRc = BKException.Code.MetaStoreException;
                    }
                    cb.operationComplete(bkRc, (Void) null);
                }
            };
            ledgerTable.remove(ledgerId2Key(ledgerId), version, msCallback, null);
        }

        @Override
        public void readLedgerMetadata(final long ledgerId, final GenericCallback<LedgerMetadata> readCb) {
            final String key = ledgerId2Key(ledgerId);
            MetastoreCallback<Versioned<Value>> msCallback = new MetastoreCallback<Versioned<Value>>() {
                @Override
                public void complete(int rc, Versioned<Value> value, Object ctx) {
                    if (MSException.Code.NoKey.getCode() == rc) {
                        LOG.error("No ledger metadata found for ledger " + ledgerId + " : ",
                                MSException.create(MSException.Code.get(rc), "No key " + key + " found."));
                        readCb.operationComplete(BKException.Code.NoSuchLedgerExistsException, null);
                        return;
                    }
                    if (MSException.Code.OK.getCode() != rc) {
                        LOG.error("Could not read metadata for ledger " + ledgerId + " : ",
                                MSException.create(MSException.Code.get(rc), "Failed to get key " + key));
                        readCb.operationComplete(BKException.Code.MetaStoreException, null);
                        return;
                    }
                    LedgerMetadata metadata;
                    try {
                        metadata = LedgerMetadata
                                .parseConfig(value.getValue().getField(META_FIELD), value.getVersion());
                    } catch (IOException e) {
                        LOG.error("Could not parse ledger metadata for ledger " + ledgerId + " : ", e);
                        readCb.operationComplete(BKException.Code.MetaStoreException, null);
                        return;
                    }
                    readCb.operationComplete(BKException.Code.OK, metadata);
                }
            };
            ledgerTable.get(key, this, msCallback, ALL_FIELDS);
        }

        @Override
        public void writeLedgerMetadata(final long ledgerId, final LedgerMetadata metadata,
                final GenericCallback<Void> cb) {
            Value data = new Value().setField(META_FIELD, metadata.serialize());

            LOG.debug("Writing ledger {} metadata, version {}", new Object[] { ledgerId, metadata.getVersion() });

            final String key = ledgerId2Key(ledgerId);
            MetastoreCallback<Version> msCallback = new MetastoreCallback<Version>() {
                @Override
                public void complete(int rc, Version version, Object ctx) {
                    int bkRc;
                    if (MSException.Code.BadVersion.getCode() == rc) {
                        LOG.info("Bad version provided to updat metadata for ledger {}", ledgerId);
                        bkRc = BKException.Code.MetadataVersionException;
                    } else if (MSException.Code.NoKey.getCode() == rc) {
                        LOG.warn("Ledger {} doesn't exist when writing its ledger metadata.", ledgerId);
                        bkRc = BKException.Code.NoSuchLedgerExistsException;
                    } else if (MSException.Code.OK.getCode() == rc) {
                        metadata.setVersion(version);
                        bkRc = BKException.Code.OK;
                    } else {
                        LOG.warn("Conditional update ledger metadata failed: ",
                                MSException.create(MSException.Code.get(rc), "Failed to put key " + key));
                        bkRc = BKException.Code.MetaStoreException;
                    }

                    cb.operationComplete(bkRc, null);
                }
            };
            ledgerTable.put(key, data, metadata.getVersion(), msCallback, null);
        }

        @Override
        public void asyncProcessLedgers(final Processor<Long> processor, final AsyncCallback.VoidCallback finalCb,
                final Object context, final int successRc, final int failureRc) {
            MetastoreCallback<MetastoreCursor> openCursorCb = new MetastoreCallback<MetastoreCursor>() {
                @Override
                public void complete(int rc, MetastoreCursor cursor, Object ctx) {
                    if (MSException.Code.OK.getCode() != rc) {
                        finalCb.processResult(failureRc, null, context);
                        return;
                    }
                    if (!cursor.hasMoreEntries()) {
                        finalCb.processResult(successRc, null, context);
                        return;
                    }
                    asyncProcessLedgers(cursor, processor, finalCb, context, successRc, failureRc);
                }
            };
            ledgerTable.openCursor(NON_FIELDS, openCursorCb, null);
        }

        void asyncProcessLedgers(final MetastoreCursor cursor, final Processor<Long> processor,
                                 final AsyncCallback.VoidCallback finalCb, final Object context,
                                 final int successRc, final int failureRc) {
            scheduler.submit(new Runnable() {
                @Override
                public void run() {
                    doAsyncProcessLedgers(cursor, processor, finalCb, context, successRc, failureRc);
                }
            });
        }

        void doAsyncProcessLedgers(final MetastoreCursor cursor, final Processor<Long> processor,
                                   final AsyncCallback.VoidCallback finalCb, final Object context,
                                   final int successRc, final int failureRc) {
            // no entries now
            if (!cursor.hasMoreEntries()) {
                finalCb.processResult(successRc, null, context);
                return;
            }
            ReadEntriesCallback msCallback = new ReadEntriesCallback() {
                @Override
                public void complete(int rc, Iterator<MetastoreTableItem> entries, Object ctx) {
                    if (MSException.Code.OK.getCode() != rc) {
                        finalCb.processResult(failureRc, null, context);
                        return;
                    }

                    SortedSet<Long> ledgers = new TreeSet<Long>();
                    while (entries.hasNext()) {
                        MetastoreTableItem item = entries.next();
                        try {
                            ledgers.add(key2LedgerId(item.getKey()));
                        } catch (NumberFormatException nfe) {
                            LOG.warn("Found invalid ledger key {}", item.getKey());
                        }
                    }

                    if (0 == ledgers.size()) {
                        // process next batch of ledgers
                        asyncProcessLedgers(cursor, processor, finalCb, context, successRc, failureRc);
                        return;
                    }

                    final long startLedger = ledgers.first();
                    final long endLedger = ledgers.last();

                    AsyncSetProcessor<Long> setProcessor = new AsyncSetProcessor<Long>(scheduler);
                    // process set
                    setProcessor.process(ledgers, processor, new AsyncCallback.VoidCallback() {
                        @Override
                        public void processResult(int rc, String path, Object ctx) {
                            if (successRc != rc) {
                                LOG.error("Failed when processing range "
                                        + rangeToString(startLedger, true, endLedger, true));
                                finalCb.processResult(failureRc, null, context);
                                return;
                            }
                            // process next batch of ledgers
                            asyncProcessLedgers(cursor, processor, finalCb, context, successRc, failureRc);
                        }
                    }, context, successRc, failureRc);
                }
            };
            cursor.asyncReadEntries(maxEntriesPerScan, msCallback, null);
        }

        class MSLedgerRangeIterator implements LedgerRangeIterator {
            final CountDownLatch openCursorLatch = new CountDownLatch(1);
            MetastoreCursor cursor = null;
            // last ledger id in previous range

            MSLedgerRangeIterator() {
                MetastoreCallback<MetastoreCursor> openCursorCb = new MetastoreCallback<MetastoreCursor>() {
                    @Override
                    public void complete(int rc, MetastoreCursor newCursor, Object ctx) {
                        if (MSException.Code.OK.getCode() != rc) {
                            LOG.error("Error opening cursor for ledger range iterator {}", rc);
                        } else {
                            cursor = newCursor;
                        }
                        openCursorLatch.countDown();
                    }
                };
                ledgerTable.openCursor(NON_FIELDS, openCursorCb, null);
            }

            @Override
            public boolean hasNext() throws IOException {
                try {
                    openCursorLatch.await();
                } catch (InterruptedException ie) {
                    LOG.error("Interrupted waiting for cursor to open", ie);
                    Thread.currentThread().interrupt();
                    throw new IOException("Interrupted waiting to read range", ie);
                }
                if (cursor == null) {
                    throw new IOException("Failed to open ledger range cursor, check logs");
                }
                return cursor.hasMoreEntries();
            }

            @Override
            public LedgerRange next() throws IOException {
                try {
                    SortedSet<Long> ledgerIds = new TreeSet<Long>();
                    Iterator<MetastoreTableItem> iter = cursor.readEntries(maxEntriesPerScan);
                    while (iter.hasNext()) {
                        ledgerIds.add(key2LedgerId(iter.next().getKey()));
                    }
                    return new LedgerRange(ledgerIds);
                } catch (MSException mse) {
                    LOG.error("Exception occurred reading from metastore", mse);
                    throw new IOException("Couldn't read from metastore", mse);
                }
            }
        }

        @Override
        public LedgerRangeIterator getLedgerRanges() {
            return new MSLedgerRangeIterator();
        }
    }

    @Override
    public LedgerManager newLedgerManager() {
        return new MsLedgerManager(conf, zk, metastore);
    }

    @Override
    public LedgerUnderreplicationManager newLedgerUnderreplicationManager() throws KeeperException,
            InterruptedException, ReplicationException.CompatibilityException {
        // TODO: currently just use zk ledger underreplication manager
        return new ZkLedgerUnderreplicationManager(conf, zk);
    }

    /**
     * Process set one by one in asynchronize way. Process will be stopped
     * immediately when error occurred.
     */
    private static class AsyncSetProcessor<T> {
        // use this to prevent long stack chains from building up in callbacks
        ScheduledExecutorService scheduler;

        /**
         * Constructor
         *
         * @param scheduler
         *            Executor used to prevent long stack chains
         */
        public AsyncSetProcessor(ScheduledExecutorService scheduler) {
            this.scheduler = scheduler;
        }

        /**
         * Process set of items
         *
         * @param data
         *            Set of data to process
         * @param processor
         *            Callback to process element of list when success
         * @param finalCb
         *            Final callback to be called after all elements in the list
         *            are processed
         * @param context
         *            Context of final callback
         * @param successRc
         *            RC passed to final callback on success
         * @param failureRc
         *            RC passed to final callback on failure
         */
        public void process(final Set<T> data, final Processor<T> processor, final AsyncCallback.VoidCallback finalCb,
                final Object context, final int successRc, final int failureRc) {
            if (data == null || data.size() == 0) {
                finalCb.processResult(successRc, null, context);
                return;
            }
            final Iterator<T> iter = data.iterator();
            AsyncCallback.VoidCallback stubCallback = new AsyncCallback.VoidCallback() {
                @Override
                public void processResult(int rc, String path, Object ctx) {
                    if (rc != successRc) {
                        // terminal immediately
                        finalCb.processResult(failureRc, null, context);
                        return;
                    }
                    if (!iter.hasNext()) { // reach the end of list
                        finalCb.processResult(successRc, null, context);
                        return;
                    }
                    // process next element
                    final T dataToProcess = iter.next();
                    final AsyncCallback.VoidCallback stub = this;
                    scheduler.submit(new Runnable() {
                        @Override
                        public final void run() {
                            processor.process(dataToProcess, stub);
                        }
                    });
                }
            };
            T firstElement = iter.next();
            processor.process(firstElement, stubCallback);
        }
    }

    @Override
    public void format(AbstractConfiguration conf, ZooKeeper zk) throws InterruptedException,
            KeeperException, IOException {
        MetastoreTable ledgerTable;
        try {
            ledgerTable = metastore.createScannableTable(TABLE_NAME);
        } catch (MetastoreException mse) {
            throw new IOException("Failed to instantiate table " + TABLE_NAME + " in metastore "
                    + metastore.getName());
        }
        try {
            MetastoreUtils.cleanTable(ledgerTable, conf.getMetastoreMaxEntriesPerScan());
        } catch (MSException mse) {
            throw new IOException("Exception when cleanning up table " + TABLE_NAME, mse);
        }
        LOG.info("Finished cleaning up table {}.", TABLE_NAME);
        // Delete and recreate the LAYOUT information.
        super.format(conf, zk);
    }

}