/**
 *
 * 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 java.nio.charset.StandardCharsets.UTF_8;
import static org.apache.bookkeeper.bookie.BookieJournalTest.writeV5Journal;
import static org.apache.bookkeeper.meta.MetadataDrivers.runFunctionWithRegistrationManager;
import static org.apache.bookkeeper.util.BookKeeperConstants.AVAILABLE_NODE;
import static org.apache.bookkeeper.util.BookKeeperConstants.BOOKIE_STATUS_FILENAME;
import static org.apache.bookkeeper.util.TestUtils.countNumOfFiles;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.hasItem;
import static org.hamcrest.Matchers.hasProperty;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;

import com.fasterxml.jackson.databind.ObjectMapper;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.UnpooledByteBufAllocator;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.net.BindException;
import java.net.InetAddress;
import java.net.URL;
import java.net.URLConnection;
import java.security.AccessControlException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Random;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;

import org.apache.bookkeeper.bookie.BookieException.DiskPartitionDuplicationException;
import org.apache.bookkeeper.bookie.BookieException.MetadataStoreException;
import org.apache.bookkeeper.bookie.Journal.LastLogMark;
import org.apache.bookkeeper.bookie.LedgerDirsManager.NoWritableLedgerDirException;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.BookKeeper.DigestType;
import org.apache.bookkeeper.client.BookKeeperAdmin;
import org.apache.bookkeeper.client.BookKeeperClientStats;
import org.apache.bookkeeper.client.LedgerHandle;
import org.apache.bookkeeper.common.component.ComponentStarter;
import org.apache.bookkeeper.common.component.Lifecycle;
import org.apache.bookkeeper.common.component.LifecycleComponent;
import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.conf.TestBKConfiguration;
import org.apache.bookkeeper.discover.BookieServiceInfo;
import org.apache.bookkeeper.discover.BookieServiceInfo.Endpoint;
import org.apache.bookkeeper.discover.RegistrationManager;
import org.apache.bookkeeper.http.HttpRouter;
import org.apache.bookkeeper.http.HttpServerLoader;
import org.apache.bookkeeper.meta.MetadataBookieDriver;
import org.apache.bookkeeper.meta.exceptions.MetadataException;
import org.apache.bookkeeper.meta.zk.ZKMetadataBookieDriver;
import org.apache.bookkeeper.meta.zk.ZKMetadataDriverBase;
import org.apache.bookkeeper.net.BookieId;
import org.apache.bookkeeper.proto.BookieServer;
import org.apache.bookkeeper.proto.DataFormats.BookieServiceInfoFormat;
import org.apache.bookkeeper.replication.AutoRecoveryMain;
import org.apache.bookkeeper.replication.ReplicationException.CompatibilityException;
import org.apache.bookkeeper.replication.ReplicationException.UnavailableException;
import org.apache.bookkeeper.replication.ReplicationStats;
import org.apache.bookkeeper.server.Main;
import org.apache.bookkeeper.server.conf.BookieConfiguration;
import org.apache.bookkeeper.server.service.AutoRecoveryService;
import org.apache.bookkeeper.server.service.BookieService;
import org.apache.bookkeeper.stats.NullStatsLogger;
import org.apache.bookkeeper.stats.StatsLogger;
import org.apache.bookkeeper.stats.prometheus.PrometheusMetricsProvider;
import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
import org.apache.bookkeeper.tls.SecurityException;
import org.apache.bookkeeper.util.DiskChecker;
import org.apache.bookkeeper.util.LoggerOutput;
import org.apache.bookkeeper.util.PortManager;
import org.apache.bookkeeper.versioning.Version;
import org.apache.bookkeeper.versioning.Versioned;
import org.apache.bookkeeper.zookeeper.ZooKeeperClient;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.data.Stat;
import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestName;
import org.powermock.reflect.Whitebox;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.event.LoggingEvent;

/**
 * Testing bookie initialization cases.
 */
public class BookieInitializationTest extends BookKeeperClusterTestCase {
    private static final Logger LOG = LoggerFactory
            .getLogger(BookieInitializationTest.class);

    private static ObjectMapper om = new ObjectMapper();

    @Rule
    public final TestName runtime = new TestName();
    @Rule
    public LoggerOutput loggerOutput = new LoggerOutput();
    ZKMetadataBookieDriver driver;

    public BookieInitializationTest() {
        super(0);
    }

    @Override
    public void setUp() throws Exception {
        String ledgersPath = "/ledgers" + runtime.getMethodName();
        super.setUp(ledgersPath);
        zkUtil.createBKEnsemble(ledgersPath);
        driver = new ZKMetadataBookieDriver();
    }

    @Override
    public void tearDown() throws Exception {
        if (driver != null) {
            driver.close();
        }
        super.tearDown();
    }

    @Test
    public void testOneJournalReplayForBookieRestartInReadOnlyMode() throws Exception {
        testJournalReplayForBookieRestartInReadOnlyMode(1);
    }

    @Test
    public void testMultipleJournalReplayForBookieRestartInReadOnlyMode() throws Exception {
        testJournalReplayForBookieRestartInReadOnlyMode(4);
    }

    /**
     * Tests that journal replay works correctly when bookie crashes and starts up in RO mode.
     */
    private void testJournalReplayForBookieRestartInReadOnlyMode(int numOfJournalDirs) throws Exception {
        File tmpLedgerDir = createTempDir("DiskCheck", "test");
        File tmpJournalDir = createTempDir("DiskCheck", "test");

        String[] journalDirs = new String[numOfJournalDirs];
        for (int i = 0; i < numOfJournalDirs; i++) {
            journalDirs[i] = tmpJournalDir.getAbsolutePath() + "/journal-" + i;
        }

        final ServerConfiguration conf = newServerConfiguration()
                .setJournalDirsName(journalDirs)
                .setLedgerDirNames(new String[] { tmpLedgerDir.getPath() })
                .setDiskCheckInterval(1000)
                .setLedgerStorageClass(SortedLedgerStorage.class.getName())
                .setAutoRecoveryDaemonEnabled(false)
                .setZkTimeout(5000);

        BookieServer server = new MockBookieServer(conf);
        server.start();

        List<LastLogMark> lastLogMarkList = new ArrayList<>(journalDirs.length);

        for (int i = 0; i < journalDirs.length; i++) {
            Journal journal = ((BookieImpl) server.getBookie()).journals.get(i);
            // LastLogMark should be (0, 0) at the bookie clean start
            journal.getLastLogMark().readLog();
            lastLogMarkList.add(journal.getLastLogMark().markLog());
            assertEquals(0, lastLogMarkList.get(i).getCurMark().compare(new LogMark(0, 0)));
        }

        ClientConfiguration clientConf = new ClientConfiguration();
        clientConf.setMetadataServiceUri(metadataServiceUri);
        BookKeeper bkClient = new BookKeeper(clientConf);

        // Create multiple ledgers for adding entries to multiple journals
        for (int i = 0; i < journalDirs.length; i++) {
            LedgerHandle lh = bkClient.createLedger(1, 1, 1, DigestType.CRC32, "passwd".getBytes());
            long entryId = -1;
            // Ensure that we have non-zero number of entries
            long numOfEntries = new Random().nextInt(10) + 3;
            for (int j = 0; j < numOfEntries; j++) {
                entryId = lh.addEntry("data".getBytes());
            }
            assertEquals(entryId, (numOfEntries - 1));
            lh.close();
        }

        for (int i = 0; i < journalDirs.length; i++) {
            Journal journal = ((BookieImpl) server.getBookie()).journals.get(i);
            // In-memory LastLogMark should be updated with every write to journal
            assertTrue(journal.getLastLogMark().getCurMark().compare(lastLogMarkList.get(i).getCurMark()) > 0);
            lastLogMarkList.set(i, journal.getLastLogMark().markLog());
        }

        // Kill Bookie abruptly before entries are flushed to disk
        server.shutdown();

        conf.setDiskUsageThreshold(0.001f)
                .setDiskUsageWarnThreshold(0.0f).setReadOnlyModeEnabled(true).setIsForceGCAllowWhenNoSpace(true)
                .setMinUsableSizeForIndexFileCreation(5 * 1024);

        server = new BookieServer(conf);

        for (int i = 0; i < journalDirs.length; i++) {
            Journal journal = ((BookieImpl) server.getBookie()).journals.get(i);
            // LastLogMark should be (0, 0) before bookie restart since bookie crashed before persisting lastMark
            assertEquals(0, journal.getLastLogMark().getCurMark().compare(new LogMark(0, 0)));
        }

        int numOfRestarts = 3;
        // Restart server multiple times to ensure that logs are never replayed and new files are not generated
        for (int i = 0; i < numOfRestarts; i++) {

            int txnBefore = countNumOfFiles(conf.getJournalDirs(), "txn");
            int logBefore = countNumOfFiles(conf.getLedgerDirs(), "log");
            int idxBefore = countNumOfFiles(conf.getLedgerDirs(), "idx");

            server.start();

            for (int j = 0; j < journalDirs.length; j++) {
                Journal journal = ((BookieImpl) server.getBookie()).journals.get(j);
                assertTrue(journal.getLastLogMark().getCurMark().compare(lastLogMarkList.get(j).getCurMark()) > 0);
                lastLogMarkList.set(j, journal.getLastLogMark().markLog());
            }

            server.shutdown();

            // Every bookie restart initiates a new journal file
            // Journals should not be replayed everytime since lastMark gets updated everytime
            // New EntryLog files should not be generated.
            assertEquals(journalDirs.length, (countNumOfFiles(conf.getJournalDirs(), "txn") - txnBefore));

            // First restart should replay journal and generate new log/index files
            // Subsequent runs should not generate new files (but can delete older ones)
            if (i == 0) {
                assertTrue((countNumOfFiles(conf.getLedgerDirs(), "log") - logBefore) > 0);
                assertTrue((countNumOfFiles(conf.getLedgerDirs(), "idx") - idxBefore) > 0);
            } else {
                assertTrue((countNumOfFiles(conf.getLedgerDirs(), "log") - logBefore) <= 0);
                assertTrue((countNumOfFiles(conf.getLedgerDirs(), "idx") - idxBefore) <= 0);
            }

            server = new BookieServer(conf);
        }
        bkClient.close();
    }

