bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ExplicitLacTest.java

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


import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import java.util.Arrays;
import java.util.Collection;
import java.util.Enumeration;
import org.apache.bookkeeper.bookie.InterleavedLedgerStorage;
import org.apache.bookkeeper.bookie.LedgerStorage;
import org.apache.bookkeeper.bookie.SortedLedgerStorage;
import org.apache.bookkeeper.bookie.storage.ldb.DbLedgerStorage;
import org.apache.bookkeeper.client.BookKeeper.DigestType;
import org.apache.bookkeeper.client.api.WriteFlag;
import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
import org.apache.bookkeeper.util.TestUtils;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameters;

/**
 * Test cases for `Explicit Lac` feature.
 */
@RunWith(Parameterized.class)
public class ExplicitLacTest extends BookKeeperClusterTestCase {

    private final DigestType digestType;

    public ExplicitLacTest(Class<? extends LedgerStorage> storageClass) {
        super(1);
        this.digestType = DigestType.CRC32;
        baseConf.setLedgerStorageClass(storageClass.getName());
        /*
         * to persist explicitLac, journalFormatVersionToWrite should be atleast
         * V6 and fileInfoFormatVersionToWrite should be atleast V1
         */
        baseConf.setJournalFormatVersionToWrite(6);
        baseConf.setFileInfoFormatVersionToWrite(1);
    }

    @Parameters
    public static Collection<Object[]> configs() {
        return Arrays.asList(new Object[][] {
            { InterleavedLedgerStorage.class },
            { SortedLedgerStorage.class },
            { DbLedgerStorage.class },
        });
    }

    @Test
    public void testReadHandleWithNoExplicitLAC() throws Exception {
        ClientConfiguration confWithNoExplicitLAC = new ClientConfiguration();
        confWithNoExplicitLAC.setMetadataServiceUri(zkUtil.getMetadataServiceUri());
        confWithNoExplicitLAC.setExplictLacInterval(0);

        BookKeeper bkcWithNoExplicitLAC = new BookKeeper(confWithNoExplicitLAC);

        LedgerHandle wlh = bkcWithNoExplicitLAC.createLedger(
            1, 1, 1,
            digestType, "testPasswd".getBytes());
        long ledgerId = wlh.getId();
        int numOfEntries = 5;
        for (int i = 0; i < numOfEntries; i++) {
            wlh.addEntry(("foobar" + i).getBytes());
        }

        LedgerHandle rlh = bkcWithNoExplicitLAC.openLedgerNoRecovery(ledgerId, digestType, "testPasswd".getBytes());
        assertTrue(
                "Expected LAC of rlh: " + (numOfEntries - 2) + " actual LAC of rlh: " + rlh.getLastAddConfirmed(),
                (rlh.getLastAddConfirmed() == (numOfEntries - 2)));

        Enumeration<LedgerEntry> entries = rlh.readEntries(0, numOfEntries - 2);
        int entryId = 0;
        while (entries.hasMoreElements()) {
            LedgerEntry entry = entries.nextElement();
            String entryString = new String(entry.getEntry());
            assertTrue("Expected entry String: " + ("foobar" + entryId) + " actual entry String: " + entryString,
                    entryString.equals("foobar" + entryId));
            entryId++;
        }

        for (int i = numOfEntries; i < 2 * numOfEntries; i++) {
            wlh.addEntry(("foobar" + i).getBytes());
        }

        TestUtils.waitUntilLacUpdated(rlh, numOfEntries - 2);

        assertTrue(
                "Expected LAC of wlh: " + (2 * numOfEntries - 1) + " actual LAC of rlh: " + wlh.getLastAddConfirmed(),
                (wlh.getLastAddConfirmed() == (2 * numOfEntries - 1)));
        assertTrue(
                "Expected LAC of rlh: " + (numOfEntries - 2) + " actual LAC of rlh: " + rlh.getLastAddConfirmed(),
                (rlh.getLastAddConfirmed() == (numOfEntries - 2)));

        // since explicitlacflush policy is not enabled for writeledgerhandle, when we try
        // to read explicitlac for rlh, it will be reading up to the piggyback value.
        long explicitlac = rlh.readExplicitLastConfirmed();
        assertTrue(
                "Expected Explicit LAC of rlh: " + (numOfEntries - 2) + " actual ExplicitLAC of rlh: " + explicitlac,
                (explicitlac == (2 * numOfEntries - 2)));

        try {
            rlh.readEntries(2 * numOfEntries - 1, 2 * numOfEntries - 1);
            fail("rlh readEntries beyond " + (2 * numOfEntries - 2) + " should fail with ReadException");
        } catch (BKException.BKReadException readException) {
        }

        rlh.close();
        wlh.close();
        bkcWithNoExplicitLAC.close();
    }

