/*
 *
 * 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.bookie;

import static org.apache.bookkeeper.bookie.BookKeeperServerStats.CATEGORY_SERVER;
import static org.apache.bookkeeper.bookie.BookKeeperServerStats.ENTRYLOGGER_SCOPE;
import static org.apache.bookkeeper.bookie.BookKeeperServerStats.ENTRYLOGS_PER_LEDGER;
import static org.apache.bookkeeper.bookie.BookKeeperServerStats.NUM_LEDGERS_HAVING_MULTIPLE_ENTRYLOGS;
import static org.apache.bookkeeper.bookie.BookKeeperServerStats.NUM_OF_WRITE_ACTIVE_LEDGERS;
import static org.apache.bookkeeper.bookie.BookKeeperServerStats.NUM_OF_WRITE_LEDGERS_REMOVED_CACHE_EXPIRY;
import static org.apache.bookkeeper.bookie.BookKeeperServerStats.NUM_OF_WRITE_LEDGERS_REMOVED_CACHE_MAXSIZE;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.cache.RemovalCause;
import com.google.common.cache.RemovalListener;
import com.google.common.cache.RemovalNotification;
import io.netty.buffer.ByteBuf;
import java.io.File;
import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReferenceArray;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.bookie.DefaultEntryLogger.BufferedLogChannel;
import org.apache.bookkeeper.bookie.LedgerDirsManager.LedgerDirsListener;
import org.apache.bookkeeper.common.util.MathUtils;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.stats.Counter;
import org.apache.bookkeeper.stats.OpStatsLogger;
import org.apache.bookkeeper.stats.StatsLogger;
import org.apache.bookkeeper.stats.annotations.StatsDoc;
import org.apache.bookkeeper.util.IOUtils;
import org.apache.bookkeeper.util.collections.ConcurrentLongHashMap;
import org.apache.commons.lang3.mutable.MutableInt;

@Slf4j
class EntryLogManagerForEntryLogPerLedger extends EntryLogManagerBase {

    static class BufferedLogChannelWithDirInfo {
        private final BufferedLogChannel logChannel;
        volatile boolean ledgerDirFull = false;

        private BufferedLogChannelWithDirInfo(BufferedLogChannel logChannel) {
            this.logChannel = logChannel;
        }

        private boolean isLedgerDirFull() {
            return ledgerDirFull;
        }

        private void setLedgerDirFull(boolean ledgerDirFull) {
            this.ledgerDirFull = ledgerDirFull;
        }

        BufferedLogChannel getLogChannel() {
            return logChannel;
        }
    }

    class EntryLogAndLockTuple {
        private final Lock ledgerLock;
        private BufferedLogChannelWithDirInfo entryLogWithDirInfo;

        private EntryLogAndLockTuple(long ledgerId) {
            int lockIndex = MathUtils.signSafeMod(Long.hashCode(ledgerId), lockArrayPool.length());
            if (lockArrayPool.get(lockIndex) == null) {
                lockArrayPool.compareAndSet(lockIndex, null, new ReentrantLock());
            }
            ledgerLock = lockArrayPool.get(lockIndex);
        }

        private Lock getLedgerLock() {
            return ledgerLock;
        }

        BufferedLogChannelWithDirInfo getEntryLogWithDirInfo() {
            return entryLogWithDirInfo;
        }

        private void setEntryLogWithDirInfo(BufferedLogChannelWithDirInfo entryLogWithDirInfo) {
            this.entryLogWithDirInfo = entryLogWithDirInfo;
        }
    }

    @StatsDoc(
        name = ENTRYLOGGER_SCOPE,
        category = CATEGORY_SERVER,
        help = "EntryLogger related stats"
    )
    class EntryLogsPerLedgerCounter {

        @StatsDoc(
            name = NUM_OF_WRITE_ACTIVE_LEDGERS,
            help = "Number of write active ledgers"
        )
        private final Counter numOfWriteActiveLedgers;
        @StatsDoc(
            name = NUM_OF_WRITE_LEDGERS_REMOVED_CACHE_EXPIRY,
            help = "Number of write ledgers removed after cache expiry"
        )
        private final Counter numOfWriteLedgersRemovedCacheExpiry;
        @StatsDoc(
            name = NUM_OF_WRITE_LEDGERS_REMOVED_CACHE_MAXSIZE,
            help = "Number of write ledgers removed due to reach max cache size"
        )
        private final Counter numOfWriteLedgersRemovedCacheMaxSize;
        @StatsDoc(
            name = NUM_LEDGERS_HAVING_MULTIPLE_ENTRYLOGS,
            help = "Number of ledgers having multiple entry logs"
        )
        private final Counter numLedgersHavingMultipleEntrylogs;
        @StatsDoc(
            name = ENTRYLOGS_PER_LEDGER,
            help = "The distribution of number of entry logs per ledger"
        )
        private final OpStatsLogger entryLogsPerLedger;
        /*
         * ledgerIdEntryLogCounterCacheMap cache will be used to store count of
         * entrylogs as value for its ledgerid key. This cacheMap limits -
         * 'expiry duration' and 'maximumSize' will be set to
         * entryLogPerLedgerCounterLimitsMultFactor times of
         * 'ledgerIdEntryLogMap' cache limits. This is needed because entries
         * from 'ledgerIdEntryLogMap' can be removed from cache because of
         * accesstime expiry or cache size limits, but to know the actual number
         * of entrylogs per ledger, we should maintain this count for long time.
         */
        private final LoadingCache<Long, MutableInt> ledgerIdEntryLogCounterCacheMap;

        EntryLogsPerLedgerCounter(StatsLogger statsLogger) {
            this.numOfWriteActiveLedgers = statsLogger.getCounter(NUM_OF_WRITE_ACTIVE_LEDGERS);
            this.numOfWriteLedgersRemovedCacheExpiry = statsLogger
                    .getCounter(NUM_OF_WRITE_LEDGERS_REMOVED_CACHE_EXPIRY);
            this.numOfWriteLedgersRemovedCacheMaxSize = statsLogger
                    .getCounter(NUM_OF_WRITE_LEDGERS_REMOVED_CACHE_MAXSIZE);
            this.numLedgersHavingMultipleEntrylogs = statsLogger.getCounter(NUM_LEDGERS_HAVING_MULTIPLE_ENTRYLOGS);
            this.entryLogsPerLedger = statsLogger.getOpStatsLogger(ENTRYLOGS_PER_LEDGER);

            ledgerIdEntryLogCounterCacheMap = CacheBuilder.newBuilder()
                    .expireAfterAccess(entrylogMapAccessExpiryTimeInSeconds * entryLogPerLedgerCounterLimitsMultFactor,
                            TimeUnit.SECONDS)
                    .maximumSize(maximumNumberOfActiveEntryLogs * entryLogPerLedgerCounterLimitsMultFactor)
                    .removalListener(new RemovalListener<Long, MutableInt>() {
                        @Override
                        public void onRemoval(RemovalNotification<Long, MutableInt> removedEntryFromCounterMap) {
                            if ((removedEntryFromCounterMap != null)
                                    && (removedEntryFromCounterMap.getValue() != null)) {
                                synchronized (EntryLogsPerLedgerCounter.this) {
                                    entryLogsPerLedger
                                            .registerSuccessfulValue(removedEntryFromCounterMap.getValue().intValue());
                                }
                            }
                        }
                    }).build(new CacheLoader<Long, MutableInt>() {
                        @Override
                        public MutableInt load(Long key) throws Exception {
                            synchronized (EntryLogsPerLedgerCounter.this) {
                                return new MutableInt();
                            }
                        }
                    });
        }

        private synchronized void openNewEntryLogForLedger(Long ledgerId, boolean newLedgerInEntryLogMapCache) {
            int numOfEntrylogsForThisLedger = ledgerIdEntryLogCounterCacheMap.getUnchecked(ledgerId).incrementAndGet();
            if (numOfEntrylogsForThisLedger == 2) {
                numLedgersHavingMultipleEntrylogs.inc();
            }
            if (newLedgerInEntryLogMapCache) {
                numOfWriteActiveLedgers.inc();
            }
        }

        private synchronized void removedLedgerFromEntryLogMapCache(Long ledgerId, RemovalCause cause) {
            numOfWriteActiveLedgers.dec();
            if (cause.equals(RemovalCause.EXPIRED)) {
                numOfWriteLedgersRemovedCacheExpiry.inc();
            } else if (cause.equals(RemovalCause.SIZE)) {
                numOfWriteLedgersRemovedCacheMaxSize.inc();
            }
        }

        /*
         * this is for testing purpose only. guava's cache doesnt cleanup
         * completely (including calling expiry removal listener) automatically
         * when access timeout elapses.
         *
         * https://google.github.io/guava/releases/19.0/api/docs/com/google/
         * common/cache/CacheBuilder.html
         *
         * If expireAfterWrite or expireAfterAccess is requested entries may be
         * evicted on each cache modification, on occasional cache accesses, or
         * on calls to Cache.cleanUp(). Expired entries may be counted by
         * Cache.size(), but will never be visible to read or write operations.
         *
         * Certain cache configurations will result in the accrual of periodic
         * maintenance tasks which will be performed during write operations, or
         * during occasional read operations in the absence of writes. The
         * Cache.cleanUp() method of the returned cache will also perform
         * maintenance, but calling it should not be necessary with a high
         * throughput cache. Only caches built with removalListener,
         * expireAfterWrite, expireAfterAccess, weakKeys, weakValues, or
         * softValues perform periodic maintenance.
         */
        @VisibleForTesting
        void doCounterMapCleanup() {
            ledgerIdEntryLogCounterCacheMap.cleanUp();
        }

        @VisibleForTesting
        ConcurrentMap<Long, MutableInt> getCounterMap() {
            return ledgerIdEntryLogCounterCacheMap.asMap();
        }
    }

    private final AtomicReferenceArray<Lock> lockArrayPool;
    private final LoadingCache<Long, EntryLogAndLockTuple> ledgerIdEntryLogMap;
    /*
     * every time active logChannel is accessed from ledgerIdEntryLogMap
     * cache, the accesstime of that entry is updated. But for certain
     * operations we dont want to impact accessTime of the entries (like
     * periodic flush of current active logChannels), and those operations
     * can use this copy of references.
     */
    private final ConcurrentLongHashMap<BufferedLogChannelWithDirInfo> replicaOfCurrentLogChannels;
    private final CacheLoader<Long, EntryLogAndLockTuple> entryLogAndLockTupleCacheLoader;
    private final DefaultEntryLogger.RecentEntryLogsStatus recentlyCreatedEntryLogsStatus;
    private final int entrylogMapAccessExpiryTimeInSeconds;
    private final int maximumNumberOfActiveEntryLogs;
    private final int entryLogPerLedgerCounterLimitsMultFactor;

    // Expose Stats
    private final StatsLogger statsLogger;
    final EntryLogsPerLedgerCounter entryLogsPerLedgerCounter;

    EntryLogManagerForEntryLogPerLedger(ServerConfiguration conf, LedgerDirsManager ledgerDirsManager,
                                        EntryLoggerAllocator entryLoggerAllocator,
                                        List<DefaultEntryLogger.EntryLogListener> listeners,
                                        DefaultEntryLogger.RecentEntryLogsStatus recentlyCreatedEntryLogsStatus,
                                        StatsLogger statsLogger) throws IOException {
        super(conf, ledgerDirsManager, entryLoggerAllocator, listeners);
        this.recentlyCreatedEntryLogsStatus = recentlyCreatedEntryLogsStatus;
        this.rotatedLogChannels = new CopyOnWriteArrayList<BufferedLogChannel>();
        this.replicaOfCurrentLogChannels =
                ConcurrentLongHashMap.<BufferedLogChannelWithDirInfo>newBuilder().build();
        this.entrylogMapAccessExpiryTimeInSeconds = conf.getEntrylogMapAccessExpiryTimeInSeconds();
        this.maximumNumberOfActiveEntryLogs = conf.getMaximumNumberOfActiveEntryLogs();
        this.entryLogPerLedgerCounterLimitsMultFactor = conf.getEntryLogPerLedgerCounterLimitsMultFactor();

        ledgerDirsManager.addLedgerDirsListener(getLedgerDirsListener());
        this.lockArrayPool = new AtomicReferenceArray<Lock>(maximumNumberOfActiveEntryLogs * 2);
        this.entryLogAndLockTupleCacheLoader = new CacheLoader<Long, EntryLogAndLockTuple>() {
            @Override
            public EntryLogAndLockTuple load(Long key) throws Exception {
                return new EntryLogAndLockTuple(key);
            }
        };
        /*
         * Currently we are relying on access time based eviction policy for
         * removal of EntryLogAndLockTuple, so if the EntryLogAndLockTuple of
         * the ledger is not accessed in
         * entrylogMapAccessExpiryTimeInSeconds period, it will be removed
         * from the cache.
         *
         * We are going to introduce explicit advisory writeClose call, with
         * that explicit call EntryLogAndLockTuple of the ledger will be
         * removed from the cache. But still timebased eviciton policy is
         * needed because it is not guaranteed that Bookie/EntryLogger would
         * receive successfully write close call in all the cases.
         */
        ledgerIdEntryLogMap = CacheBuilder.newBuilder()
                .expireAfterAccess(entrylogMapAccessExpiryTimeInSeconds, TimeUnit.SECONDS)
                .maximumSize(maximumNumberOfActiveEntryLogs)
                .removalListener(new RemovalListener<Long, EntryLogAndLockTuple>() {
                    @Override
                    public void onRemoval(
                            RemovalNotification<Long, EntryLogAndLockTuple> expiredLedgerEntryLogMapEntry) {
                        onCacheEntryRemoval(expiredLedgerEntryLogMapEntry);
                    }
                }).build(entryLogAndLockTupleCacheLoader);

        this.statsLogger = statsLogger;
        this.entryLogsPerLedgerCounter = new EntryLogsPerLedgerCounter(this.statsLogger);
    }

    /*
     * This method is called when an entry is removed from the cache. This could
     * be because access time of that ledger has elapsed
     * entrylogMapAccessExpiryTimeInSeconds period, or number of active
     * currentlogs in the cache has reached the size of
     * maximumNumberOfActiveEntryLogs, or if an entry is explicitly
     * invalidated/removed. In these cases entry for that ledger is removed from
     * cache. Since the entrylog of this ledger is not active anymore it has to
     * be removed from replicaOfCurrentLogChannels and added to
     * rotatedLogChannels.
     *
     * Because of performance/optimizations concerns the cleanup maintenance
     * operations wont happen automatically, for more info on eviction cleanup
     * maintenance tasks -
     * https://google.github.io/guava/releases/19.0/api/docs/com/google/
     * common/cache/CacheBuilder.html
     *
     */
    private void onCacheEntryRemoval(RemovalNotification<Long, EntryLogAndLockTuple> removedLedgerEntryLogMapEntry) {
        Long ledgerId = removedLedgerEntryLogMapEntry.getKey();
        if (log.isDebugEnabled()) {
            log.debug("LedgerId {} is being evicted from the cache map because of {}", ledgerId,
                    removedLedgerEntryLogMapEntry.getCause());
        }
        EntryLogAndLockTuple entryLogAndLockTuple = removedLedgerEntryLogMapEntry.getValue();
        if (entryLogAndLockTuple == null) {
            log.error("entryLogAndLockTuple is not supposed to be null in entry removal listener for ledger : {}",
                    ledgerId);
            return;
        }
        Lock lock = entryLogAndLockTuple.ledgerLock;
        BufferedLogChannelWithDirInfo logChannelWithDirInfo = entryLogAndLockTuple.getEntryLogWithDirInfo();
        if (logChannelWithDirInfo == null) {
            log.error("logChannel for ledger: {} is not supposed to be null in entry removal listener", ledgerId);
            return;
        }
        lock.lock();
        try {
            BufferedLogChannel logChannel = logChannelWithDirInfo.getLogChannel();
            // Append ledgers map at the end of entry log
            try {
                logChannel.appendLedgersMap();
            } catch (Exception e) {
                log.error("Got IOException while trying to appendLedgersMap in cacheEntryRemoval callback", e);
            }
            replicaOfCurrentLogChannels.remove(logChannel.getLogId());
            rotatedLogChannels.add(logChannel);
            entryLogsPerLedgerCounter.removedLedgerFromEntryLogMapCache(ledgerId,
                    removedLedgerEntryLogMapEntry.getCause());
        } finally {
            lock.unlock();
        }
    }

    private LedgerDirsListener getLedgerDirsListener() {
        return new LedgerDirsListener() {
            @Override
            public void diskFull(File disk) {
                Set<BufferedLogChannelWithDirInfo> copyOfCurrentLogsWithDirInfo = getCopyOfCurrentLogs();
                for (BufferedLogChannelWithDirInfo currentLogWithDirInfo : copyOfCurrentLogsWithDirInfo) {
                    if (disk.equals(currentLogWithDirInfo.getLogChannel().getLogFile().getParentFile())) {
                        currentLogWithDirInfo.setLedgerDirFull(true);
                    }
                }
            }

            @Override
            public void diskWritable(File disk) {
                Set<BufferedLogChannelWithDirInfo> copyOfCurrentLogsWithDirInfo = getCopyOfCurrentLogs();
                for (BufferedLogChannelWithDirInfo currentLogWithDirInfo : copyOfCurrentLogsWithDirInfo) {
                    if (disk.equals(currentLogWithDirInfo.getLogChannel().getLogFile().getParentFile())) {
                        currentLogWithDirInfo.setLedgerDirFull(false);
                    }
                }
            }
        };
    }

    Lock getLock(long ledgerId) throws IOException {
        try {
            return ledgerIdEntryLogMap.get(ledgerId).getLedgerLock();
        } catch (Exception e) {
            log.error("Received unexpected exception while fetching lock to acquire for ledger: " + ledgerId, e);
            throw new IOException("Received unexpected exception while fetching lock to acquire", e);
        }
    }

    /*
     * sets the logChannel for the given ledgerId. It will add the new
     * logchannel to replicaOfCurrentLogChannels, and the previous one will
     * be removed from replicaOfCurrentLogChannels. Previous logChannel will
     * be added to rotatedLogChannels in both the cases.
     */
    @Override
    public void setCurrentLogForLedgerAndAddToRotate(long ledgerId, BufferedLogChannel logChannel) throws IOException {
        Lock lock = getLock(ledgerId);
        lock.lock();
        try {
            BufferedLogChannel hasToRotateLogChannel = getCurrentLogForLedger(ledgerId);
            boolean newLedgerInEntryLogMapCache = (hasToRotateLogChannel == null);
            logChannel.setLedgerIdAssigned(ledgerId);
            BufferedLogChannelWithDirInfo logChannelWithDirInfo = new BufferedLogChannelWithDirInfo(logChannel);
            ledgerIdEntryLogMap.get(ledgerId).setEntryLogWithDirInfo(logChannelWithDirInfo);
            entryLogsPerLedgerCounter.openNewEntryLogForLedger(ledgerId, newLedgerInEntryLogMapCache);
            replicaOfCurrentLogChannels.put(logChannel.getLogId(), logChannelWithDirInfo);
            if (hasToRotateLogChannel != null) {
                replicaOfCurrentLogChannels.remove(hasToRotateLogChannel.getLogId());
                rotatedLogChannels.add(hasToRotateLogChannel);
            }
        } catch (Exception e) {
            log.error("Received unexpected exception while fetching entry from map for ledger: " + ledgerId, e);
            throw new IOException("Received unexpected exception while fetching entry from map", e);
        } finally {
            lock.unlock();
        }
    }

    @Override
    public BufferedLogChannel getCurrentLogForLedger(long ledgerId) throws IOException {
        BufferedLogChannelWithDirInfo bufferedLogChannelWithDirInfo = getCurrentLogWithDirInfoForLedger(ledgerId);
        BufferedLogChannel bufferedLogChannel = null;
        if (bufferedLogChannelWithDirInfo != null) {
            bufferedLogChannel = bufferedLogChannelWithDirInfo.getLogChannel();
        }
        return bufferedLogChannel;
    }

    public BufferedLogChannelWithDirInfo getCurrentLogWithDirInfoForLedger(long ledgerId) throws IOException {
        Lock lock = getLock(ledgerId);
        lock.lock();
        try {
            EntryLogAndLockTuple entryLogAndLockTuple = ledgerIdEntryLogMap.get(ledgerId);
            return entryLogAndLockTuple.getEntryLogWithDirInfo();
        } catch (Exception e) {
            log.error("Received unexpected exception while fetching entry from map for ledger: " + ledgerId, e);
            throw new IOException("Received unexpected exception while fetching entry from map", e);
        } finally {
            lock.unlock();
        }
    }

    public Set<BufferedLogChannelWithDirInfo> getCopyOfCurrentLogs() {
        return new HashSet<BufferedLogChannelWithDirInfo>(replicaOfCurrentLogChannels.values());
    }

    @Override
    public BufferedLogChannel getCurrentLogIfPresent(long entryLogId) {
        BufferedLogChannelWithDirInfo bufferedLogChannelWithDirInfo = replicaOfCurrentLogChannels.get(entryLogId);
        BufferedLogChannel logChannel = null;
        if (bufferedLogChannelWithDirInfo != null) {
            logChannel = bufferedLogChannelWithDirInfo.getLogChannel();
        }
        return logChannel;
    }

    @Override
    public void checkpoint() throws IOException {
        /*
         * In the case of entryLogPerLedgerEnabled we need to flush
         * both rotatedlogs and currentlogs. This is needed because
         * syncThread periodically does checkpoint and at this time
         * all the logs should be flushed.
         *
         */
        super.flush();
    }

    @Override
    public void prepareSortedLedgerStorageCheckpoint(long numBytesFlushed) throws IOException {
        // do nothing
        /*
         * prepareSortedLedgerStorageCheckpoint is required for
         * singleentrylog scenario, but it is not needed for
         * entrylogperledger scenario, since entries of a ledger go
         * to a entrylog (even during compaction) and SyncThread
         * drives periodic checkpoint logic.
         */

    }

    @Override
    public void prepareEntryMemTableFlush() {
        // do nothing
    }

    @Override
    public boolean commitEntryMemTableFlush() throws IOException {
        // lock it only if there is new data
        // so that cache accesstime is not changed
        Set<BufferedLogChannelWithDirInfo> copyOfCurrentLogsWithDirInfo = getCopyOfCurrentLogs();
        for (BufferedLogChannelWithDirInfo currentLogWithDirInfo : copyOfCurrentLogsWithDirInfo) {
            BufferedLogChannel currentLog = currentLogWithDirInfo.getLogChannel();
            if (reachEntryLogLimit(currentLog, 0L)) {
                Long ledgerId = currentLog.getLedgerIdAssigned();
                Lock lock = getLock(ledgerId);
                lock.lock();
                try {
                    if (reachEntryLogLimit(currentLog, 0L)) {
                        log.info("Rolling entry logger since it reached size limitation for ledger: {}", ledgerId);
                        createNewLog(ledgerId, "after entry log file is rotated");
                    }
                } finally {
                    lock.unlock();
                }
            }
        }
        /*
         * in the case of entrylogperledger, SyncThread drives
         * checkpoint logic for every flushInterval. So
         * EntryMemtable doesn't need to call checkpoint in the case
         * of entrylogperledger.
         */
        return false;
    }

    /*
     * this is for testing purpose only. guava's cache doesnt cleanup
     * completely (including calling expiry removal listener) automatically
     * when access timeout elapses.
     *
     * https://google.github.io/guava/releases/19.0/api/docs/com/google/
     * common/cache/CacheBuilder.html
     *
     * If expireAfterWrite or expireAfterAccess is requested entries may be
     * evicted on each cache modification, on occasional cache accesses, or
     * on calls to Cache.cleanUp(). Expired entries may be counted by
     * Cache.size(), but will never be visible to read or write operations.
     *
     * Certain cache configurations will result in the accrual of periodic
     * maintenance tasks which will be performed during write operations, or
     * during occasional read operations in the absence of writes. The
     * Cache.cleanUp() method of the returned cache will also perform
     * maintenance, but calling it should not be necessary with a high
     * throughput cache. Only caches built with removalListener,
     * expireAfterWrite, expireAfterAccess, weakKeys, weakValues, or
     * softValues perform periodic maintenance.
     */
    @VisibleForTesting
    void doEntryLogMapCleanup() {
        ledgerIdEntryLogMap.cleanUp();
    }

    @VisibleForTesting
    ConcurrentMap<Long, EntryLogAndLockTuple> getCacheAsMap() {
        return ledgerIdEntryLogMap.asMap();
    }
    /*
     * Returns writable ledger dir with least number of current active
     * entrylogs.
     */
    @Override
    public File getDirForNextEntryLog(List<File> writableLedgerDirs) {
        Map<File, MutableInt> writableLedgerDirFrequency = new HashMap<File, MutableInt>();
        writableLedgerDirs.stream()
                .forEach((ledgerDir) -> writableLedgerDirFrequency.put(ledgerDir, new MutableInt()));
        for (BufferedLogChannelWithDirInfo logChannelWithDirInfo : replicaOfCurrentLogChannels.values()) {
            File parentDirOfCurrentLogChannel = logChannelWithDirInfo.getLogChannel().getLogFile().getParentFile();
            if (writableLedgerDirFrequency.containsKey(parentDirOfCurrentLogChannel)) {
                writableLedgerDirFrequency.get(parentDirOfCurrentLogChannel).increment();
            }
        }
        @SuppressWarnings("unchecked")
        Optional<Entry<File, MutableInt>> ledgerDirWithLeastNumofCurrentLogs = writableLedgerDirFrequency.entrySet()
                .stream().min(Map.Entry.comparingByValue());
        return ledgerDirWithLeastNumofCurrentLogs.get().getKey();
    }

    @Override
    public void close() throws IOException {
        Set<BufferedLogChannelWithDirInfo> copyOfCurrentLogsWithDirInfo = getCopyOfCurrentLogs();
        for (BufferedLogChannelWithDirInfo currentLogWithDirInfo : copyOfCurrentLogsWithDirInfo) {
            if (currentLogWithDirInfo.getLogChannel() != null) {
                currentLogWithDirInfo.getLogChannel().close();
            }
        }
    }

    @Override
    public void forceClose() {
        Set<BufferedLogChannelWithDirInfo> copyOfCurrentLogsWithDirInfo = getCopyOfCurrentLogs();
        for (BufferedLogChannelWithDirInfo currentLogWithDirInfo : copyOfCurrentLogsWithDirInfo) {
            IOUtils.close(log, currentLogWithDirInfo.getLogChannel());
        }
    }

    @Override
    void flushCurrentLogs() throws IOException {
        Set<BufferedLogChannelWithDirInfo> copyOfCurrentLogsWithDirInfo = getCopyOfCurrentLogs();
        for (BufferedLogChannelWithDirInfo logChannelWithDirInfo : copyOfCurrentLogsWithDirInfo) {
            /**
             * flushCurrentLogs method is called during checkpoint, so metadata
             * of the file also should be force written.
             */
            flushLogChannel(logChannelWithDirInfo.getLogChannel(), true);
        }
    }

    @Override
    public BufferedLogChannel createNewLogForCompaction() throws IOException {
        throw new UnsupportedOperationException(
                "When entryLogPerLedger is enabled, transactional compaction should have been disabled");
    }

    @Override
    public long addEntry(long ledger, ByteBuf entry, boolean rollLog) throws IOException {
        Lock lock = getLock(ledger);
        lock.lock();
        try {
            return super.addEntry(ledger, entry, rollLog);
        } finally {
            lock.unlock();
        }
    }

    @Override
    void createNewLog(long ledgerId) throws IOException {
        Lock lock = getLock(ledgerId);
        lock.lock();
        try {
            super.createNewLog(ledgerId);
        } finally {
            lock.unlock();
        }
    }


    @Override
    BufferedLogChannel getCurrentLogForLedgerForAddEntry(long ledgerId, int entrySize, boolean rollLog)
            throws IOException {
        Lock lock = getLock(ledgerId);
        lock.lock();
        try {
            BufferedLogChannelWithDirInfo logChannelWithDirInfo = getCurrentLogWithDirInfoForLedger(ledgerId);
            BufferedLogChannel logChannel = null;
            if (logChannelWithDirInfo != null) {
                logChannel = logChannelWithDirInfo.getLogChannel();
            }
            boolean reachEntryLogLimit = rollLog ? reachEntryLogLimit(logChannel, entrySize)
                    : readEntryLogHardLimit(logChannel, entrySize);
            // Create new log if logSizeLimit reached or current disk is full
            boolean diskFull = (logChannel == null) ? false : logChannelWithDirInfo.isLedgerDirFull();
            boolean allDisksFull = !ledgerDirsManager.hasWritableLedgerDirs();

            /**
             * if disk of the logChannel is full or if the entrylog limit is
             * reached of if the logchannel is not initialized, then
             * createNewLog. If allDisks are full then proceed with the current
             * logChannel, since Bookie must have turned to readonly mode and
             * the addEntry traffic would be from GC and it is ok to proceed in
             * this case.
             */
            if ((diskFull && (!allDisksFull)) || reachEntryLogLimit || (logChannel == null)) {
                if (logChannel != null) {
                    logChannel.flushAndForceWriteIfRegularFlush(false);
                }
                createNewLog(ledgerId,
                    ": diskFull = " + diskFull + ", allDisksFull = " + allDisksFull
                        + ", reachEntryLogLimit = " + reachEntryLogLimit + ", logChannel = " + logChannel);
            }

            return getCurrentLogForLedger(ledgerId);
        } finally {
            lock.unlock();
        }
    }

    @Override
    public void flushRotatedLogs() throws IOException {
        for (BufferedLogChannel channel : rotatedLogChannels) {
            channel.flushAndForceWrite(true);
            // since this channel is only used for writing, after flushing the channel,
            // we had to close the underlying file channel. Otherwise, we might end up
            // leaking fds which cause the disk spaces could not be reclaimed.
            channel.close();
            recentlyCreatedEntryLogsStatus.flushRotatedEntryLog(channel.getLogId());
            rotatedLogChannels.remove(channel);
            log.info("Synced entry logger {} to disk.", channel.getLogId());
        }
    }
}