    /**
     * Verify the bookie server exit code. On ZooKeeper exception, should return
     * exit code ZK_REG_FAIL = 4
     */
    @Test
    public void testExitCodeZK_REG_FAIL() throws Exception {
        File tmpDir = createTempDir("bookie", "test");

        final ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
        conf.setJournalDirName(tmpDir.getPath())
            .setLedgerDirNames(new String[] { tmpDir.getPath() })
            .setMetadataServiceUri(metadataServiceUri);

        RegistrationManager rm = mock(RegistrationManager.class);
        doThrow(new MetadataStoreException("mocked exception"))
            .when(rm)
            .registerBookie(any(BookieId.class), anyBoolean(), any(BookieServiceInfo.class));

        // simulating ZooKeeper exception by assigning a closed zk client to bk
        BookieServer bkServer = new BookieServer(conf) {
            @Override
            protected Bookie newBookie(ServerConfiguration conf, ByteBufAllocator allocator,
                     Supplier<BookieServiceInfo> bookieServiceInfoProvider)
                    throws IOException, KeeperException, InterruptedException,
                    BookieException {
                Bookie bookie = new BookieImpl(conf);
                MetadataBookieDriver driver = Whitebox.getInternalState(bookie, "metadataDriver");
                ((ZKMetadataBookieDriver) driver).setRegManager(rm);
                return bookie;
            }
        };

        bkServer.start();
        bkServer.join();
        assertEquals("Failed to return ExitCode.ZK_REG_FAIL",
                ExitCode.ZK_REG_FAIL, bkServer.getExitCode());
    }

    @Test
    public void testBookieRegistrationWithSameZooKeeperClient() throws Exception {
        final ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
        conf.setMetadataServiceUri(metadataServiceUri)
            .setListeningInterface(null);

        BookieId bookieId = BookieImpl.getBookieId(conf);

        driver.initialize(conf, () -> {}, NullStatsLogger.INSTANCE);
        try (StateManager manager = new BookieStateManager(conf, driver)) {
            manager.registerBookie(true).get();
            assertTrue(
                "Bookie registration node doesn't exists!",
                driver.getRegistrationManager().isBookieRegistered(bookieId));

            // test register bookie again if the registeration node is created by itself.
            manager.registerBookie(true).get();
            assertTrue(
                "Bookie registration node doesn't exists!",
                driver.getRegistrationManager().isBookieRegistered(bookieId));
        }
    }

    /**
     * Verify the bookie reg. Restarting bookie server will wait for the session
     * timeout when previous reg node exists in zk. On zNode delete event,
     * should continue startup
     */
    @Test
    public void testBookieRegistration() throws Exception {
        final ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
        conf.setMetadataServiceUri(metadataServiceUri)
            .setListeningInterface(null);

        String bookieId = BookieImpl.getBookieAddress(conf).toString();
        final String bkRegPath = ZKMetadataDriverBase.resolveZkLedgersRootPath(conf)
            + "/" + AVAILABLE_NODE + "/" + bookieId;

        driver.initialize(conf, () -> {}, NullStatsLogger.INSTANCE);
        try (StateManager manager = new BookieStateManager(conf, driver)) {
            manager.registerBookie(true).get();
        }
        Stat bkRegNode1 = zkc.exists(bkRegPath, false);
        assertNotNull("Bookie registration has been failed", bkRegNode1);

        // simulating bookie restart, on restart bookie will create new
        // zkclient and doing the registration.
        try (MetadataBookieDriver newDriver = new ZKMetadataBookieDriver()) {
            newDriver.initialize(conf, () -> {}, NullStatsLogger.INSTANCE);

            try (ZooKeeperClient newZk = createNewZKClient()) {
                // deleting the znode, so that the bookie registration should
                // continue successfully on NodeDeleted event
                new Thread(() -> {
                    try {
                        Thread.sleep(conf.getZkTimeout() / 3);
                        zkc.delete(bkRegPath, -1);
                    } catch (Exception e) {
                        // Not handling, since the testRegisterBookie will fail
                        LOG.error("Failed to delete the znode :" + bkRegPath, e);
                    }
                }).start();
                try (StateManager newMgr = new BookieStateManager(conf, newDriver)) {
                    newMgr.registerBookie(true).get();
                } catch (IOException e) {
                    Throwable t = e.getCause();
                    if (t instanceof KeeperException) {
                        KeeperException ke = (KeeperException) t;
                        assertTrue("ErrorCode:" + ke.code()
                                + ", Registration node exists",
                            ke.code() != KeeperException.Code.NODEEXISTS);
                    }
                    throw e;
                }

                // verify ephemeral owner of the bkReg znode
                Stat bkRegNode2 = newZk.exists(bkRegPath, false);
                assertNotNull("Bookie registration has been failed", bkRegNode2);
                assertTrue("Bookie is referring to old registration znode:"
                    + bkRegNode1 + ", New ZNode:" + bkRegNode2, bkRegNode1
                    .getEphemeralOwner() != bkRegNode2.getEphemeralOwner());
            }
        }
    }

    @Test(timeout = 20000)
    public void testBookieRegistrationWithFQDNHostNameAsBookieID() throws Exception {
        final ServerConfiguration conf = TestBKConfiguration.newServerConfiguration()
            .setMetadataServiceUri(metadataServiceUri)
            .setUseHostNameAsBookieID(true)
            .setListeningInterface(null);

        final BookieId bookieId =
                BookieId.parse(InetAddress.getLocalHost().getCanonicalHostName() + ":" + conf.getBookiePort());

        driver.initialize(conf, () -> {}, NullStatsLogger.INSTANCE);
        try (StateManager manager = new BookieStateManager(conf, driver)) {
            manager.registerBookie(true).get();
            assertTrue("Bookie registration node doesn't exists!",
                driver.getRegistrationManager().isBookieRegistered(bookieId));
        }
    }