    @Test
    public void testExplicitLACIsPersisted() throws Exception {
        ClientConfiguration confWithNoExplicitLAC = new ClientConfiguration();
        confWithNoExplicitLAC.setMetadataServiceUri(zkUtil.getMetadataServiceUri());
        // enable explicitLacFlush by setting non-zero value for
        // explictLacInterval
        confWithNoExplicitLAC.setExplictLacInterval(50);

        BookKeeper bkcWithExplicitLAC = new BookKeeper(confWithNoExplicitLAC);

        LedgerHandle wlh = bkcWithExplicitLAC.createLedger(1, 1, 1, digestType, "testPasswd".getBytes());
        long ledgerId = wlh.getId();
        int numOfEntries = 5;
        for (int i = 0; i < numOfEntries; i++) {
            wlh.addEntry(("foobar" + i).getBytes());
        }

        LedgerHandle rlh = bkcWithExplicitLAC.openLedgerNoRecovery(ledgerId, digestType, "testPasswd".getBytes());
        assertEquals("LAC of rlh", (long) numOfEntries - 2, rlh.getLastAddConfirmed());

        for (int i = numOfEntries; i < 2 * numOfEntries; i++) {
            wlh.addEntry(("foobar" + i).getBytes());
        }

        assertEquals("LAC of wlh", (2 * numOfEntries - 1), wlh.getLastAddConfirmed());
        assertEquals("LAC of rlh", (long) numOfEntries - 2, rlh.getLastAddConfirmed());
        assertEquals("Read LAC of rlh", (2 * numOfEntries - 2), rlh.readLastAddConfirmed());
        assertEquals("Read explicit LAC of rlh", (2 * numOfEntries - 2), rlh.readExplicitLastConfirmed());

        // we need to wait for atleast 2 explicitlacintervals,
        // since in writehandle for the first call
        // lh.getExplicitLastAddConfirmed() will be <
        // lh.getPiggyBackedLastAddConfirmed(),
        // so it wont make explicit writelac in the first run
        long readExplicitLastConfirmed = TestUtils.waitUntilExplicitLacUpdated(rlh, 2 * numOfEntries - 1);
        assertEquals("Read explicit LAC of rlh after wait for explicitlacflush", (2 * numOfEntries - 1),
                readExplicitLastConfirmed);

        // bookies have to be restarted
        restartBookies();

        /*
         * since explicitLac is persisted we should be able to read explicitLac
         * from the bookies.
         */
        LedgerHandle rlh2 = bkcWithExplicitLAC.openLedgerNoRecovery(ledgerId, digestType, "testPasswd".getBytes());
        assertEquals("Read explicit LAC of rlh2 after bookies restart", (2 * numOfEntries - 1),
                rlh2.readExplicitLastConfirmed());
        bkcWithExplicitLAC.close();
    }

