/**
 * 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.junit.Assert.assertTrue;

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

import io.netty.buffer.UnpooledByteBufAllocator;

import java.io.File;
import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.stream.IntStream;

import org.apache.bookkeeper.bookie.EntryLogManagerForEntryLogPerLedger.BufferedLogChannelWithDirInfo;
import org.apache.bookkeeper.bookie.EntryLogger.BufferedLogChannel;
import org.apache.bookkeeper.bookie.LedgerDirsManager.NoWritableLedgerDirException;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.conf.TestBKConfiguration;
import org.apache.bookkeeper.stats.Counter;
import org.apache.bookkeeper.test.TestStatsProvider;
import org.apache.bookkeeper.test.TestStatsProvider.TestOpStatsLogger;
import org.apache.bookkeeper.test.TestStatsProvider.TestStatsLogger;
import org.apache.bookkeeper.util.DiskChecker;
import org.apache.commons.lang.mutable.MutableInt;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Test new log creation.
 */
public class CreateNewLogTest {
    private static final Logger LOG = LoggerFactory
            .getLogger(CreateNewLogTest.class);

    private String[] ledgerDirs;
    private int numDirs = 100;

    @Before
    public void setUp() throws Exception{
        ledgerDirs = new String[numDirs];
        for (int i = 0; i < numDirs; i++){
            File temp = File.createTempFile("bookie", "test");
            temp.delete();
            temp.mkdir();
            File currentTemp = new File(temp.getAbsoluteFile() + "/current");
            currentTemp.mkdir();
            ledgerDirs[i] = temp.getPath();
        }
    }

    @After
    public void tearDown() throws Exception{
        for (int i = 0; i < numDirs; i++){
            File f = new File(ledgerDirs[i]);
            deleteRecursive(f);
        }
    }

    private void deleteRecursive(File f) {
        if (f.isDirectory()){
            for (File c : f.listFiles()){
                deleteRecursive(c);
            }
        }

        f.delete();
    }

    /**
     * Checks if new log file id is verified against all directories.
     *
     * {@link https://issues.apache.org/jira/browse/BOOKKEEPER-465}
     *
     * @throws Exception
     */
    @Test
    public void testCreateNewLog() throws Exception {
        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();

        // Creating a new configuration with a number of
        // ledger directories.
        conf.setLedgerDirNames(ledgerDirs);
        LedgerDirsManager ledgerDirsManager = new LedgerDirsManager(conf, conf.getLedgerDirs(),
                new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold()));

        // Extracted from createNewLog()
        String logFileName = Long.toHexString(1) + ".log";
        File dir = ledgerDirsManager.pickRandomWritableDir();
        LOG.info("Picked this directory: {}", dir);
        File newLogFile = new File(dir, logFileName);
        newLogFile.createNewFile();