    @Test(timeout = 20000)
    public void testBookieRegistrationWithShortHostNameAsBookieID() throws Exception {
        final ServerConfiguration conf = TestBKConfiguration.newServerConfiguration()
            .setMetadataServiceUri(metadataServiceUri)
            .setUseHostNameAsBookieID(true)
            .setUseShortHostName(true)
            .setListeningInterface(null);

        final BookieId bookieId = BookieId.parse(InetAddress.getLocalHost().getCanonicalHostName().split("\\.", 2)[0]
            + ":" + conf.getBookiePort());

        driver.initialize(conf, () -> {}, NullStatsLogger.INSTANCE);
        try (StateManager manager = new BookieStateManager(conf, driver)) {
            manager.registerBookie(true).get();
            assertTrue("Bookie registration node doesn't exists!",
                driver.getRegistrationManager().isBookieRegistered(bookieId));
        }
    }

    /**
     * Verify the bookie registration, it should throw
     * KeeperException.NodeExistsException if the znode still exists even after
     * the zk session timeout.
     */
    @Test
    public void testRegNodeExistsAfterSessionTimeOut() throws Exception {
        final ServerConfiguration conf = TestBKConfiguration.newServerConfiguration()
            .setMetadataServiceUri(metadataServiceUri)
            .setListeningInterface(null);

        BookieId bookieId = BookieId.parse(InetAddress.getLocalHost().getHostAddress() + ":"
                + conf.getBookiePort());
        String bkRegPath = ZKMetadataDriverBase.resolveZkLedgersRootPath(conf) + "/" + AVAILABLE_NODE + "/" + bookieId;

        driver.initialize(conf, () -> {}, NullStatsLogger.INSTANCE);
        try (StateManager manager = new BookieStateManager(conf, driver)) {
            manager.registerBookie(true).get();
            assertTrue("Bookie registration node doesn't exists!",
                driver.getRegistrationManager().isBookieRegistered(bookieId));
        }
        Stat bkRegNode1 = zkc.exists(bkRegPath, false);
        assertNotNull("Bookie registration has been failed",
            bkRegNode1);

        // simulating bookie restart, on restart bookie will create new
        // zkclient and doing the registration.
        try (MetadataBookieDriver newDriver = new ZKMetadataBookieDriver()) {
            newDriver.initialize(conf, () -> {}, NullStatsLogger.INSTANCE);
            try (StateManager newMgr = new BookieStateManager(conf, newDriver)) {
                newMgr.registerBookie(true).get();
                fail("Should throw NodeExistsException as the znode is not getting expired");
            } catch (ExecutionException ee) {
                Throwable e = ee.getCause(); // IOException
                Throwable t1 = e.getCause(); // BookieException.MetadataStoreException
                Throwable t2 = t1.getCause(); // IOException
                Throwable t3 = t2.getCause(); // KeeperException.NodeExistsException

                if (t3 instanceof KeeperException) {
                    KeeperException ke = (KeeperException) t3;
                    assertTrue("ErrorCode:" + ke.code()
                            + ", Registration node doesn't exists",
                        ke.code() == KeeperException.Code.NODEEXISTS);

                    // verify ephemeral owner of the bkReg znode
                    Stat bkRegNode2 = zkc.exists(bkRegPath, false);
                    assertNotNull("Bookie registration has been failed",
                        bkRegNode2);
                    assertTrue(
                        "Bookie wrongly registered. Old registration znode:"
                            + bkRegNode1 + ", New znode:" + bkRegNode2,
                        bkRegNode1.getEphemeralOwner() == bkRegNode2
                            .getEphemeralOwner());
                    return;
                }
                throw ee;
            }
        }
    }

    @Test(timeout = 20000)
    public void testBookieRegistrationBookieServiceInfo() throws Exception {
        final ServerConfiguration conf = TestBKConfiguration.newServerConfiguration()
            .setMetadataServiceUri(metadataServiceUri)
            .setUseHostNameAsBookieID(true)
            .setUseShortHostName(true)
            .setListeningInterface(null);

        final BookieId bookieId = BookieId.parse(InetAddress.getLocalHost().getCanonicalHostName().split("\\.", 2)[0]
                + ":" + conf.getBookiePort());
        String bkRegPath = ZKMetadataDriverBase.resolveZkLedgersRootPath(conf) + "/" + AVAILABLE_NODE + "/" + bookieId;

        driver.initialize(conf, () -> {}, NullStatsLogger.INSTANCE);

        Endpoint endpoint = new Endpoint("test", 1281, "localhost", "bookie-rpc",
                Collections.emptyList(), Collections.emptyList());
        BookieServiceInfo bsi = new BookieServiceInfo(Collections.emptyMap(), Arrays.asList(endpoint));
        Supplier<BookieServiceInfo> supplier = () -> bsi;

        DiskChecker diskChecker = new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold());
        LedgerDirsManager ledgerDirsManager = new LedgerDirsManager(
                conf, conf.getLedgerDirs(), diskChecker);
        try (StateManager manager = new BookieStateManager(conf,
                NullStatsLogger.INSTANCE, driver, ledgerDirsManager, supplier)) {
            manager.registerBookie(true).get();
            assertTrue("Bookie registration node doesn't exists!",
                    driver.getRegistrationManager().isBookieRegistered(bookieId));
        }
        Stat bkRegNode = zkc.exists(bkRegPath, false);
        assertNotNull("Bookie registration has been failed", bkRegNode);

        byte[] bkRegNodeData = zkc.getData(bkRegPath, null, null);
        assertFalse("Bookie service info not written", bkRegNodeData == null || bkRegNodeData.length == 0);

        BookieServiceInfoFormat serializedBookieServiceInfo = BookieServiceInfoFormat.parseFrom(bkRegNodeData);
        BookieServiceInfoFormat.Endpoint serializedEndpoint = serializedBookieServiceInfo.getEndpoints(0);
        assertNotNull("Serialized Bookie endpoint not found", serializedEndpoint);