    @Test
    public void testReadHandleWithExplicitLAC() throws Exception {
        ClientConfiguration confWithExplicitLAC = new ClientConfiguration();
        confWithExplicitLAC.setMetadataServiceUri(zkUtil.getMetadataServiceUri());
        int explicitLacIntervalMillis = 1000;
        confWithExplicitLAC.setExplictLacInterval(explicitLacIntervalMillis);

        BookKeeper bkcWithExplicitLAC = new BookKeeper(confWithExplicitLAC);

        LedgerHandle wlh = bkcWithExplicitLAC.createLedger(
            1, 1, 1,
            digestType, "testPasswd".getBytes());
        long ledgerId = wlh.getId();
        int numOfEntries = 5;
        for (int i = 0; i < numOfEntries; i++) {
            wlh.addEntry(("foobar" + i).getBytes());
        }

        LedgerHandle rlh = bkcWithExplicitLAC.openLedgerNoRecovery(ledgerId, digestType, "testPasswd".getBytes());

        assertTrue(
                "Expected LAC of rlh: " + (numOfEntries - 2) + " actual LAC of rlh: " + rlh.getLastAddConfirmed(),
                (rlh.getLastAddConfirmed() == (numOfEntries - 2)));

        for (int i = numOfEntries; i < 2 * numOfEntries; i++) {
            wlh.addEntry(("foobar" + i).getBytes());
        }

        // we need to wait for atleast 2 explicitlacintervals,
        // since in writehandle for the first call
        // lh.getExplicitLastAddConfirmed() will be <
        // lh.getPiggyBackedLastAddConfirmed(),
        // so it wont make explicit writelac in the first run
        TestUtils.waitUntilLacUpdated(rlh, 2 * numOfEntries - 2);

        assertTrue(
                "Expected LAC of wlh: " + (2 * numOfEntries - 1) + " actual LAC of wlh: " + wlh.getLastAddConfirmed(),
                (wlh.getLastAddConfirmed() == (2 * numOfEntries - 1)));
        // readhandle's lastaddconfirmed wont be updated until readExplicitLastConfirmed call is made
        assertTrue(
                "Expected LAC of rlh: " + (2 * numOfEntries - 2) + " actual LAC of rlh: " + rlh.getLastAddConfirmed(),
                (rlh.getLastAddConfirmed() == (2 * numOfEntries - 2)));

        long explicitlac = TestUtils.waitUntilExplicitLacUpdated(rlh, 2 * numOfEntries - 1);
        assertTrue("Expected Explicit LAC of rlh: " + (2 * numOfEntries - 1)
                + " actual ExplicitLAC of rlh: " + explicitlac,
                (explicitlac == (2 * numOfEntries - 1)));
        // readExplicitLastConfirmed updates the lac of rlh.
        assertTrue(
                "Expected LAC of rlh: " + (2 * numOfEntries - 1) + " actual LAC of rlh: " + rlh.getLastAddConfirmed(),
                (rlh.getLastAddConfirmed() == (2 * numOfEntries - 1)));

        Enumeration<LedgerEntry> entries = rlh.readEntries(numOfEntries, 2 * numOfEntries - 1);
        int entryId = numOfEntries;
        while (entries.hasMoreElements()) {
            LedgerEntry entry = entries.nextElement();
            String entryString = new String(entry.getEntry());
            assertTrue("Expected entry String: " + ("foobar" + entryId) + " actual entry String: " + entryString,
                    entryString.equals("foobar" + entryId));
            entryId++;
        }

        rlh.close();
        wlh.close();
        bkcWithExplicitLAC.close();
    }