        EntryLogger el = new EntryLogger(conf, ledgerDirsManager);
        // Calls createNewLog, and with the number of directories we
        // are using, if it picks one at random it will fail.
        EntryLogManagerForSingleEntryLog entryLogManager = (EntryLogManagerForSingleEntryLog) el.getEntryLogManager();
        entryLogManager.createNewLog(0L);
        LOG.info("This is the current log id: {}", entryLogManager.getCurrentLogId());
        assertTrue("Wrong log id", entryLogManager.getCurrentLogId() > 1);
    }

    @Test
    public void testCreateNewLogWithNoWritableLedgerDirs() throws Exception {
        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();

        // Creating a new configuration with a number of ledger directories.
        conf.setLedgerDirNames(ledgerDirs);
        conf.setIsForceGCAllowWhenNoSpace(true);
        LedgerDirsManager ledgerDirsManager = new LedgerDirsManager(conf, conf.getLedgerDirs(),
                new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold()));

        // Extracted from createNewLog()
        String logFileName = Long.toHexString(1) + ".log";
        File dir = ledgerDirsManager.pickRandomWritableDir();
        LOG.info("Picked this directory: {}", dir);
        File newLogFile = new File(dir, logFileName);
        newLogFile.createNewFile();

        // Now let us move all dirs to filled dirs
        List<File> wDirs = ledgerDirsManager.getWritableLedgerDirs();
        for (File tdir: wDirs) {
            ledgerDirsManager.addToFilledDirs(tdir);
        }

        EntryLogger el = new EntryLogger(conf, ledgerDirsManager);
        // Calls createNewLog, and with the number of directories we
        // are using, if it picks one at random it will fail.
        EntryLogManagerForSingleEntryLog entryLogManager = (EntryLogManagerForSingleEntryLog) el.getEntryLogManager();
        entryLogManager.createNewLog(0L);
        LOG.info("This is the current log id: {}", entryLogManager.getCurrentLogId());
        assertTrue("Wrong log id", entryLogManager.getCurrentLogId() > 1);
    }

    void setSameThreadExecutorForEntryLoggerAllocator(EntryLoggerAllocator entryLoggerAllocator) {
        ExecutorService executorService = entryLoggerAllocator.allocatorExecutor;
        executorService.shutdown();
        entryLoggerAllocator.allocatorExecutor = MoreExecutors.newDirectExecutorService();
    }

    /*
     * entryLogPerLedger is enabled and various scenarios of entrylogcreation are tested
     */
    @Test
    public void testEntryLogPerLedgerCreationWithPreAllocation() throws Exception {
        /*
         * I wish I could shorten this testcase or split it into multiple testcases,
         * but I want to cover a scenario and it requires multiple operations in
         * sequence and validations along the way. Please bear with the length of this
         * testcase, I added as many comments as I can to simplify it.
         */

        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();

        // Creating a new configuration with a number of ledger directories.
        conf.setLedgerDirNames(ledgerDirs);
        conf.setIsForceGCAllowWhenNoSpace(true);
        // preAllocation is Enabled
        conf.setEntryLogFilePreAllocationEnabled(true);
        conf.setEntryLogPerLedgerEnabled(true);
        LedgerDirsManager ledgerDirsManager = new LedgerDirsManager(conf, conf.getLedgerDirs(),
                new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold()));
        EntryLogger entryLogger = new EntryLogger(conf, ledgerDirsManager);
        EntryLoggerAllocator entryLoggerAllocator = entryLogger.entryLoggerAllocator;
        EntryLogManagerForEntryLogPerLedger entryLogManager = (EntryLogManagerForEntryLogPerLedger) entryLogger
                .getEntryLogManager();
        // set same thread executor for entryLoggerAllocator's allocatorExecutor
        setSameThreadExecutorForEntryLoggerAllocator(entryLoggerAllocator);

        /*
         * no entrylog will be created during initialization
         */
        int expectedPreAllocatedLogID = -1;
        Assert.assertEquals("PreallocatedlogId after initialization of Entrylogger",
                expectedPreAllocatedLogID, entryLoggerAllocator.getPreallocatedLogId());

        int numOfLedgers = 6;

        for (long i = 0; i < numOfLedgers; i++) {
            /* since we are starting creation of new ledgers, entrylogid will be ledgerid */
            entryLogManager.createNewLog(i);
        }

        /*
         * preallocation is enabled so though entryLogId starts with 0, preallocatedLogId would be equal to numOfLedgers
         */
        expectedPreAllocatedLogID = numOfLedgers;
        Assert.assertEquals("PreallocatedlogId after creation of logs for ledgers", expectedPreAllocatedLogID,
                entryLoggerAllocator.getPreallocatedLogId());
        Assert.assertEquals("Number of current ", numOfLedgers,
                entryLogManager.getCopyOfCurrentLogs().size());
        Assert.assertEquals("Number of LogChannels to flush", 0,
                entryLogManager.getRotatedLogChannels().size());

        // create dummy entrylog file with id - (expectedPreAllocatedLogID + 1)
        String logFileName = Long.toHexString(expectedPreAllocatedLogID + 1) + ".log";
        File dir = ledgerDirsManager.pickRandomWritableDir();
        LOG.info("Picked this directory: " + dir);
        File newLogFile = new File(dir, logFileName);
        newLogFile.createNewFile();

        /*
         * since there is already preexisting entrylog file with id -
         * (expectedPreAllocatedLogIDDuringInitialization + 1), when new
         * entrylog is created it should have
         * (expectedPreAllocatedLogIDDuringInitialization + 2) id
         */
        long rotatedLedger = 1L;
        entryLogManager.createNewLog(rotatedLedger);

        expectedPreAllocatedLogID = expectedPreAllocatedLogID + 2;
        Assert.assertEquals("PreallocatedlogId ",
                expectedPreAllocatedLogID, entryLoggerAllocator.getPreallocatedLogId());
        Assert.assertEquals("Number of current ", numOfLedgers,
                entryLogManager.getCopyOfCurrentLogs().size());
        List<BufferedLogChannel> rotatedLogChannels = entryLogManager.getRotatedLogChannels();
        Assert.assertEquals("Number of LogChannels rotated", 1, rotatedLogChannels.size());
        Assert.assertEquals("Rotated logchannel logid", rotatedLedger, rotatedLogChannels.iterator().next().getLogId());
        entryLogger.flush();
        /*
         * when flush is called all the rotatedlogchannels are flushed and
         * removed from rotatedlogchannels list. But here since entrylogId - 0,
         * is not yet rotated and flushed yet, getLeastUnflushedLogId will still
         * return 0.
         */
        rotatedLogChannels = entryLogManager.getRotatedLogChannels();
        Assert.assertEquals("Number of LogChannels rotated", 0, rotatedLogChannels.size());
        Assert.assertEquals("Least UnflushedLoggerId", 0, entryLogger.getLeastUnflushedLogId());

        entryLogManager.createNewLog(0L);
        rotatedLogChannels = entryLogManager.getRotatedLogChannels();
        Assert.assertEquals("Number of LogChannels rotated", 1, rotatedLogChannels.size());
        Assert.assertEquals("Least UnflushedLoggerId", 0, entryLogger.getLeastUnflushedLogId());
        entryLogger.flush();
        /*
         * since both entrylogids 0, 1 are rotated and flushed,
         * leastunFlushedLogId should be 2
         */
        Assert.assertEquals("Least UnflushedLoggerId", 2, entryLogger.getLeastUnflushedLogId());
        expectedPreAllocatedLogID = expectedPreAllocatedLogID + 1;

        /*
         * we should be able to get entryLogMetadata from all the active
         * entrylogs and the logs which are moved toflush list. Since no entry
         * is added, all the meta should be empty.
         */
        for (int i = 0; i <= expectedPreAllocatedLogID; i++) {
            EntryLogMetadata meta = entryLogger.getEntryLogMetadata(i);
            Assert.assertTrue("EntryLogMetadata should be empty", meta.isEmpty());
            Assert.assertTrue("EntryLog usage should be 0", meta.getTotalSize() == 0);
        }
    }

    /**
     * In this testcase entryLogPerLedger is Enabled and entrylogs are created
     * while ledgerdirs are getting full.
     */
    @Test
    public void testEntryLogCreationWithFilledDirs() throws Exception {
        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();

        // Creating a new configuration with a number of ledger directories.
        conf.setLedgerDirNames(ledgerDirs);
        // forceGCAllowWhenNoSpace is disabled
        conf.setIsForceGCAllowWhenNoSpace(false);
        // pre-allocation is not enabled
        conf.setEntryLogFilePreAllocationEnabled(false);
        conf.setEntryLogPerLedgerEnabled(true);
        LedgerDirsManager ledgerDirsManager = new LedgerDirsManager(conf, conf.getLedgerDirs(),
                new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold()));
        EntryLogger entryLogger = new EntryLogger(conf, ledgerDirsManager);
        EntryLoggerAllocator entryLoggerAllocator = entryLogger.entryLoggerAllocator;
        EntryLogManagerForEntryLogPerLedger entryLogManager = (EntryLogManagerForEntryLogPerLedger)
                entryLogger.getEntryLogManager();
        // set same thread executor for entryLoggerAllocator's allocatorExecutor
        setSameThreadExecutorForEntryLoggerAllocator(entryLoggerAllocator);

        int expectedPreAllocatedLogIDDuringInitialization = -1;
        Assert.assertEquals("PreallocatedlogId after initialization of Entrylogger",
                expectedPreAllocatedLogIDDuringInitialization, entryLoggerAllocator.getPreallocatedLogId());
        Assert.assertEquals("Preallocation Future of this slot should be null", null,
                entryLogger.entryLoggerAllocator.preallocation);

        long ledgerId = 0L;

        entryLogManager.createNewLog(ledgerId);

        /*
         * pre-allocation is not enabled, so it would not preallocate for next entrylog
         */
        Assert.assertEquals("PreallocatedlogId after initialization of Entrylogger",
                expectedPreAllocatedLogIDDuringInitialization + 1, entryLoggerAllocator.getPreallocatedLogId());

        for (int i = 0; i < numDirs - 1; i++) {
            ledgerDirsManager.addToFilledDirs(Bookie.getCurrentDirectory(new File(ledgerDirs[i])));
        }

        /*
         * this is the only non-filled ledgerDir so it should be used for creating new entryLog
         */
        File nonFilledLedgerDir = Bookie.getCurrentDirectory(new File(ledgerDirs[numDirs - 1]));

        entryLogManager.createNewLog(ledgerId);
        BufferedLogChannel newLogChannel = entryLogManager.getCurrentLogForLedger(ledgerId);
        Assert.assertEquals("Directory of newly created BufferedLogChannel file", nonFilledLedgerDir.getAbsolutePath(),
                newLogChannel.getLogFile().getParentFile().getAbsolutePath());

        ledgerDirsManager.addToFilledDirs(Bookie.getCurrentDirectory(new File(ledgerDirs[numDirs - 1])));

        // new entrylog creation should succeed, though there is no writable ledgerDir
        entryLogManager.createNewLog(ledgerId);
    }

    /*
     * In this testcase it is validated if the entryLog is created in the
     * ledgerDir with least number of current active entrylogs
     */
    @Test
    public void testLedgerDirsUniformityDuringCreation() throws Exception {
        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();

        // Creating a new configuration with a number of ledger directories.
        conf.setLedgerDirNames(ledgerDirs);
        // pre-allocation is not enabled
        conf.setEntryLogFilePreAllocationEnabled(false);
        conf.setEntryLogPerLedgerEnabled(true);
        LedgerDirsManager ledgerDirsManager = new LedgerDirsManager(conf, conf.getLedgerDirs(),
                new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold()));
        EntryLogger entryLogger = new EntryLogger(conf, ledgerDirsManager);
        EntryLogManagerForEntryLogPerLedger entrylogManager = (EntryLogManagerForEntryLogPerLedger)
                entryLogger.getEntryLogManager();

        for (long i = 0; i < ledgerDirs.length; i++) {
            entrylogManager.createNewLog(i);
        }

        int numberOfLedgersCreated = ledgerDirs.length;

        Assert.assertEquals("Highest frequency of entrylogs per ledgerdir", 1,
                highestFrequencyOfEntryLogsPerLedgerDir(entrylogManager.getCopyOfCurrentLogs()));

        long newLedgerId = numberOfLedgersCreated;
        entrylogManager.createNewLog(newLedgerId);
        numberOfLedgersCreated++;

        Assert.assertEquals("Highest frequency of entrylogs per ledgerdir", 2,
                highestFrequencyOfEntryLogsPerLedgerDir(entrylogManager.getCopyOfCurrentLogs()));

        for (long i = numberOfLedgersCreated; i < 2 * ledgerDirs.length; i++) {
            entrylogManager.createNewLog(i);
        }

        Assert.assertEquals("Highest frequency of entrylogs per ledgerdir", 2,
                highestFrequencyOfEntryLogsPerLedgerDir(entrylogManager.getCopyOfCurrentLogs()));
    }


    int highestFrequencyOfEntryLogsPerLedgerDir(Set<BufferedLogChannelWithDirInfo> copyOfCurrentLogsWithDirInfo) {
        Map<File, MutableInt> frequencyOfEntryLogsInLedgerDirs = new HashMap<File, MutableInt>();
        for (BufferedLogChannelWithDirInfo logChannelWithDirInfo : copyOfCurrentLogsWithDirInfo) {
            File parentDir = logChannelWithDirInfo.getLogChannel().getLogFile().getParentFile();
            if (frequencyOfEntryLogsInLedgerDirs.containsKey(parentDir)) {
                frequencyOfEntryLogsInLedgerDirs.get(parentDir).increment();
            } else {
                frequencyOfEntryLogsInLedgerDirs.put(parentDir, new MutableInt(1));
            }
        }
        @SuppressWarnings("unchecked")
        int highestFreq = ((Entry<File, MutableInt>) (frequencyOfEntryLogsInLedgerDirs.entrySet().stream()
                .max(Map.Entry.comparingByValue()).get())).getValue().intValue();
        return highestFreq;
    }

    @Test
    public void testConcurrentCreateNewLogWithEntryLogFilePreAllocationEnabled() throws Exception {
        testConcurrentCreateNewLog(true);
    }

    @Test
    public void testConcurrentCreateNewLogWithEntryLogFilePreAllocationDisabled() throws Exception {
        testConcurrentCreateNewLog(false);
    }

    public void testConcurrentCreateNewLog(boolean entryLogFilePreAllocationEnabled) throws Exception {
        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();

        // Creating a new configuration with a number of
        // ledger directories.
        conf.setLedgerDirNames(ledgerDirs);
        conf.setEntryLogFilePreAllocationEnabled(entryLogFilePreAllocationEnabled);
        LedgerDirsManager ledgerDirsManager = new LedgerDirsManager(conf, conf.getLedgerDirs(),
                new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold()));

        EntryLogger el = new EntryLogger(conf, ledgerDirsManager);
        EntryLogManagerBase entryLogManager = (EntryLogManagerBase) el.getEntryLogManager();
        // set same thread executor for entryLoggerAllocator's allocatorExecutor
        setSameThreadExecutorForEntryLoggerAllocator(el.getEntryLoggerAllocator());

        Assert.assertEquals("previousAllocatedEntryLogId after initialization", -1,
                el.getPreviousAllocatedEntryLogId());
        Assert.assertEquals("leastUnflushedLogId after initialization", 0, el.getLeastUnflushedLogId());
        int createNewLogNumOfTimes = 10;
        AtomicBoolean receivedException = new AtomicBoolean(false);

        IntStream.range(0, createNewLogNumOfTimes).parallel().forEach((i) -> {
            try {
                (entryLogManager).createNewLog((long) i);
            } catch (IOException e) {
                LOG.error("Received exception while creating newLog", e);
                receivedException.set(true);
            }
        });

        Assert.assertFalse("There shouldn't be any exceptions while creating newlog", receivedException.get());
        int expectedPreviousAllocatedEntryLogId = createNewLogNumOfTimes - 1;
        if (entryLogFilePreAllocationEnabled) {
            expectedPreviousAllocatedEntryLogId = createNewLogNumOfTimes;
        }

        Assert.assertEquals(
                "previousAllocatedEntryLogId after " + createNewLogNumOfTimes
                + " number of times createNewLog is called",
                expectedPreviousAllocatedEntryLogId, el.getPreviousAllocatedEntryLogId());
        Assert.assertEquals("Number of RotatedLogChannels", createNewLogNumOfTimes - 1,
                entryLogManager.getRotatedLogChannels().size());
    }

    @Test
    public void testCreateNewLogWithGaps() throws Exception {
        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();

        // Creating a new configuration with a number of
        // ledger directories.
        conf.setLedgerDirNames(ledgerDirs);
        conf.setEntryLogFilePreAllocationEnabled(false);
        LedgerDirsManager ledgerDirsManager = new LedgerDirsManager(conf, conf.getLedgerDirs(),
                new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold()));

        EntryLogger el = new EntryLogger(conf, ledgerDirsManager);
        EntryLogManagerBase entryLogManagerBase = (EntryLogManagerBase) el.getEntryLogManager();
        entryLogManagerBase.createNewLog(0L);

        Assert.assertEquals("previousAllocatedEntryLogId after initialization", 0, el.getPreviousAllocatedEntryLogId());

        // Extracted from createNewLog()
        String logFileName = Long.toHexString(1) + ".log";
        File dir = ledgerDirsManager.pickRandomWritableDir();
        LOG.info("Picked this directory: {}", dir);
        File newLogFile = new File(dir, logFileName);
        newLogFile.createNewFile();

        entryLogManagerBase.createNewLog(0L);
        Assert.assertEquals("previousAllocatedEntryLogId since entrylogid 1 is already taken", 2,
                el.getPreviousAllocatedEntryLogId());

        // Extracted from createNewLog()
        logFileName = Long.toHexString(3) + ".log";
        dir = ledgerDirsManager.pickRandomWritableDir();
        LOG.info("Picked this directory: {}", dir);
        newLogFile = new File(dir, logFileName);
        newLogFile.createNewFile();

        entryLogManagerBase.createNewLog(0L);
        Assert.assertEquals("previousAllocatedEntryLogId since entrylogid 3 is already taken", 4,
                el.getPreviousAllocatedEntryLogId());
    }

    @Test
    public void testCreateNewLogAndCompactionLog() throws Exception {
        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();

        // Creating a new configuration with a number of
        // ledger directories.
        conf.setLedgerDirNames(ledgerDirs);
        conf.setEntryLogFilePreAllocationEnabled(true);
        LedgerDirsManager ledgerDirsManager = new LedgerDirsManager(conf, conf.getLedgerDirs(),
                new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold()));
        EntryLogger el = new EntryLogger(conf, ledgerDirsManager);
        // set same thread executor for entryLoggerAllocator's allocatorExecutor
        setSameThreadExecutorForEntryLoggerAllocator(el.getEntryLoggerAllocator());
        AtomicBoolean receivedException = new AtomicBoolean(false);

        IntStream.range(0, 2).parallel().forEach((i) -> {
            try {
                if (i % 2 == 0) {
                    ((EntryLogManagerBase) el.getEntryLogManager()).createNewLog((long) i);
                } else {
                    el.createNewCompactionLog();
                }
            } catch (IOException e) {
                LOG.error("Received exception while creating newLog", e);
                receivedException.set(true);
            }
        });

        Assert.assertFalse("There shouldn't be any exceptions while creating newlog", receivedException.get());
        Assert.assertEquals(
                "previousAllocatedEntryLogId after 2 times createNewLog is called", 2,
                el.getPreviousAllocatedEntryLogId());
    }

    /*
     * In this testcase entrylogs for ledgers are tried to create concurrently.
     */
    @Test
    public void testConcurrentEntryLogCreations() throws Exception {
        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();

        // Creating a new configuration with a number of ledger directories.
        conf.setLedgerDirNames(ledgerDirs);
        // pre-allocation is enabled
        conf.setEntryLogFilePreAllocationEnabled(true);
        conf.setEntryLogPerLedgerEnabled(true);
        LedgerDirsManager ledgerDirsManager = new LedgerDirsManager(conf, conf.getLedgerDirs(),
                new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold()));
        EntryLogger entryLogger = new EntryLogger(conf, ledgerDirsManager);
        EntryLogManagerForEntryLogPerLedger entrylogManager = (EntryLogManagerForEntryLogPerLedger)
                entryLogger.getEntryLogManager();

        int numOfLedgers = 10;
        int numOfThreadsForSameLedger = 10;
        AtomicInteger createdEntryLogs = new AtomicInteger(0);
        CountDownLatch startLatch = new CountDownLatch(1);
        CountDownLatch createdLatch = new CountDownLatch(numOfLedgers * numOfThreadsForSameLedger);

        for (long i = 0; i < numOfLedgers; i++) {
            for (int j = 0; j < numOfThreadsForSameLedger; j++) {
                long ledgerId = i;
                new Thread(() -> {
                    try {
                        startLatch.await();
                        entrylogManager.createNewLog(ledgerId);
                        createdEntryLogs.incrementAndGet();
                        Thread.sleep(2000);
                    } catch (InterruptedException | IOException e) {
                        LOG.error("Got exception while trying to createNewLog for Ledger: " + ledgerId, e);
                    } finally {
                        createdLatch.countDown();
                    }
                }).start();
            }
        }

        startLatch.countDown();
        createdLatch.await(20, TimeUnit.SECONDS);
        Assert.assertEquals("Created EntryLogs", numOfLedgers * numOfThreadsForSameLedger, createdEntryLogs.get());
        Assert.assertEquals("Active currentlogs size", numOfLedgers, entrylogManager.getCopyOfCurrentLogs().size());
        Assert.assertEquals("Rotated entrylogs size", (numOfThreadsForSameLedger - 1) * numOfLedgers,
                entrylogManager.getRotatedLogChannels().size());
        /*
         * EntryLogFilePreAllocation is Enabled so
         * getPreviousAllocatedEntryLogId would be (numOfLedgers *
         * numOfThreadsForSameLedger) instead of (numOfLedgers *
         * numOfThreadsForSameLedger - 1)
         */
        Assert.assertEquals("PreviousAllocatedEntryLogId", numOfLedgers * numOfThreadsForSameLedger,
                entryLogger.getPreviousAllocatedEntryLogId());
    }

    /*
     * In this testcase metrics of EntryLogManagerForEntryLogPerLedger are
     * validated.
     */
    @Test
    public void testEntryLogManagerMetrics() throws Exception {
        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
        TestStatsProvider statsProvider = new TestStatsProvider();
        TestStatsLogger statsLogger = statsProvider.getStatsLogger(BookKeeperServerStats.ENTRYLOGGER_SCOPE);
        int maximumNumberOfActiveEntryLogs = 3;
        int entryLogPerLedgerCounterLimitsMultFactor = 2;

        // Creating a new configuration with a number of ledger directories.
        conf.setLedgerDirNames(ledgerDirs);
        // pre-allocation is enabled
        conf.setEntryLogFilePreAllocationEnabled(true);
        conf.setEntryLogPerLedgerEnabled(true);
        conf.setMaximumNumberOfActiveEntryLogs(maximumNumberOfActiveEntryLogs);
        conf.setEntryLogPerLedgerCounterLimitsMultFactor(entryLogPerLedgerCounterLimitsMultFactor);
        LedgerDirsManager ledgerDirsManager = new LedgerDirsManager(conf, conf.getLedgerDirs(),
                new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold()));
        EntryLogger entryLogger = new EntryLogger(conf, ledgerDirsManager, null, statsLogger,
                UnpooledByteBufAllocator.DEFAULT);
        EntryLogManagerForEntryLogPerLedger entrylogManager = (EntryLogManagerForEntryLogPerLedger) entryLogger
                .getEntryLogManager();
        // set same thread executor for entryLoggerAllocator's allocatorExecutor
        setSameThreadExecutorForEntryLoggerAllocator(entryLogger.getEntryLoggerAllocator());

        Counter numOfWriteActiveLedgers = statsLogger.getCounter(BookKeeperServerStats.NUM_OF_WRITE_ACTIVE_LEDGERS);
        Counter numOfWriteLedgersRemovedCacheExpiry = statsLogger
                .getCounter(BookKeeperServerStats.NUM_OF_WRITE_LEDGERS_REMOVED_CACHE_EXPIRY);
        Counter numOfWriteLedgersRemovedCacheMaxSize = statsLogger
                .getCounter(BookKeeperServerStats.NUM_OF_WRITE_LEDGERS_REMOVED_CACHE_MAXSIZE);
        Counter numLedgersHavingMultipleEntrylogs = statsLogger
                .getCounter(BookKeeperServerStats.NUM_LEDGERS_HAVING_MULTIPLE_ENTRYLOGS);
        TestOpStatsLogger entryLogsPerLedger = (TestOpStatsLogger) statsLogger
                .getOpStatsLogger(BookKeeperServerStats.ENTRYLOGS_PER_LEDGER);
        // initially all the counters should be 0
        Assert.assertEquals("NUM_OF_WRITE_ACTIVE_LEDGERS", 0, numOfWriteActiveLedgers.get().intValue());
        Assert.assertEquals("NUM_OF_WRITE_LEDGERS_REMOVED_CACHE_EXPIRY", 0,
                numOfWriteLedgersRemovedCacheExpiry.get().intValue());
        Assert.assertEquals("NUM_OF_WRITE_LEDGERS_REMOVED_CACHE_MAXSIZE", 0,
                numOfWriteLedgersRemovedCacheMaxSize.get().intValue());
        Assert.assertEquals("NUM_LEDGERS_HAVING_MULTIPLE_ENTRYLOGS", 0,
                numLedgersHavingMultipleEntrylogs.get().intValue());
        Assert.assertEquals("ENTRYLOGS_PER_LEDGER SuccessCount", 0, entryLogsPerLedger.getSuccessCount());

        // lid-1 : 3 entrylogs, lid-2 : 2 entrylogs, lid-3 : 1 entrylog
        int numOfEntrylogsForLedger1 = 3;
        createNewLogs(entrylogManager, 1L, numOfEntrylogsForLedger1);
        int numOfEntrylogsForLedger2 = 2;
        createNewLogs(entrylogManager, 2L, numOfEntrylogsForLedger2);
        createNewLogs(entrylogManager, 3L, 1);

        Assert.assertEquals("NUM_OF_WRITE_ACTIVE_LEDGERS", 3, numOfWriteActiveLedgers.get().intValue());
        Assert.assertEquals("NUM_OF_WRITE_LEDGERS_REMOVED_CACHE_EXPIRY", 0,
                numOfWriteLedgersRemovedCacheExpiry.get().intValue());
        Assert.assertEquals("NUM_OF_WRITE_LEDGERS_REMOVED_CACHE_MAXSIZE", 0,
                numOfWriteLedgersRemovedCacheMaxSize.get().intValue());
        Assert.assertEquals("NUM_LEDGERS_HAVING_MULTIPLE_ENTRYLOGS", 2,
                numLedgersHavingMultipleEntrylogs.get().intValue());
        Assert.assertEquals("ENTRYLOGS_PER_LEDGER SuccessCount", 0, entryLogsPerLedger.getSuccessCount());

        /*
         * since entrylog for lid-4 is created and entrylogmap cachesize is 3,
         * lid-1 will be removed from entrylogmap cache
         */
        createNewLogs(entrylogManager, 4L, 1);
        Assert.assertEquals("NUM_OF_WRITE_ACTIVE_LEDGERS", maximumNumberOfActiveEntryLogs,
                numOfWriteActiveLedgers.get().intValue());
        Assert.assertEquals("NUM_OF_WRITE_LEDGERS_REMOVED_CACHE_MAXSIZE", 1,
                numOfWriteLedgersRemovedCacheMaxSize.get().intValue());
        Assert.assertEquals("ENTRYLOGS_PER_LEDGER SuccessCount", 0, entryLogsPerLedger.getSuccessCount());

        /*
         * entrylog for lid-5, lid-6, lid-7 are created. Since
         * maximumNumberOfActiveEntryLogs = 3 and
         * entryLogPerLedgerCounterLimitsMultFactor = 2, when the entrylog for
         * lid-7 is created, count of lid-1 should be removed from countermap.
         */
        createNewLogs(entrylogManager, 5L, 1);
        createNewLogs(entrylogManager, 6L, 1);
        createNewLogs(entrylogManager, 7L, 1);
        Assert.assertEquals("NUM_OF_WRITE_ACTIVE_LEDGERS", maximumNumberOfActiveEntryLogs,
                numOfWriteActiveLedgers.get().intValue());
        Assert.assertEquals("NUM_OF_WRITE_LEDGERS_REMOVED_CACHE_MAXSIZE", 4,
                numOfWriteLedgersRemovedCacheMaxSize.get().intValue());
        Assert.assertEquals("ENTRYLOGS_PER_LEDGER SuccessCount", 1, entryLogsPerLedger.getSuccessCount());
        Assert.assertTrue("ENTRYLOGS_PER_LEDGER average value",
                Double.compare(numOfEntrylogsForLedger1, entryLogsPerLedger.getSuccessAverage()) == 0);

        /*
         * entrylog for new lid-8 is created so one more entry from countermap
         * should be removed.
         */
        createNewLogs(entrylogManager, 8L, 4);
        Assert.assertEquals("NUM_OF_WRITE_ACTIVE_LEDGERS", maximumNumberOfActiveEntryLogs,
                numOfWriteActiveLedgers.get().intValue());
        Assert.assertEquals("NUM_OF_WRITE_LEDGERS_REMOVED_CACHE_MAXSIZE", 5,
                numOfWriteLedgersRemovedCacheMaxSize.get().intValue());
        Assert.assertEquals("NUM_LEDGERS_HAVING_MULTIPLE_ENTRYLOGS", 3,
                numLedgersHavingMultipleEntrylogs.get().intValue());
        Assert.assertEquals("ENTRYLOGS_PER_LEDGER SuccessCount", 2, entryLogsPerLedger.getSuccessCount());
        Assert.assertTrue("ENTRYLOGS_PER_LEDGER average value",
                Double.compare((numOfEntrylogsForLedger1 + numOfEntrylogsForLedger2) / 2.0,
                        entryLogsPerLedger.getSuccessAverage()) == 0);

        /*
         * lid-3 is still in countermap. So when new entrylogs are created for
         * lid-3, no new entry from counter should be removed. so
         * entryLogsPerLedger.getSuccessCount() should be still old value. Also,
         * since lid-3 is still in countermap, these new 4 entrylogs should be
         * added to previous value 1 and hence the EntryLogsPerLedger for ledger
         * - 3l should be updated to 5.
         */
        createNewLogs(entrylogManager, 3L, 4);
        Assert.assertEquals("NUM_OF_WRITE_LEDGERS_REMOVED_CACHE_MAXSIZE", 6,
                numOfWriteLedgersRemovedCacheMaxSize.get().intValue());
        Assert.assertEquals("NUM_LEDGERS_HAVING_MULTIPLE_ENTRYLOGS", 4,
                numLedgersHavingMultipleEntrylogs.get().intValue());
        Assert.assertEquals("Numofentrylogs for ledger: 3l", 5,
                entrylogManager.entryLogsPerLedgerCounter.getCounterMap().get(3L).intValue());
        Assert.assertEquals("ENTRYLOGS_PER_LEDGER SuccessCount", 2, entryLogsPerLedger.getSuccessCount());
    }

    /*
     * In this testcase metrics of EntryLogManagerForEntryLogPerLedger are
     * validated.
     */
    @Test
    public void testEntryLogManagerMetricsFromExpiryAspect() throws Exception {
        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
        TestStatsProvider statsProvider = new TestStatsProvider();
        TestStatsLogger statsLogger = statsProvider.getStatsLogger(BookKeeperServerStats.ENTRYLOGGER_SCOPE);

        int entrylogMapAccessExpiryTimeInSeconds = 1;
        int entryLogPerLedgerCounterLimitsMultFactor = 2;

        // Creating a new configuration with a number of ledger directories.
        conf.setLedgerDirNames(ledgerDirs);
        // pre-allocation is enabled
        conf.setEntryLogFilePreAllocationEnabled(true);
        conf.setEntryLogPerLedgerEnabled(true);
        conf.setEntrylogMapAccessExpiryTimeInSeconds(entrylogMapAccessExpiryTimeInSeconds);
        conf.setEntryLogPerLedgerCounterLimitsMultFactor(entryLogPerLedgerCounterLimitsMultFactor);
        LedgerDirsManager ledgerDirsManager = new LedgerDirsManager(conf, conf.getLedgerDirs(),
                new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold()));
        EntryLogger entryLogger = new EntryLogger(conf, ledgerDirsManager, null, statsLogger,
                UnpooledByteBufAllocator.DEFAULT);
        EntryLogManagerForEntryLogPerLedger entrylogManager = (EntryLogManagerForEntryLogPerLedger) entryLogger
                .getEntryLogManager();
        // set same thread executor for entryLoggerAllocator's allocatorExecutor
        setSameThreadExecutorForEntryLoggerAllocator(entryLogger.getEntryLoggerAllocator());

        Counter numOfWriteLedgersRemovedCacheExpiry = statsLogger
                .getCounter(BookKeeperServerStats.NUM_OF_WRITE_LEDGERS_REMOVED_CACHE_EXPIRY);
        TestOpStatsLogger entryLogsPerLedger = (TestOpStatsLogger) statsLogger
                .getOpStatsLogger(BookKeeperServerStats.ENTRYLOGS_PER_LEDGER);

        int numOfEntrylogsForLedger1 = 3;
        createNewLogs(entrylogManager, 1L, numOfEntrylogsForLedger1);
        Assert.assertEquals("ENTRYLOGS_PER_LEDGER SuccessCount", 0, entryLogsPerLedger.getSuccessCount());
        Assert.assertEquals("NUM_OF_WRITE_LEDGERS_REMOVED_CACHE_EXPIRY", 0,
                numOfWriteLedgersRemovedCacheExpiry.get().intValue());

        Thread.sleep(entrylogMapAccessExpiryTimeInSeconds * 1000 + 100);
        entrylogManager.doEntryLogMapCleanup();
        entrylogManager.entryLogsPerLedgerCounter.doCounterMapCleanup();
        Assert.assertEquals("NUM_OF_WRITE_LEDGERS_REMOVED_CACHE_EXPIRY", 1,
                numOfWriteLedgersRemovedCacheExpiry.get().intValue());
        Assert.assertEquals("ENTRYLOGS_PER_LEDGER SuccessCount", 0, entryLogsPerLedger.getSuccessCount());

        Thread.sleep(entrylogMapAccessExpiryTimeInSeconds * 1000 + 100);
        entrylogManager.doEntryLogMapCleanup();
        entrylogManager.entryLogsPerLedgerCounter.doCounterMapCleanup();
        Assert.assertEquals("NUM_OF_WRITE_LEDGERS_REMOVED_CACHE_EXPIRY", 1,
                numOfWriteLedgersRemovedCacheExpiry.get().intValue());
        Assert.assertEquals("ENTRYLOGS_PER_LEDGER SuccessCount", 1, entryLogsPerLedger.getSuccessCount());
        Assert.assertTrue("ENTRYLOGS_PER_LEDGER average value",
                Double.compare(numOfEntrylogsForLedger1, entryLogsPerLedger.getSuccessAverage()) == 0);
    }

    private static void createNewLogs(EntryLogManagerForEntryLogPerLedger entrylogManager, long ledgerId,
            int numOfTimes) throws IOException {
        for (int i = 0; i < numOfTimes; i++) {
            entrylogManager.createNewLog(ledgerId);
        }
    }

    @Test
    public void testLockConsistency() throws Exception {
        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();

        conf.setLedgerDirNames(ledgerDirs);
        conf.setEntryLogFilePreAllocationEnabled(false);
        conf.setEntryLogPerLedgerEnabled(true);
        conf.setMaximumNumberOfActiveEntryLogs(5);

        CountDownLatch latch = new CountDownLatch(1);
        AtomicInteger count = new AtomicInteger(0);

        /*
         * Inject wait operation in 'getWritableLedgerDirsForNewLog' method of
         * ledgerDirsManager. getWritableLedgerDirsForNewLog will be called when
         * entryLogManager.createNewLog is called.
         */
        LedgerDirsManager ledgerDirsManager = new LedgerDirsManager(conf, conf.getLedgerDirs(),
                new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold())) {
            /*
             * getWritableLedgerDirsForNewLog is called for the first time, it
             * will await on 'latch' latch before calling super
             * getWritableLedgerDirsForNewLog.
             */
            @Override
            public List<File> getWritableLedgerDirsForNewLog() throws NoWritableLedgerDirException {
                if (count.incrementAndGet() == 1) {
                    try {
                        latch.await();
                    } catch (InterruptedException e) {
                        LOG.error("Got InterruptedException while awaiting for latch countdown", e);
                    }
                }
                return super.getWritableLedgerDirsForNewLog();
            }
        };

        EntryLogger el = new EntryLogger(conf, ledgerDirsManager);
        EntryLogManagerForEntryLogPerLedger entryLogManager = (EntryLogManagerForEntryLogPerLedger) el
                .getEntryLogManager();

        long firstLedgerId = 100L;
        AtomicBoolean newLogCreated = new AtomicBoolean(false);

        Assert.assertFalse("EntryLogManager cacheMap should not contain entry for firstLedgerId",
                entryLogManager.getCacheAsMap().containsKey(firstLedgerId));
        Assert.assertEquals("Value of the count should be 0", 0, count.get());
        /*
         * In a new thread, create newlog for 'firstLedgerId' and then set
         * 'newLogCreated' to true. Since this is the first createNewLog call,
         * it is going to be blocked untill latch is countdowned to 0.
         */
        new Thread() {
            @Override
            public void run() {
                try {
                    entryLogManager.createNewLog(firstLedgerId);
                    newLogCreated.set(true);
                } catch (IOException e) {
                    LOG.error("Got IOException while creating new log", e);
                }
            }
        }.start();

        /*
         * Wait until entry for 'firstLedgerId' is created in cacheMap. It will
         * be created because in the other thread createNewLog is called.
         */
        while (!entryLogManager.getCacheAsMap().containsKey(firstLedgerId)) {
            Thread.sleep(200);
        }
        Lock firstLedgersLock = entryLogManager.getLock(firstLedgerId);

        /*
         * since 'latch' is not counteddown, newlog should not be created even
         * after waitign for 2 secs.
         */
        Thread.sleep(2000);
        Assert.assertFalse("New log shouldn't have created", newLogCreated.get());

        /*
         * create MaximumNumberOfActiveEntryLogs of entrylogs and do cache
         * cleanup, so that the earliest entry from cache will be removed.
         */
        for (int i = 1; i <= conf.getMaximumNumberOfActiveEntryLogs(); i++) {
            entryLogManager.createNewLog(firstLedgerId + i);
        }
        entryLogManager.doEntryLogMapCleanup();
        Assert.assertFalse("Entry for that ledger shouldn't be there",
                entryLogManager.getCacheAsMap().containsKey(firstLedgerId));

        /*
         * now countdown the latch, so that the other thread can make progress
         * with createNewLog and since this entry is evicted from cache,
         * entrylog of the newly created entrylog will be added to
         * rotatedentrylogs.
         */
        latch.countDown();
        while (!newLogCreated.get()) {
            Thread.sleep(200);
        }
        while (entryLogManager.getRotatedLogChannels().size() < 1) {
            Thread.sleep(200);
        }

        /*
         * Entry for 'firstLedgerId' is removed from cache, but even in this
         * case when we get lock for the 'firstLedgerId' it should be the same
         * as we got earlier.
         */
        Lock lockForThatLedgerAfterRemoval = entryLogManager.getLock(firstLedgerId);
        Assert.assertEquals("For a given ledger lock should be the same before and after removal", firstLedgersLock,
                lockForThatLedgerAfterRemoval);
    }
}