        assertEquals(endpoint.getId(), serializedEndpoint.getId());
        assertEquals(endpoint.getHost(), serializedEndpoint.getHost());
        assertEquals(endpoint.getPort(), serializedEndpoint.getPort());
    }

    /**
     * Verify user cannot start if user is in permittedStartupUsers conf list BKException BKUnauthorizedAccessException
     * if cannot start.
     */
    @Test
    public void testUserNotPermittedToStart() throws Exception {
        File tmpDir = createTempDir("bookie", "test");

        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
        int port = PortManager.nextFreePort();
        conf.setMetadataServiceUri(null)
            .setBookiePort(port)
            .setJournalDirName(tmpDir.getPath())
            .setLedgerDirNames(new String[] { tmpDir.getPath() });
        String userString = "larry, curly,moe,,";
        conf.setPermittedStartupUsers(userString);
        BookieServer bs1 = null;

        boolean sawException = false;
        try {
            bs1 = new BookieServer(conf);
            Assert.fail("Bookkeeper should not have started since current user isn't in permittedStartupUsers");
        } catch (AccessControlException buae) {
            sawException = true;
        } finally {
            if (bs1 != null && bs1.isRunning()) {
                bs1.shutdown();
            }
        }
        assertTrue("Should have thrown exception", sawException);
    }

    /**
     * Verify user cannot start if user is in permittedStartupUsers conf list BKException BKUnauthorizedAccessException
     * if cannot start.
     */
    @Test
    public void testUserPermittedToStart() throws Exception {
        File tmpDir = createTempDir("bookie", "test");

        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
        int port = PortManager.nextFreePort();
        conf.setMetadataServiceUri(null)
            .setBookiePort(port)
            .setJournalDirName(tmpDir.getPath())
            .setLedgerDirNames(new String[] { tmpDir.getPath() });

        BookieServer bs1 = null;

        // Multiple commas
        String userString = "larry,,,curly ," + System.getProperty("user.name") + " ,moe";
        conf.setPermittedStartupUsers(userString);
        try {
            bs1 = new BookieServer(conf);
            bs1.start();
        } catch (AccessControlException buae) {
            Assert.fail("Bookkeeper should have started since current user is in permittedStartupUsers");
        } finally {
            if (bs1 != null && bs1.isRunning()) {
                bs1.shutdown();
            }
        }

        // Comma at end
        userString = "larry ,curly, moe," + System.getProperty("user.name") + ",";
        conf.setPermittedStartupUsers(userString);
        try {
            bs1 = new BookieServer(conf);
            bs1.start();
        } catch (AccessControlException buae) {
            Assert.fail("Bookkeeper should have started since current user is in permittedStartupUsers");
        } finally {
            if (bs1 != null && bs1.isRunning()) {
                bs1.shutdown();
            }
        }
    }

    /**
     * Verify user can start if user is not in permittedStartupUsers but it is empty BKException
     * BKUnauthorizedAccessException if cannot start.
     */
    @Test
    public void testUserPermittedToStartWithMissingProperty() throws Exception {
        File tmpDir = createTempDir("bookie", "test");

        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
        LOG.info("{}", conf);

        int port = PortManager.nextFreePort();
        conf.setMetadataServiceUri(null)
            .setBookiePort(port)
            .setJournalDirName(tmpDir.getPath())
            .setLedgerDirNames(new String[] { tmpDir.getPath() });
        BookieServer bs1 = null;
        try {
            bs1 = new BookieServer(conf);
            bs1.start();
        } catch (AccessControlException buae) {
            Assert.fail("Bookkeeper should have started since permittedStartupUser is not specified");
        } finally {
            if (bs1 != null && bs1.isRunning()) {
                bs1.shutdown();
            }
        }
    }

    /**
     * Verify duplicate bookie server startup. Should throw
     * java.net.BindException if already BK server is running
     */
    @Test
    public void testDuplicateBookieServerStartup() throws Exception {
        File tmpDir = createTempDir("bookie", "test");

        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
        int port = PortManager.nextFreePort();
        conf.setBookiePort(port)
            .setJournalDirName(tmpDir.getPath())
            .setLedgerDirNames(new String[] { tmpDir.getPath() })
            .setMetadataServiceUri(metadataServiceUri);
        BookieServer bs1 = new BookieServer(conf);
        bs1.start();
        BookieServer bs2 = null;
        // starting bk server with same conf
        try {
            bs2 = new BookieServer(conf);
            bs2.start();
            fail("Should throw BindException, as the bk server is already running!");
        } catch (BindException e) {
            // Ok
        } catch (IOException e) {
            assertTrue("BKServer allowed duplicate Startups!",
                    e.getMessage().contains("bind"));
        } finally {
            bs1.shutdown();
            if (bs2 != null) {
                bs2.shutdown();
            }
        }
    }

    @Test
    public void testBookieServiceExceptionHandler() throws Exception {
        File tmpDir = createTempDir("bookie", "exception-handler");
        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
        int port = PortManager.nextFreePort();
        conf.setBookiePort(port)
            .setJournalDirName(tmpDir.getPath())
            .setLedgerDirNames(new String[] { tmpDir.getPath() })
            .setMetadataServiceUri(metadataServiceUri);

        BookieConfiguration bkConf = new BookieConfiguration(conf);
        BookieService service = new BookieService(bkConf, NullStatsLogger.INSTANCE, BookieServiceInfo.NO_INFO);
        CompletableFuture<Void> startFuture = ComponentStarter.startComponent(service);

        // shutdown the bookie service
        service.getServer().getBookie().shutdown();

        // the bookie service lifecycle component should be shutdown.
        startFuture.get();
    }

    /**
     * Mock InterleavedLedgerStorage class where addEntry is mocked to throw
     * OutOfMemoryError.
     */
    public static class MockInterleavedLedgerStorage extends InterleavedLedgerStorage {
        AtomicInteger atmoicInt = new AtomicInteger(0);

        @Override
        public long addEntry(ByteBuf entry) throws IOException {
            if (atmoicInt.incrementAndGet() == 10) {
                throw new OutOfMemoryError("Some Injected Exception");
            }
            return super.addEntry(entry);
        }
    }

    @Test
    public void testBookieStartException() throws Exception {
        File journalDir = createTempDir("bookie", "journal");
        BookieImpl.checkDirectoryStructure(BookieImpl.getCurrentDirectory(journalDir));

        File ledgerDir = createTempDir("bookie", "ledger");
        BookieImpl.checkDirectoryStructure(BookieImpl.getCurrentDirectory(ledgerDir));

        /*
         * add few entries to journal file.
         */
        int numOfEntries = 100;
        writeV5Journal(BookieImpl.getCurrentDirectory(journalDir), numOfEntries,
                "testV5Journal".getBytes());

        /*
         * This Bookie is configured to use MockInterleavedLedgerStorage.
         * MockInterleavedLedgerStorage throws an Error for addEntry request.
         * This is to simulate Bookie/BookieServer/BookieService 'start' failure
         * because of 'Bookie.readJournal' failure.
         */
        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
        int port = PortManager.nextFreePort();
        conf.setBookiePort(port).setJournalDirName(journalDir.getPath())
                .setLedgerDirNames(new String[] { ledgerDir.getPath() }).setMetadataServiceUri(metadataServiceUri)
                .setLedgerStorageClass(MockInterleavedLedgerStorage.class.getName());

        BookieConfiguration bkConf = new BookieConfiguration(conf);
        driver.initialize(conf, () -> {}, NullStatsLogger.INSTANCE);

        /*
         * create cookie and write it to JournalDir/LedgerDir.
         */
        Cookie.Builder cookieBuilder = Cookie.generateCookie(conf);
        Cookie cookie = cookieBuilder.build();
        cookie.writeToDirectory(new File(journalDir, "current"));
        cookie.writeToDirectory(new File(ledgerDir, "current"));
        Versioned<byte[]> newCookie = new Versioned<>(
                cookie.toString().getBytes(UTF_8), Version.NEW
        );
        driver.getRegistrationManager().writeCookie(BookieImpl.getBookieId(conf), newCookie);

        /*
         * Create LifecycleComponent for BookieServer and start it.
         */
        LifecycleComponent server = Main.buildBookieServer(bkConf);
        CompletableFuture<Void> startFuture = ComponentStarter.startComponent(server);

        /*
         * Since Bookie/BookieServer/BookieService is expected to fail, it would
         * cause bookie-server component's exceptionHandler to get triggered.
         * This exceptionHandler will make sure all of the components to get
         * closed and then finally completes the Future.
         */
        startFuture.get();

        /*
         * make sure that Component's exceptionHandler is called by checking if
         * the error message of ExceptionHandler is logged. This Log message is
         * defined in anonymous exceptionHandler class defined in
         * ComponentStarter.startComponent method.
         */
        loggerOutput.expect((List<LoggingEvent> logEvents) -> {
            assertThat(logEvents,
                    hasItem(hasProperty("message", containsString("Triggered exceptionHandler of Component:"))));
        });
    }

    /**
     * Test that if the journal reads an entry with negative length, it shuts down
     * the bookie normally. An admin should look to see what has
     * happened in this case.
     */
    @Test
    public void testNegativeLengthEntryBookieShutdown() throws Exception {
        File journalDir = createTempDir("bookie", "journal");
        BookieImpl.checkDirectoryStructure(BookieImpl.getCurrentDirectory(journalDir));

        File ledgerDir = createTempDir("bookie", "ledger");
        BookieImpl.checkDirectoryStructure(BookieImpl.getCurrentDirectory(ledgerDir));

        writeV5Journal(BookieImpl.getCurrentDirectory(journalDir), 5,
                "testV5Journal".getBytes(), true);

        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
        conf.setJournalDirName(journalDir.getPath())
                .setLedgerDirNames(new String[] { ledgerDir.getPath() })
                .setMetadataServiceUri(null);

        Bookie b = null;
        try {
            b = new BookieImpl(conf);
            b.start();
            assertFalse("Bookie should shutdown normally after catching IOException"
                    + " due to corrupt entry with negative length", b.isRunning());
        } finally {
            if (b != null) {
                b.shutdown();
            }
        }
    }

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

        BookieConfiguration bkConf = new BookieConfiguration(conf);
        AutoRecoveryService service = new AutoRecoveryService(bkConf, NullStatsLogger.INSTANCE);
        CompletableFuture<Void> startFuture = ComponentStarter.startComponent(service);

        // shutdown the AutoRecovery service
        service.getAutoRecoveryServer().shutdown();

        // the AutoRecovery lifecycle component should be shutdown.
        startFuture.get();
    }

    /**
     * Verify bookie server starts up on ephemeral ports.
     */
    @Test
    public void testBookieServerStartupOnEphemeralPorts() throws Exception {
        File tmpDir1 = createTempDir("bookie", "test1");
        File tmpDir2 = createTempDir("bookie", "test2");

        ServerConfiguration conf1 = TestBKConfiguration.newServerConfiguration();
        conf1.setBookiePort(0)
            .setJournalDirName(tmpDir1.getPath())
            .setLedgerDirNames(
                new String[] { tmpDir1.getPath() })
            .setMetadataServiceUri(null);
        assertEquals(0, conf1.getBookiePort());
        BookieServer bs1 = new BookieServer(conf1);
        bs1.start();
        assertFalse(0 == conf1.getBookiePort());

        // starting bk server with same conf
        ServerConfiguration conf2 = TestBKConfiguration.newServerConfiguration();
        conf2.setBookiePort(0)
            .setJournalDirName(tmpDir2.getPath())
            .setLedgerDirNames(
                new String[] { tmpDir2.getPath() })
            .setMetadataServiceUri(null);
        BookieServer bs2 = new BookieServer(conf2);
        bs2.start();
        assertFalse(0 == conf2.getBookiePort());

        // these two bookies are listening on different ports eventually
        assertFalse(conf1.getBookiePort() == conf2.getBookiePort());
    }

    /**
     * Verify bookie start behaviour when ZK Server is not running.
     */
    @Test
    public void testStartBookieWithoutZKServer() throws Exception {
        zkUtil.killCluster();

        File tmpDir = createTempDir("bookie", "test");

        final ServerConfiguration conf = TestBKConfiguration.newServerConfiguration()
                .setJournalDirName(tmpDir.getPath())
                .setLedgerDirNames(new String[] { tmpDir.getPath() });
        conf.setMetadataServiceUri(zkUtil.getMetadataServiceUri()).setZkTimeout(5000);

        try {
            new BookieImpl(conf);
            fail("Should throw ConnectionLossException as ZKServer is not running!");
        } catch (BookieException.MetadataStoreException e) {
            // expected behaviour
        }
    }

    /**
     * Verify that if I try to start a bookie without zk initialized, it won't
     * prevent me from starting the bookie when zk is initialized.
     */
    @Test
    public void testStartBookieWithoutZKInitialized() throws Exception {
        File tmpDir = createTempDir("bookie", "test");
        final String zkRoot = "/ledgers2";

        final ServerConfiguration conf = TestBKConfiguration.newServerConfiguration()
            .setJournalDirName(tmpDir.getPath())
            .setLedgerDirNames(new String[] { tmpDir.getPath() })
            .setMetadataServiceUri(zkUtil.getMetadataServiceUri(zkRoot))
            .setZkTimeout(5000);
        try {
            new BookieImpl(conf);
            fail("Should throw NoNodeException");
        } catch (Exception e) {
            // shouldn't be able to start
        }
        ServerConfiguration adminConf = new ServerConfiguration();
        adminConf.setMetadataServiceUri(zkUtil.getMetadataServiceUri(zkRoot));
        BookKeeperAdmin.format(adminConf, false, false);

        Bookie b = new BookieImpl(conf);
        b.shutdown();
    }

    /**
     * Check disk full. Expected to fail on start.
     */
    @Test
    public void testWithDiskFullReadOnlyDisabledOrForceGCAllowDisabled() throws Exception {
        File tmpDir = createTempDir("DiskCheck", "test");
        long usableSpace = tmpDir.getUsableSpace();
        long totalSpace = tmpDir.getTotalSpace();
        final ServerConfiguration conf = TestBKConfiguration.newServerConfiguration()
                .setLedgerStorageClass(InterleavedLedgerStorage.class.getName())
                .setJournalDirName(tmpDir.getPath())
                .setLedgerDirNames(new String[] { tmpDir.getPath() })
                .setDiskCheckInterval(1000)
                .setDiskUsageThreshold((1.0f - ((float) usableSpace / (float) totalSpace)) * 0.999f)
                .setDiskUsageWarnThreshold(0.0f)
                .setMetadataServiceUri(metadataServiceUri)
                .setZkTimeout(5000);

        // if isForceGCAllowWhenNoSpace or readOnlyModeEnabled is not set and Bookie is
        // started when Disk is full, then it will fail to start with NoWritableLedgerDirException

        conf.setMinUsableSizeForEntryLogCreation(Long.MAX_VALUE)
            .setReadOnlyModeEnabled(false);
        try {
            new BookieImpl(conf);
            fail("NoWritableLedgerDirException expected");
        } catch (NoWritableLedgerDirException e) {
            // expected
        }

        conf.setMinUsableSizeForEntryLogCreation(Long.MIN_VALUE)
            .setReadOnlyModeEnabled(false);
        try {
            new BookieImpl(conf);
            fail("NoWritableLedgerDirException expected");
        } catch (NoWritableLedgerDirException e) {
            // expected
        }

        conf.setMinUsableSizeForEntryLogCreation(Long.MAX_VALUE)
            .setReadOnlyModeEnabled(true);
        Bookie bookie = null;
        try {
            // bookie is okay to start up when readonly mode is enabled because entry log file creation
            // is deferred.
            bookie = new BookieImpl(conf);
        } catch (NoWritableLedgerDirException e) {
            fail("NoWritableLedgerDirException unexpected");
        } finally {
            if (null != bookie) {
                bookie.shutdown();
            }
        }
    }

    /**
     * Check disk full. Expected to start as read-only.
     */
    @Test
    public void testWithDiskFullReadOnlyEnabledAndForceGCAllowAllowed() throws Exception {
        File tmpDir = createTempDir("DiskCheck", "test");
        long usableSpace = tmpDir.getUsableSpace();
        long totalSpace = tmpDir.getTotalSpace();
        final ServerConfiguration conf = TestBKConfiguration.newServerConfiguration()
                .setJournalDirName(tmpDir.getPath())
                .setLedgerDirNames(new String[] { tmpDir.getPath() })
                .setDiskCheckInterval(1000)
                .setDiskUsageThreshold((1.0f - ((float) usableSpace / (float) totalSpace)) * 0.999f)
                .setDiskUsageWarnThreshold(0.0f)
                .setMetadataServiceUri(metadataServiceUri)
                .setZkTimeout(5000);

        // if isForceGCAllowWhenNoSpace and readOnlyModeEnabled are set, then Bookie should
        // start with readonlymode when Disk is full (assuming there is no need for creation of index file
        // while replaying the journal)
        conf.setReadOnlyModeEnabled(true)
            .setIsForceGCAllowWhenNoSpace(true);
        final Bookie bk = new BookieImpl(conf);
        bk.start();
        Thread.sleep((conf.getDiskCheckInterval() * 2) + 100);

        assertTrue(bk.isReadOnly());
        bk.shutdown();
    }

    class MockBookieServer extends BookieServer {
        ServerConfiguration conf;

        public MockBookieServer(ServerConfiguration conf) throws IOException, KeeperException, InterruptedException,
                BookieException, UnavailableException, CompatibilityException, SecurityException {
            super(conf);
            this.conf = conf;
        }

        @Override
        protected Bookie newBookie(ServerConfiguration conf, ByteBufAllocator allocator,
                     Supplier<BookieServiceInfo> bookieServiceInfoProvider)
                throws IOException, KeeperException, InterruptedException, BookieException {
            return new MockBookieWithNoopShutdown(conf, NullStatsLogger.INSTANCE);
        }
    }

    class MockBookieWithNoopShutdown extends BookieImpl {
        public MockBookieWithNoopShutdown(ServerConfiguration conf, StatsLogger statsLogger)
                throws IOException, KeeperException, InterruptedException, BookieException {
            super(conf, statsLogger, UnpooledByteBufAllocator.DEFAULT, BookieServiceInfo.NO_INFO);
        }

        // making Bookie Shutdown no-op. Ideally for this testcase we need to
        // kill bookie abruptly to simulate the scenario where bookie is killed
        // without execution of shutdownhook (and corresponding shutdown logic).
        // Since there is no easy way to simulate abrupt kill of Bookie we are
        // injecting noop Bookie Shutdown
        @Override
        synchronized int shutdown(int exitCode) {
            return exitCode;
        }
    }

    @Test
    public void testWithDiskFullAndAbilityToCreateNewIndexFile() throws Exception {
        File tmpDir = createTempDir("DiskCheck", "test");

        final ServerConfiguration conf = newServerConfiguration()
            .setJournalDirName(tmpDir.getPath())
            .setLedgerDirNames(new String[] { tmpDir.getPath() })
            .setDiskCheckInterval(1000)
            .setLedgerStorageClass(SortedLedgerStorage.class.getName())
            .setAutoRecoveryDaemonEnabled(false)
            .setZkTimeout(5000);

        BookieServer server = new MockBookieServer(conf);
        server.start();
        ClientConfiguration clientConf = new ClientConfiguration();
        clientConf.setMetadataServiceUri(metadataServiceUri);
        BookKeeper bkClient = new BookKeeper(clientConf);
        LedgerHandle lh = bkClient.createLedger(1, 1, 1, DigestType.CRC32, "passwd".getBytes());
        long entryId = -1;
        long numOfEntries = 5;
        for (int i = 0; i < numOfEntries; i++) {
            entryId = lh.addEntry("data".getBytes());
        }
        assertTrue("EntryId of the recently added entry should be 0", entryId == (numOfEntries - 1));
        // We want to simulate the scenario where Bookie is killed abruptly, so
        // SortedLedgerStorage's EntryMemTable and IndexInMemoryPageManager are
        // not flushed and hence when bookie is restarted it will replay the
        // journal. Since there is no easy way to kill the Bookie abruptly, we
        // are injecting no-op shutdown.
        server.shutdown();

        conf.setDiskUsageThreshold(0.001f)
                .setDiskUsageWarnThreshold(0.0f).setReadOnlyModeEnabled(true).setIsForceGCAllowWhenNoSpace(true)
                .setMinUsableSizeForIndexFileCreation(Long.MAX_VALUE);
        server = new BookieServer(conf);
        // Now we are trying to start the Bookie, which tries to replay the
        // Journal. While replaying the Journal it tries to create the IndexFile
        // for the ledger (whose entries are not flushed). but since we set
        // minUsableSizeForIndexFileCreation to very high value, it wouldn't. be
        // able to find any index dir when all discs are full
        server.start();
        assertFalse("Bookie should be Shutdown", server.getBookie().isRunning());
        server.shutdown();

        // Here we are setting MinUsableSizeForIndexFileCreation to very low
        // value. So if index dirs are full then it will consider the dirs which
        // have atleast MinUsableSizeForIndexFileCreation usable space for the
        // creation of new Index file.
        conf.setMinUsableSizeForIndexFileCreation(5 * 1024);
        server = new BookieServer(conf);
        server.start();
        Thread.sleep((conf.getDiskCheckInterval() * 2) + 100);
        assertTrue("Bookie should be up and running", server.getBookie().isRunning());
        assertTrue(server.getBookie().isReadOnly());
        server.shutdown();
        bkClient.close();
    }

    /**
     * Check disk error for file. Expected to throw DiskErrorException.
     */
    @Test
    public void testWithDiskError() throws Exception {
        File parent = createTempDir("DiskCheck", "test");
        File child = File.createTempFile("DiskCheck", "test", parent);
        final ServerConfiguration conf = TestBKConfiguration.newServerConfiguration()
                .setJournalDirName(child.getPath())
                .setLedgerDirNames(new String[] { child.getPath() });
        conf.setMetadataServiceUri(metadataServiceUri)
            .setZkTimeout(5000);
        try {
            // LedgerDirsManager#init() is used in Bookie instantiation.
            // Simulating disk errors by directly calling #init
            LedgerDirsManager ldm = new LedgerDirsManager(conf, conf.getLedgerDirs(),
                    new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold()));
            LedgerDirsMonitor ledgerMonitor = new LedgerDirsMonitor(conf,
                    new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold()),
                    Collections.singletonList(ldm));
            ledgerMonitor.init();
            fail("should throw exception");
        } catch (Exception e) {
            // expected
        }
    }

    /**
     * if ALLOW_MULTIPLEDIRS_UNDER_SAME_DISKPARTITION is disabled then Bookie initialization
     * will fail if there are multiple ledger/index/journal dirs are in same partition/filesystem.
     */
    @Test
    public void testAllowDiskPartitionDuplicationDisabled() throws Exception {
        File tmpDir1 = createTempDir("bookie", "test");
        File tmpDir2 = createTempDir("bookie", "test");

        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
        int port = PortManager.nextFreePort();
        // multiple ledgerdirs in same diskpartition
        conf.setMetadataServiceUri(metadataServiceUri)
            .setZkTimeout(5000)
            .setBookiePort(port)
            .setJournalDirName(tmpDir1.getPath())
            .setLedgerDirNames(new String[] { tmpDir1.getPath(), tmpDir2.getPath() })
            .setIndexDirName(new String[] { tmpDir1.getPath() })
            .setAllowMultipleDirsUnderSameDiskPartition(false);
        BookieServer bs1 = null;
        try {
            bs1 = new BookieServer(conf);
            fail("Bookkeeper should not have started since AllowMultipleDirsUnderSameDiskPartition is not enabled");
        } catch (DiskPartitionDuplicationException dpde) {
            // Expected
        } finally {
            if (bs1 != null) {
                bs1.shutdown();
            }
        }

        tmpDir1 = createTempDir("bookie", "test");
        tmpDir2 = createTempDir("bookie", "test");
        port = PortManager.nextFreePort();
        // multiple indexdirs in same diskpartition
        conf.setMetadataServiceUri(metadataServiceUri)
            .setZkTimeout(5000)
            .setBookiePort(port)
            .setJournalDirName(tmpDir1.getPath())
            .setLedgerDirNames(new String[] { tmpDir1.getPath() })
            .setIndexDirName(new String[] { tmpDir1.getPath(), tmpDir2.getPath() })
            .setAllowMultipleDirsUnderSameDiskPartition(false);
        bs1 = null;
        try {
            bs1 = new BookieServer(conf);
            fail("Bookkeeper should not have started since AllowMultipleDirsUnderSameDiskPartition is not enabled");
        } catch (DiskPartitionDuplicationException dpde) {
            // Expected
        } finally {
            if (bs1 != null) {
                bs1.shutdown();
            }
        }

        tmpDir1 = createTempDir("bookie", "test");
        tmpDir2 = createTempDir("bookie", "test");
        port = PortManager.nextFreePort();
        // multiple journaldirs in same diskpartition
        conf.setMetadataServiceUri(metadataServiceUri)
            .setZkTimeout(5000)
            .setBookiePort(port)
            .setJournalDirsName(new String[] { tmpDir1.getPath(), tmpDir2.getPath() })
            .setLedgerDirNames(new String[] { tmpDir1.getPath() })
            .setIndexDirName(new String[] { tmpDir1.getPath()})
            .setAllowMultipleDirsUnderSameDiskPartition(false);
        bs1 = null;
        try {
            bs1 = new BookieServer(conf);
            fail("Bookkeeper should not have started since AllowMultipleDirsUnderSameDiskPartition is not enabled");
        } catch (DiskPartitionDuplicationException dpde) {
            // Expected
        } finally {
            if (bs1 != null) {
                bs1.shutdown();
            }
        }
    }

    /**
     * if ALLOW_MULTIPLEDIRS_UNDER_SAME_DISKPARTITION is enabled then Bookie initialization
     * should succeed even if there are multiple ledger/index/journal dirs in the same diskpartition/filesystem.
     */
    @Test
    public void testAllowDiskPartitionDuplicationAllowed() throws Exception {
        File tmpDir1 = createTempDir("bookie", "test");
        File tmpDir2 = createTempDir("bookie", "test");
        File tmpDir3 = createTempDir("bookie", "test");
        File tmpDir4 = createTempDir("bookie", "test");
        File tmpDir5 = createTempDir("bookie", "test");
        File tmpDir6 = createTempDir("bookie", "test");

        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
        int port = 12555;
        conf.setMetadataServiceUri(metadataServiceUri)
            .setZkTimeout(5000)
            .setBookiePort(port)
            .setJournalDirsName(new String[] { tmpDir1.getPath(), tmpDir2.getPath() })
            .setLedgerDirNames(new String[] { tmpDir3.getPath(), tmpDir4.getPath() })
            .setIndexDirName(new String[] { tmpDir5.getPath(), tmpDir6.getPath() });
        conf.setAllowMultipleDirsUnderSameDiskPartition(true);
        BookieServer bs1 = null;
        try {
            bs1 = new BookieServer(conf);
        } catch (DiskPartitionDuplicationException dpde) {
            fail("Bookkeeper should have started since AllowMultipleDirsUnderSameDiskPartition is enabled");
        } finally {
            if (bs1 != null) {
                bs1.shutdown();
            }
        }
    }

    private ZooKeeperClient createNewZKClient() throws Exception {
        // create a zookeeper client
        LOG.debug("Instantiate ZK Client");
        return ZooKeeperClient.newBuilder()
                .connectString(zkUtil.getZooKeeperConnectString())
                .build();
    }

    /**
     * Check bookie status should be able to persist on disk and retrieve when restart the bookie.
     */
    @Test(timeout = 10000)
    public void testPersistBookieStatus() throws Exception {
        // enable persistent bookie status
        File tmpDir = createTempDir("bookie", "test");
        final ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
        conf.setJournalDirName(tmpDir.getPath())
            .setLedgerDirNames(new String[] { tmpDir.getPath() })
            .setReadOnlyModeEnabled(true)
            .setPersistBookieStatusEnabled(true)
            .setMetadataServiceUri(metadataServiceUri);
        BookieServer bookieServer = new BookieServer(conf);
        bookieServer.start();
        Bookie bookie = bookieServer.getBookie();
        assertFalse(bookie.isReadOnly());
        // transition to readonly mode, bookie status should be persisted in ledger disks
        bookie.getStateManager().transitionToReadOnlyMode().get();
        assertTrue(bookie.isReadOnly());

        // restart bookie should start in read only mode
        bookieServer.shutdown();
        bookieServer = new BookieServer(conf);
        bookieServer.start();
        bookie = bookieServer.getBookie();
        assertTrue(bookie.isReadOnly());
        // transition to writable mode
        bookie.getStateManager().transitionToWritableMode().get();
        // restart bookie should start in writable mode
        bookieServer.shutdown();
        bookieServer = new BookieServer(conf);
        bookieServer.start();
        bookie = bookieServer.getBookie();
        assertFalse(bookie.isReadOnly());
        bookieServer.shutdown();
    }

    /**
     * Check when we start a ReadOnlyBookie, we should ignore bookie status.
     */
    @Test(timeout = 10000)
    public void testReadOnlyBookieShouldIgnoreBookieStatus() throws Exception {
        File tmpDir = createTempDir("bookie", "test");
        final ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
        conf.setJournalDirName(tmpDir.getPath())
            .setLedgerDirNames(new String[] { tmpDir.getPath() })
            .setReadOnlyModeEnabled(true)
            .setPersistBookieStatusEnabled(true)
            .setMetadataServiceUri(metadataServiceUri);
        // start new bookie
        BookieServer bookieServer = new BookieServer(conf);
        bookieServer.start();
        Bookie bookie = bookieServer.getBookie();
        // persist bookie status
        bookie.getStateManager().transitionToReadOnlyMode().get();
        bookie.getStateManager().transitionToWritableMode().get();
        assertFalse(bookie.isReadOnly());
        bookieServer.shutdown();
        // start read only bookie
        final ServerConfiguration readOnlyConf = TestBKConfiguration.newServerConfiguration();
        readOnlyConf.loadConf(conf);
        readOnlyConf.setForceReadOnlyBookie(true);
        bookieServer = new BookieServer(readOnlyConf);
        bookieServer.start();
        bookie = bookieServer.getBookie();
        assertTrue(bookie.isReadOnly());
        // transition to writable should fail
        bookie.getStateManager().transitionToWritableMode().get();
        assertTrue(bookie.isReadOnly());
        bookieServer.shutdown();
    }

    /**
     * Check that if there's multiple bookie status copies, as long as not all of them are corrupted,
     * the bookie status should be retrievable.
     */
    @Test(timeout = 10000)
    public void testRetrieveBookieStatusWhenStatusFileIsCorrupted() throws Exception {
        File[] tmpLedgerDirs = new File[3];
        String[] filePath = new String[tmpLedgerDirs.length];
        for (int i = 0; i < tmpLedgerDirs.length; i++) {
            tmpLedgerDirs[i] = createTempDir("bookie", "test" + i);
            filePath[i] = tmpLedgerDirs[i].getPath();
        }
        final ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
        conf.setJournalDirName(filePath[0])
            .setLedgerDirNames(filePath)
            .setReadOnlyModeEnabled(true)
            .setPersistBookieStatusEnabled(true)
            .setMetadataServiceUri(metadataServiceUri);
        // start a new bookie
        BookieServer bookieServer = new BookieServer(conf);
        bookieServer.start();
        // transition in to read only and persist the status on disk
        Bookie bookie = (BookieImpl) bookieServer.getBookie();
        assertFalse(bookie.isReadOnly());
        bookie.getStateManager().transitionToReadOnlyMode().get();
        assertTrue(bookie.isReadOnly());
        // corrupt status file
        List<File> ledgerDirs = ((BookieImpl) bookie).getLedgerDirsManager().getAllLedgerDirs();
        corruptFile(new File(ledgerDirs.get(0), BOOKIE_STATUS_FILENAME));
        corruptFile(new File(ledgerDirs.get(1), BOOKIE_STATUS_FILENAME));
        // restart the bookie should be in read only mode
        bookieServer.shutdown();
        bookieServer = new BookieServer(conf);
        bookieServer.start();
        bookie = bookieServer.getBookie();
        assertTrue(bookie.isReadOnly());
        bookieServer.shutdown();
    }

    /**
     * Check if the bookie would read the latest status if the status files are not consistent.
     * @throws Exception
     */
    @Test(timeout = 10000)
    public void testReadLatestBookieStatus() throws Exception {
        File[] tmpLedgerDirs = new File[3];
        String[] filePath = new String[tmpLedgerDirs.length];
        for (int i = 0; i < tmpLedgerDirs.length; i++) {
            tmpLedgerDirs[i] = createTempDir("bookie", "test" + i);
            filePath[i] = tmpLedgerDirs[i].getPath();
        }
        final ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
        conf.setJournalDirName(filePath[0])
            .setLedgerDirNames(filePath)
            .setReadOnlyModeEnabled(true)
            .setPersistBookieStatusEnabled(true)
            .setMetadataServiceUri(metadataServiceUri);
        // start a new bookie
        BookieServer bookieServer = new BookieServer(conf);
        bookieServer.start();
        // transition in to read only and persist the status on disk
        Bookie bookie = (BookieImpl) bookieServer.getBookie();
        assertFalse(bookie.isReadOnly());
        bookie.getStateManager().transitionToReadOnlyMode().get();
        assertTrue(bookie.isReadOnly());
        // Manually update a status file, so it becomes the latest
        Thread.sleep(1);
        BookieStatus status = new BookieStatus();
        List<File> dirs = new ArrayList<File>();
        dirs.add(((BookieImpl) bookie).getLedgerDirsManager().getAllLedgerDirs().get(0));
        status.writeToDirectories(dirs);
        // restart the bookie should start in writable state
        bookieServer.shutdown();
        bookieServer = new BookieServer(conf);
        bookieServer.start();
        bookie = bookieServer.getBookie();
        assertFalse(bookie.isReadOnly());
        bookieServer.shutdown();
    }

    private void corruptFile(File file) throws IOException {
        FileOutputStream fos = new FileOutputStream(file);
        BufferedWriter bw = null;
        try {
            bw = new BufferedWriter(new OutputStreamWriter(fos, UTF_8));
            byte[] bytes = new byte[64];
            new Random().nextBytes(bytes);
            bw.write(new String(bytes));
        } finally {
            if (bw != null) {
                bw.close();
            }
            fos.close();
        }
    }

    @Test
    public void testIOVertexHTTPServerEndpointForBookieWithPrometheusProvider() throws Exception {
        File tmpDir = createTempDir("bookie", "test");

        final ServerConfiguration conf = TestBKConfiguration.newServerConfiguration()
                .setJournalDirName(tmpDir.getPath()).setLedgerDirNames(new String[] { tmpDir.getPath() })
                .setBookiePort(PortManager.nextFreePort()).setMetadataServiceUri(metadataServiceUri)
                .setListeningInterface(null);

        /*
         * enable io.vertx http server
         */
        int nextFreePort = PortManager.nextFreePort();
        conf.setStatsProviderClass(PrometheusMetricsProvider.class);
        conf.setHttpServerEnabled(true);
        conf.setProperty(HttpServerLoader.HTTP_SERVER_CLASS, "org.apache.bookkeeper.http.vertx.VertxHttpServer");
        conf.setHttpServerPort(nextFreePort);

        // 1. building the component stack:
        LifecycleComponent server = Main.buildBookieServer(new BookieConfiguration(conf));
        // 2. start the server
        CompletableFuture<Void> stackComponentFuture = ComponentStarter.startComponent(server);
        while (server.lifecycleState() != Lifecycle.State.STARTED) {
            Thread.sleep(100);
        }

        // Now, hit the rest endpoint for metrics
        URL url = new URL("http://localhost:" + nextFreePort + HttpRouter.METRICS);
        URLConnection urlc = url.openConnection();
        BufferedReader in = new BufferedReader(new InputStreamReader(urlc.getInputStream()));
        String inputLine;
        StringBuilder metricsStringBuilder = new StringBuilder();
        while ((inputLine = in.readLine()) != null) {
            metricsStringBuilder.append(inputLine);
        }
        in.close();
        String metrics = metricsStringBuilder.toString();
        // do primitive checks if metrics string contains some stats
        assertTrue("Metrics should contain basic counters", metrics.contains(BookKeeperServerStats.BOOKIE_ADD_ENTRY));

        // Now, hit the rest endpoint for configs
        url = new URL("http://localhost:" + nextFreePort + HttpRouter.SERVER_CONFIG);
        @SuppressWarnings("unchecked")
        Map<String, Object> configMap = om.readValue(url, Map.class);
        if (configMap.isEmpty() || !configMap.containsKey("bookiePort")) {
            Assert.fail("Failed to map configurations to valid JSON entries.");
        }
        stackComponentFuture.cancel(true);
    }

    @Test
    public void testIOVertexHTTPServerEndpointForARWithPrometheusProvider() throws Exception {
        final ServerConfiguration conf = TestBKConfiguration.newServerConfiguration()
                .setMetadataServiceUri(metadataServiceUri).setListeningInterface(null);

        /*
         * enable io.vertx http server
         */
        int nextFreePort = PortManager.nextFreePort();
        conf.setStatsProviderClass(PrometheusMetricsProvider.class);
        conf.setHttpServerEnabled(true);
        conf.setProperty(HttpServerLoader.HTTP_SERVER_CLASS, "org.apache.bookkeeper.http.vertx.VertxHttpServer");
        conf.setHttpServerPort(nextFreePort);

        // 1. building the component stack:
        LifecycleComponent server = AutoRecoveryMain.buildAutoRecoveryServer(new BookieConfiguration(conf));
        // 2. start the server
        CompletableFuture<Void> stackComponentFuture = ComponentStarter.startComponent(server);
        while (server.lifecycleState() != Lifecycle.State.STARTED) {
            Thread.sleep(100);
        }

        // Now, hit the rest endpoint for metrics
        URL url = new URL("http://localhost:" + nextFreePort + HttpRouter.METRICS);
        URLConnection urlc = url.openConnection();
        BufferedReader in = new BufferedReader(new InputStreamReader(urlc.getInputStream()));
        String inputLine;
        StringBuilder metricsStringBuilder = new StringBuilder();
        while ((inputLine = in.readLine()) != null) {
            metricsStringBuilder.append(inputLine);
        }
        in.close();
        String metrics = metricsStringBuilder.toString();
        // do primitive checks if metrics string contains some stats
        assertTrue("Metrics should contain basic counters",
                metrics.contains(ReplicationStats.NUM_FULL_OR_PARTIAL_LEDGERS_REPLICATED));
        assertTrue("Metrics should contain basic counters from BookKeeper client",
                metrics.contains(BookKeeperClientStats.CREATE_OP));

        // Now, hit the rest endpoint for configs
        url = new URL("http://localhost:" + nextFreePort + HttpRouter.SERVER_CONFIG);
        @SuppressWarnings("unchecked")
        Map<String, Object> configMap = om.readValue(url, Map.class);
        if (configMap.isEmpty() || !configMap.containsKey("metadataServiceUri")) {
            Assert.fail("Failed to map configurations to valid JSON entries.");
        }
        stackComponentFuture.cancel(true);
    }

    /**
     * Test that verifies if a bookie can't come up without its cookie in metadata store.
     * @throws Exception
     */
    @Test
    public void testBookieConnectAfterCookieDelete() throws BookieException.UpgradeException {
        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
        conf.setMetadataServiceUri(zkUtil.getMetadataServiceUri());

        try {
            runFunctionWithRegistrationManager(conf, rm -> {
                try {
                    bookieConnectAfterCookieDeleteWorker(conf, rm);
                } catch (BookieException | IOException | InterruptedException e) {
                    fail("Test failed to run: " + e.getMessage());
                }
                return null;
            });
        } catch (MetadataException | ExecutionException e) {
            throw new BookieException.UpgradeException(e);
        }
    }

    private void bookieConnectAfterCookieDeleteWorker(ServerConfiguration conf, RegistrationManager rm)
            throws BookieException, InterruptedException, IOException {

        File tmpLedgerDir = createTempDir("BootupTest", "test");
        File tmpJournalDir = createTempDir("BootupTest", "test");
        Integer numOfJournalDirs = 2;

        String[] journalDirs = new String[numOfJournalDirs];
        for (int i = 0; i < numOfJournalDirs; i++) {
            journalDirs[i] = tmpJournalDir.getAbsolutePath() + "/journal-" + i;
        }

        conf.setJournalDirsName(journalDirs);
        conf.setLedgerDirNames(new String[] { tmpLedgerDir.getPath() });

        Bookie b = new BookieImpl(conf);

        final BookieId bookieAddress = BookieImpl.getBookieId(conf);

        // Read cookie from registation manager
        Versioned<Cookie> rmCookie = Cookie.readFromRegistrationManager(rm, bookieAddress);

        // Shutdown bookie
        b.shutdown();

        // Remove cookie from registration manager
        rmCookie.getValue().deleteFromRegistrationManager(rm, conf, rmCookie.getVersion());

        try {
            b = new BookieImpl(conf);
            Assert.fail("Bookie should not have come up. Cookie no present in metadata store.");
        } catch (Exception e) {
            LOG.info("As expected Bookie fails to come up without a cookie in metadata store.");
        }
    }

    @Test
    public void testInvalidServiceMetadataURI() throws Exception {
       testInvalidServiceMetadataURICase("zk+null:///ledgers"); // no hostname
       testInvalidServiceMetadataURICase("zk+null://ledgers");
       testInvalidServiceMetadataURICase("zk+null:ledgers");
        {
            ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
            conf.setMetadataServiceUri("//ledgers");
            try {
                new BookieServer(conf);
                Assert.fail("Bookie metadata initialization must fail with metadata service uri: //ledgers");
            } catch (NullPointerException e) {
                assertTrue(e.getMessage().contains("Invalid metadata service uri : //ledgers"));
            }
        }

        {
            ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
            conf.setMetadataServiceUri("");
            try {
                new BookieServer(conf);
                Assert.fail("Bookie metadata initialization must fail with empty metadata service uri");
            } catch (NullPointerException e) {
                assertTrue(e.getMessage().contains("Invalid metadata service uri :"));
            }
        }
    }

    private void testInvalidServiceMetadataURICase(String uri) throws Exception {
        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
        conf.setMetadataServiceUri(uri);
        try {
            new BookieServer(conf);
            Assert.fail("Bookie metadata initialization must fail with an invalid metadata service uri: " + uri);
        } catch (MetadataStoreException e) {
            // ok
        }
    }
}