    @Test
    public void testReadHandleWithExplicitLACAndDeferredSync() throws Exception {
        ClientConfiguration confWithExplicitLAC = new ClientConfiguration();
        confWithExplicitLAC.setMetadataServiceUri(zkUtil.getMetadataServiceUri());
        int explicitLacIntervalMillis = 1000;
        confWithExplicitLAC.setExplictLacInterval(explicitLacIntervalMillis);

        BookKeeper bkcWithExplicitLAC = new BookKeeper(confWithExplicitLAC);

        LedgerHandle wlh = (LedgerHandle) bkcWithExplicitLAC.newCreateLedgerOp()
                .withEnsembleSize(1)
                .withWriteQuorumSize(1)
                .withAckQuorumSize(1)
                .withWriteFlags(WriteFlag.DEFERRED_SYNC)
                .withDigestType(digestType.toApiDigestType())
                .withPassword("testPasswd".getBytes())
                .execute()
                .get();
        long ledgerId = wlh.getId();

        // start like testReadHandleWithExplicitLAC
        int numOfEntries = 5;
        for (int i = 0; i < numOfEntries; i++) {
            // if you perform force() + addEntry() you will piggy back LAC as usual
            wlh.force().get();
            wlh.addEntry(("foobar" + i).getBytes());
        }

        LedgerHandle rlh = bkcWithExplicitLAC.openLedgerNoRecovery(ledgerId, digestType, "testPasswd".getBytes());

        assertTrue(
                "Expected LAC of rlh: " + (numOfEntries - 2) + " actual LAC of rlh: " + rlh.getLastAddConfirmed(),
                (rlh.getLastAddConfirmed() == (numOfEntries - 2)));

        for (int i = numOfEntries; i < 2 * numOfEntries; i++) {
            wlh.addEntry(("foobar" + i).getBytes());
        }

        // running a force() will update local LAC on the writer
        // ExplicitLAC timer will send the value even without writes
        wlh.force().get();

        // wait for explicit lac to be sent to bookies
        TestUtils.waitUntilExplicitLacUpdated(rlh, 2 * numOfEntries - 2);

        // we need to wait for atleast 2 explicitlacintervals,
        // since in writehandle for the first call
        // lh.getExplicitLastAddConfirmed() will be <
        // lh.getPiggyBackedLastAddConfirmed(),
        // so it wont make explicit writelac in the first run
        TestUtils.waitUntilLacUpdated(rlh, 2 * numOfEntries - 2);

        assertTrue(
                "Expected LAC of wlh: " + (2 * numOfEntries - 1) + " actual LAC of wlh: " + wlh.getLastAddConfirmed(),
                (wlh.getLastAddConfirmed() == (2 * numOfEntries - 1)));

        long explicitlac = TestUtils.waitUntilExplicitLacUpdated(rlh, 2 * numOfEntries - 1);
        assertTrue("Expected Explicit LAC of rlh: " + (2 * numOfEntries - 1)
                + " actual ExplicitLAC of rlh: " + explicitlac,
                (explicitlac == (2 * numOfEntries - 1)));
        // readExplicitLastConfirmed updates the lac of rlh.
        assertTrue(
                "Expected LAC of rlh: " + (2 * numOfEntries - 1) + " actual LAC of rlh: " + rlh.getLastAddConfirmed(),
                (rlh.getLastAddConfirmed() == (2 * numOfEntries - 1)));

        Enumeration<LedgerEntry> entries = rlh.readEntries(numOfEntries, 2 * numOfEntries - 1);
        int entryId = numOfEntries;
        while (entries.hasMoreElements()) {
            LedgerEntry entry = entries.nextElement();
            String entryString = new String(entry.getEntry());
            assertTrue("Expected entry String: " + ("foobar" + entryId) + " actual entry String: " + entryString,
                    entryString.equals("foobar" + entryId));
            entryId++;
        }

        rlh.close();
        wlh.close();
        bkcWithExplicitLAC.close();
    }

    @Test
    public void fallbackV3() throws Exception {
        ClientConfiguration v2Conf = new ClientConfiguration();
        v2Conf.setUseV2WireProtocol(true);
        v2Conf.setMetadataServiceUri(zkUtil.getMetadataServiceUri());
        v2Conf.setExplictLacInterval(10);

        BookKeeper bookKeeper = new BookKeeper(v2Conf);
        LedgerHandle write = (LedgerHandle) bookKeeper.createLedger(1,
                                                                    1,
                                                                    1,
                                                                    DigestType.MAC,
                                                                    "pass".getBytes());
        write.addEntry("test".getBytes());
        TestUtils.waitUntilExplicitLacUpdated(write, 0);
        long lac = write.readExplicitLastConfirmed();
        assertEquals(0, lac);
        write.close();
        bookKeeper.close();
    }

}