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

import org.apache.distributedlog.exceptions.BKTransmitException;
import org.apache.distributedlog.exceptions.EndOfStreamException;
import org.apache.distributedlog.exceptions.WriteCancelledException;
import org.apache.distributedlog.exceptions.WriteException;
import org.apache.distributedlog.impl.BKNamespaceDriver;
import org.apache.distributedlog.impl.logsegment.BKLogSegmentEntryWriter;
import org.apache.distributedlog.io.Abortables;
import org.apache.distributedlog.lock.SessionLockFactory;
import org.apache.distributedlog.lock.ZKDistributedLock;
import org.apache.distributedlog.lock.ZKSessionLockFactory;
import org.apache.distributedlog.impl.metadata.BKDLConfig;
import org.apache.distributedlog.util.ConfUtils;
import org.apache.distributedlog.util.FutureUtils;
import org.apache.distributedlog.util.OrderedScheduler;
import org.apache.distributedlog.util.PermitLimiter;
import org.apache.distributedlog.util.Utils;
import com.twitter.util.Await;
import com.twitter.util.Future;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.LedgerHandle;
import org.apache.bookkeeper.feature.SettableFeatureProvider;
import org.apache.bookkeeper.stats.AlertStatsLogger;
import org.apache.bookkeeper.stats.NullStatsLogger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestName;
import scala.runtime.AbstractFunction0;

import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;

import static com.google.common.base.Charsets.UTF_8;
import static org.junit.Assert.*;

/**
 * Test Case for BookKeeper Based Log Segment Writer
 */
public class TestBKLogSegmentWriter extends TestDistributedLogBase {

    @Rule
    public TestName runtime = new TestName();

    private OrderedScheduler scheduler;
    private OrderedScheduler lockStateExecutor;
    private ZooKeeperClient zkc;
    private ZooKeeperClient zkc0;
    private BookKeeperClient bkc;

    @Before
    @Override
    public void setup() throws Exception {
        super.setup();
        scheduler = OrderedScheduler.newBuilder().corePoolSize(1).build();
        lockStateExecutor = OrderedScheduler.newBuilder().corePoolSize(1).build();
        // build zookeeper client
        URI uri = createDLMURI("");
        zkc = TestZooKeeperClientBuilder.newBuilder(conf)
                .name("test-zkc")
                .uri(uri)
                .build();
        zkc0 = TestZooKeeperClientBuilder.newBuilder(conf)
                .name("test-zkc0")
                .uri(uri)
                .build();
        // build bookkeeper client
        BKDLConfig bkdlConfig = BKDLConfig.resolveDLConfig(zkc, uri);
        bkc = BookKeeperClientBuilder.newBuilder()
                .dlConfig(conf)
                .name("test-bkc")
                .ledgersPath(bkdlConfig.getBkLedgersPath())
                .zkServers(BKNamespaceDriver.getZKServersFromDLUri(uri))
                .build();
    }

    @After
    @Override
    public void teardown() throws Exception {
        if (null != bkc) {
            bkc.close();
        }
        if (null != zkc) {
            zkc.close();
        }
        if (null != lockStateExecutor) {
            lockStateExecutor.shutdown();
        }
        if (null != scheduler) {
            scheduler.shutdown();
        }
        super.teardown();
    }

    private DistributedLogConfiguration newLocalConf() {
        DistributedLogConfiguration confLocal = new DistributedLogConfiguration();
        confLocal.addConfiguration(conf);
        return confLocal;
    }

    private ZKDistributedLock createLock(String path,
                                         ZooKeeperClient zkClient,
                                         boolean acquireLock)
            throws Exception {
        try {
            Await.result(Utils.zkAsyncCreateFullPathOptimistic(zkClient, path, new byte[0],
                    ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT));
        } catch (KeeperException.NodeExistsException nee) {
            // node already exists
        }
        SessionLockFactory lockFactory = new ZKSessionLockFactory(
                zkClient,
                "test-lock",
                lockStateExecutor,
                0,
                Long.MAX_VALUE,
                conf.getZKSessionTimeoutMilliseconds(),
                NullStatsLogger.INSTANCE
        );
        ZKDistributedLock lock = new ZKDistributedLock(
                lockStateExecutor,
                lockFactory,
                path,
                Long.MAX_VALUE,
                NullStatsLogger.INSTANCE);
        if (acquireLock) {
            return FutureUtils.result(lock.asyncAcquire());
        } else {
            return lock;
        }
    }

    private void closeWriterAndLock(BKLogSegmentWriter writer,
                                    ZKDistributedLock lock)
            throws IOException {
        try {
            FutureUtils.result(writer.asyncClose());
        } finally {
            Utils.closeQuietly(lock);
        }
    }

    private void abortWriterAndLock(BKLogSegmentWriter writer,
                                    ZKDistributedLock lock)
            throws IOException {
        try {
            Abortables.abort(writer, false);
        } finally {
            Utils.closeQuietly(lock);
        }
    }

    private BKLogSegmentWriter createLogSegmentWriter(DistributedLogConfiguration conf,
                                                      long logSegmentSequenceNumber,
                                                      long startTxId,
                                                      ZKDistributedLock lock) throws Exception {
        LedgerHandle lh = bkc.get().createLedger(3, 2, 2,
                BookKeeper.DigestType.CRC32, conf.getBKDigestPW().getBytes(UTF_8));
        return new BKLogSegmentWriter(
                runtime.getMethodName(),
                runtime.getMethodName(),
                conf,
                LogSegmentMetadata.LEDGER_METADATA_CURRENT_LAYOUT_VERSION,
                new BKLogSegmentEntryWriter(lh),
                lock,
                startTxId,
                logSegmentSequenceNumber,
                scheduler,
                NullStatsLogger.INSTANCE,
                NullStatsLogger.INSTANCE,
                new AlertStatsLogger(NullStatsLogger.INSTANCE, "test"),
                PermitLimiter.NULL_PERMIT_LIMITER,
                new SettableFeatureProvider("", 0),
                ConfUtils.getConstDynConf(conf));
    }

    private LedgerHandle openLedgerNoRecovery(LedgerHandle lh) throws Exception {
        return bkc.get().openLedgerNoRecovery(lh.getId(),
                BookKeeper.DigestType.CRC32, conf.getBKDigestPW().getBytes(UTF_8));
    }

    private LedgerHandle openLedger(LedgerHandle lh) throws Exception {
        return bkc.get().openLedger(lh.getId(),
                BookKeeper.DigestType.CRC32, conf.getBKDigestPW().getBytes(UTF_8));
    }

    private void fenceLedger(LedgerHandle lh) throws Exception {
        bkc.get().openLedger(lh.getId(), BookKeeper.DigestType.CRC32,
                conf.getBKDigestPW().getBytes(UTF_8));
    }

    /**
     * Close a segment log writer should flush buffered data.
     *
     * @throws Exception
     */
    @Test(timeout = 60000)
    public void testCloseShouldFlush() throws Exception {
        DistributedLogConfiguration confLocal = newLocalConf();
        confLocal.setImmediateFlushEnabled(false);
        confLocal.setOutputBufferSize(Integer.MAX_VALUE);
        confLocal.setPeriodicFlushFrequencyMilliSeconds(0);
        ZKDistributedLock lock = createLock("/test/lock-" + runtime.getMethodName(), zkc, true);
        BKLogSegmentWriter writer =
                createLogSegmentWriter(confLocal, 0L, -1L, lock);
        // Use another lock to wait for writer releasing lock
        ZKDistributedLock lock0 = createLock("/test/lock-" + runtime.getMethodName(), zkc0, false);
        Future<ZKDistributedLock> lockFuture0 = lock0.asyncAcquire();
        // add 10 records
        int numRecords = 10;
        List<Future<DLSN>> futureList = new ArrayList<Future<DLSN>>(numRecords);
        for (int i = 0; i < numRecords; i++) {
            futureList.add(writer.asyncWrite(DLMTestUtil.getLogRecordInstance(i)));
        }
        assertEquals("Last tx id should be " + (numRecords - 1),
                numRecords - 1, writer.getLastTxId());
        assertEquals("Last acked tx id should be -1",
                -1L, writer.getLastTxIdAcknowledged());
        assertEquals("Last DLSN should be " + DLSN.InvalidDLSN,
                DLSN.InvalidDLSN, writer.getLastDLSN());
        assertEquals("Position should be " + numRecords,
                10, writer.getPositionWithinLogSegment());
        // close the writer should flush buffered data and release lock
        closeWriterAndLock(writer, lock);
        Await.result(lockFuture0);
        lock0.checkOwnership();
        assertEquals("Last tx id should still be " + (numRecords - 1),
                numRecords - 1, writer.getLastTxId());
        assertEquals("Last acked tx id should become " + (numRecords - 1),
                numRecords - 1, writer.getLastTxIdAcknowledged());
        assertEquals("Position should still be " + numRecords,
                10, writer.getPositionWithinLogSegment());
        List<DLSN> dlsns = Await.result(Future.collect(futureList));
        assertEquals("All records should be written",
                numRecords, dlsns.size());
        for (int i = 0; i < numRecords; i++) {
            DLSN dlsn = dlsns.get(i);
            assertEquals("Incorrent ledger sequence number",
                    0L, dlsn.getLogSegmentSequenceNo());
            assertEquals("Incorrent entry id",
                    0L, dlsn.getEntryId());
            assertEquals("Inconsistent slot id",
                    i, dlsn.getSlotId());
        }
        assertEquals("Last DLSN should be " + dlsns.get(dlsns.size() - 1),
                dlsns.get(dlsns.size() - 1), writer.getLastDLSN());
        LedgerHandle lh = getLedgerHandle(writer);
        LedgerHandle readLh = openLedgerNoRecovery(lh);
        assertTrue("Ledger " + lh.getId() + " should be closed", readLh.isClosed());
        assertEquals("There should be two entries in ledger " + lh.getId(),
                1L, readLh.getLastAddConfirmed());
    }

    /**
     * Abort a segment log writer should just abort pending writes and not flush buffered data.
     *
     * @throws Exception
     */
    @Test(timeout = 60000)
    public void testAbortShouldNotFlush() throws Exception {
        DistributedLogConfiguration confLocal = newLocalConf();
        confLocal.setImmediateFlushEnabled(false);
        confLocal.setOutputBufferSize(Integer.MAX_VALUE);
        confLocal.setPeriodicFlushFrequencyMilliSeconds(0);
        ZKDistributedLock lock = createLock("/test/lock-" + runtime.getMethodName(), zkc, true);
        BKLogSegmentWriter writer =
                createLogSegmentWriter(confLocal, 0L, -1L, lock);
        // Use another lock to wait for writer releasing lock
        ZKDistributedLock lock0 = createLock("/test/lock-" + runtime.getMethodName(), zkc0, false);
        Future<ZKDistributedLock> lockFuture0 = lock0.asyncAcquire();
        // add 10 records
        int numRecords = 10;
        List<Future<DLSN>> futureList = new ArrayList<Future<DLSN>>(numRecords);
        for (int i = 0; i < numRecords; i++) {
            futureList.add(writer.asyncWrite(DLMTestUtil.getLogRecordInstance(i)));
        }
        assertEquals("Last tx id should be " + (numRecords - 1),
                numRecords - 1, writer.getLastTxId());
        assertEquals("Last acked tx id should be -1",
                -1L, writer.getLastTxIdAcknowledged());
        assertEquals("Last DLSN should be " + DLSN.InvalidDLSN,
                DLSN.InvalidDLSN, writer.getLastDLSN());
        assertEquals("Position should be " + numRecords,
                10, writer.getPositionWithinLogSegment());
        // close the writer should flush buffered data and release lock
        abortWriterAndLock(writer, lock);
        Await.result(lockFuture0);
        lock0.checkOwnership();
        assertEquals("Last tx id should still be " + (numRecords - 1),
                numRecords - 1, writer.getLastTxId());
        assertEquals("Last acked tx id should still be " + (numRecords - 1),
                -1L, writer.getLastTxIdAcknowledged());
        assertEquals("Last DLSN should still be " + DLSN.InvalidDLSN,
                DLSN.InvalidDLSN, writer.getLastDLSN());
        assertEquals("Position should still be " + numRecords,
                10, writer.getPositionWithinLogSegment());

        for (int i = 0; i < numRecords; i++) {
            try {
                Await.result(futureList.get(i));
                fail("Should be aborted record " + i + " with transmit exception");
            } catch (WriteCancelledException wce) {
                assertTrue("Record " + i + " should be aborted because of ledger fenced",
                        wce.getCause() instanceof BKTransmitException);
                BKTransmitException bkte = (BKTransmitException) wce.getCause();
                assertEquals("Record " + i + " should be aborted",
                        BKException.Code.InterruptedException, bkte.getBKResultCode());
            }
        }

        // check no entries were written
        LedgerHandle lh = getLedgerHandle(writer);
        LedgerHandle readLh = openLedgerNoRecovery(lh);
        assertTrue("Ledger " + lh.getId() + " should not be closed", readLh.isClosed());
        assertEquals("There should be no entries in ledger " + lh.getId(),
                LedgerHandle.INVALID_ENTRY_ID, readLh.getLastAddConfirmed());
    }


    /**
     * Close a log segment writer that already detect ledger fenced, should not flush buffered data.
     * And should throw exception on closing.
     *
     * @throws Exception
     */
    @Test(timeout = 60000)
    public void testCloseShouldNotFlushIfLedgerFenced() throws Exception {
        testCloseShouldNotFlushIfInErrorState(BKException.Code.LedgerFencedException);
    }

    /**
     * Close a log segment writer that is already in error state, should not flush buffered data.
     *
     * @throws Exception
     */
    void testCloseShouldNotFlushIfInErrorState(int rcToFailComplete) throws Exception {
        DistributedLogConfiguration confLocal = newLocalConf();
        confLocal.setImmediateFlushEnabled(false);
        confLocal.setOutputBufferSize(Integer.MAX_VALUE);
        confLocal.setPeriodicFlushFrequencyMilliSeconds(0);
        ZKDistributedLock lock = createLock("/test/lock-" + runtime.getMethodName(), zkc, true);
        BKLogSegmentWriter writer =
                createLogSegmentWriter(confLocal, 0L, -1L, lock);
        // Use another lock to wait for writer releasing lock
        ZKDistributedLock lock0 = createLock("/test/lock-" + runtime.getMethodName(), zkc0, false);
        Future<ZKDistributedLock> lockFuture0 = lock0.asyncAcquire();
        // add 10 records
        int numRecords = 10;
        List<Future<DLSN>> futureList = new ArrayList<Future<DLSN>>(numRecords);
        for (int i = 0; i < numRecords; i++) {
            futureList.add(writer.asyncWrite(DLMTestUtil.getLogRecordInstance(i)));
        }
        assertEquals("Last tx id should be " + (numRecords - 1),
                numRecords - 1, writer.getLastTxId());
        assertEquals("Last acked tx id should be -1",
                -1L, writer.getLastTxIdAcknowledged());
        assertEquals("Last DLSN should be " + DLSN.InvalidDLSN,
                DLSN.InvalidDLSN, writer.getLastDLSN());
        assertEquals("Position should be " + numRecords,
                10, writer.getPositionWithinLogSegment());
        writer.setTransmitResult(rcToFailComplete);
        // close the writer should release lock but not flush data
        try {
            closeWriterAndLock(writer, lock);
            fail("Close a log segment writer in error state should throw exception");
        } catch (BKTransmitException bkte) {
            assertEquals("Inconsistent rc is thrown",
                    rcToFailComplete, bkte.getBKResultCode());
        }
        Await.result(lockFuture0);
        lock0.checkOwnership();
        assertEquals("Last tx id should still be " + (numRecords - 1),
                numRecords - 1, writer.getLastTxId());
        assertEquals("Last acked tx id should still be " + (numRecords - 1),
                -1L, writer.getLastTxIdAcknowledged());
        assertEquals("Last DLSN should still be " + DLSN.InvalidDLSN,
                DLSN.InvalidDLSN, writer.getLastDLSN());
        assertEquals("Position should still be " + numRecords,
                10, writer.getPositionWithinLogSegment());

        for (int i = 0; i < numRecords; i++) {
            try {
                Await.result(futureList.get(i));
                fail("Should be aborted record " + i + " with transmit exception");
            } catch (WriteCancelledException wce) {
                assertTrue("Record " + i + " should be aborted because of ledger fenced",
                        wce.getCause() instanceof BKTransmitException);
                BKTransmitException bkte = (BKTransmitException) wce.getCause();
                assertEquals("Record " + i + " should be aborted",
                        rcToFailComplete, bkte.getBKResultCode());
            }
        }

        // check no entries were written
        LedgerHandle lh = getLedgerHandle(writer);
        LedgerHandle readLh = openLedgerNoRecovery(lh);
        assertFalse("Ledger " + lh.getId() + " should not be closed", readLh.isClosed());
        assertEquals("There should be no entries in ledger " + lh.getId(),
                LedgerHandle.INVALID_ENTRY_ID, readLh.getLastAddConfirmed());
    }

    /**
     * Close the writer when ledger is fenced: it should release the lock, fail on flushing data and throw exception
     *
     * @throws Exception
     */
    @Test(timeout = 60000)
    public void testCloseShouldFailIfLedgerFenced() throws Exception {
        DistributedLogConfiguration confLocal = newLocalConf();
        confLocal.setImmediateFlushEnabled(false);
        confLocal.setOutputBufferSize(Integer.MAX_VALUE);
        confLocal.setPeriodicFlushFrequencyMilliSeconds(0);
        ZKDistributedLock lock = createLock("/test/lock-" + runtime.getMethodName(), zkc, true);
        BKLogSegmentWriter writer =
                createLogSegmentWriter(confLocal, 0L, -1L, lock);
        // Use another lock to wait for writer releasing lock
        ZKDistributedLock lock0 = createLock("/test/lock-" + runtime.getMethodName(), zkc0, false);
        Future<ZKDistributedLock> lockFuture0 = lock0.asyncAcquire();
        // add 10 records
        int numRecords = 10;
        List<Future<DLSN>> futureList = new ArrayList<Future<DLSN>>(numRecords);
        for (int i = 0; i < numRecords; i++) {
            futureList.add(writer.asyncWrite(DLMTestUtil.getLogRecordInstance(i)));
        }
        assertEquals("Last tx id should be " + (numRecords - 1),
                numRecords - 1, writer.getLastTxId());
        assertEquals("Last acked tx id should be -1",
                -1L, writer.getLastTxIdAcknowledged());
        assertEquals("Last DLSN should be " + DLSN.InvalidDLSN,
                DLSN.InvalidDLSN, writer.getLastDLSN());
        assertEquals("Position should be " + numRecords,
                10, writer.getPositionWithinLogSegment());
        // fence the ledger
        fenceLedger(getLedgerHandle(writer));
        // close the writer: it should release the lock, fail on flushing data and throw exception
        try {
            closeWriterAndLock(writer, lock);
            fail("Close a log segment writer when ledger is fenced should throw exception");
        } catch (BKTransmitException bkte) {
            assertEquals("Inconsistent rc is thrown",
                    BKException.Code.LedgerFencedException, bkte.getBKResultCode());
        }

        Await.result(lockFuture0);
        lock0.checkOwnership();

        assertEquals("Last tx id should still be " + (numRecords - 1),
                numRecords - 1, writer.getLastTxId());
        assertEquals("Last acked tx id should still be " + (numRecords - 1),
                -1L, writer.getLastTxIdAcknowledged());
        assertEquals("Last DLSN should still be " + DLSN.InvalidDLSN,
                DLSN.InvalidDLSN, writer.getLastDLSN());
        assertEquals("Position should still be " + numRecords,
                10, writer.getPositionWithinLogSegment());

        for (int i = 0; i < numRecords; i++) {
            try {
                Await.result(futureList.get(i));
                fail("Should be aborted record " + i + " with transmit exception");
            } catch (BKTransmitException bkte) {
                assertEquals("Record " + i + " should be aborted",
                        BKException.Code.LedgerFencedException, bkte.getBKResultCode());
            }
        }

        // check no entries were written
        LedgerHandle lh = getLedgerHandle(writer);
        LedgerHandle readLh = openLedgerNoRecovery(lh);
        assertTrue("Ledger " + lh.getId() + " should be closed", readLh.isClosed());
        assertEquals("There should be no entries in ledger " + lh.getId(),
                LedgerHandle.INVALID_ENTRY_ID, readLh.getLastAddConfirmed());
    }

    /**
     * Abort should wait for outstanding transmits to be completed and cancel buffered data.
     *
     * @throws Exception
     */
    @Test(timeout = 60000)
    public void testAbortShouldFailAllWrites() throws Exception {
        DistributedLogConfiguration confLocal = newLocalConf();
        confLocal.setImmediateFlushEnabled(false);
        confLocal.setOutputBufferSize(Integer.MAX_VALUE);
        confLocal.setPeriodicFlushFrequencyMilliSeconds(0);
        ZKDistributedLock lock = createLock("/test/lock-" + runtime.getMethodName(), zkc, true);
        BKLogSegmentWriter writer =
                createLogSegmentWriter(confLocal, 0L, -1L, lock);
        // Use another lock to wait for writer releasing lock
        ZKDistributedLock lock0 = createLock("/test/lock-" + runtime.getMethodName(), zkc0, false);
        Future<ZKDistributedLock> lockFuture0 = lock0.asyncAcquire();
        // add 10 records
        int numRecords = 10;
        List<Future<DLSN>> futureList = new ArrayList<Future<DLSN>>(numRecords);
        for (int i = 0; i < numRecords; i++) {
            futureList.add(writer.asyncWrite(DLMTestUtil.getLogRecordInstance(i)));
        }
        assertEquals("Last tx id should be " + (numRecords - 1),
                numRecords - 1, writer.getLastTxId());
        assertEquals("Last acked tx id should be -1",
                -1L, writer.getLastTxIdAcknowledged());
        assertEquals("Last DLSN should be " + DLSN.InvalidDLSN,
                DLSN.InvalidDLSN, writer.getLastDLSN());
        assertEquals("Position should be " + numRecords,
                numRecords, writer.getPositionWithinLogSegment());

        final CountDownLatch deferLatch = new CountDownLatch(1);
        writer.getFuturePool().apply(new AbstractFunction0<Object>() {
            @Override
            public Object apply() {
                try {
                    deferLatch.await();
                } catch (InterruptedException e) {
                    LOG.warn("Interrupted on deferring completion : ", e);
                }
                return null;
            }
        });

        // transmit the buffered data
        FutureUtils.result(writer.flush());

        // add another 10 records
        List<Future<DLSN>> anotherFutureList = new ArrayList<Future<DLSN>>(numRecords);
        for (int i = numRecords; i < 2 * numRecords; i++) {
            anotherFutureList.add(writer.asyncWrite(DLMTestUtil.getLogRecordInstance(i)));
        }
        assertEquals("Last tx id should become " + (2 * numRecords - 1),
                2 * numRecords - 1, writer.getLastTxId());
        assertEquals("Last acked tx id should become " + (numRecords - 1),
                (long) (numRecords - 1), writer.getLastTxIdAcknowledged());
        assertEquals("Last DLSN should still be " + DLSN.InvalidDLSN,
                DLSN.InvalidDLSN, writer.getLastDLSN());
        assertEquals("Position should become " + (2 * numRecords),
                2 * numRecords, writer.getPositionWithinLogSegment());

        // abort the writer: it waits for outstanding transmits and abort buffered data
        abortWriterAndLock(writer, lock);

        Await.result(lockFuture0);
        lock0.checkOwnership();

        // release defer latch so completion would go through
        deferLatch.countDown();

        List<DLSN> dlsns = Await.result(Future.collect(futureList));
        assertEquals("All first 10 records should be written",
                numRecords, dlsns.size());
        for (int i = 0; i < numRecords; i++) {
            DLSN dlsn = dlsns.get(i);
            assertEquals("Incorrent ledger sequence number",
                    0L, dlsn.getLogSegmentSequenceNo());
            assertEquals("Incorrent entry id",
                    0L, dlsn.getEntryId());
            assertEquals("Inconsistent slot id",
                    i, dlsn.getSlotId());
        }
        for (int i = 0; i < numRecords; i++) {
            try {
                Await.result(anotherFutureList.get(i));
                fail("Should be aborted record " + (numRecords + i) + " with transmit exception");
            } catch (WriteCancelledException wce) {
                // writes should be cancelled.
            }
        }

        assertEquals("Last tx id should still be " + (2 * numRecords - 1),
                2 * numRecords - 1, writer.getLastTxId());
        assertEquals("Last acked tx id should be still " + (numRecords - 1),
                (long) (numRecords - 1), writer.getLastTxIdAcknowledged());
        assertEquals("Last DLSN should become " + futureList.get(futureList.size() - 1),
                dlsns.get(futureList.size() - 1), writer.getLastDLSN());
        assertEquals("Position should become " + 2 * numRecords,
                2 * numRecords, writer.getPositionWithinLogSegment());

        // check only 1 entry were written
        LedgerHandle lh = getLedgerHandle(writer);
        LedgerHandle readLh = openLedgerNoRecovery(lh);
        assertTrue("Ledger " + lh.getId() + " should not be closed", readLh.isClosed());
        assertEquals("Only one entry is written for ledger " + lh.getId(),
                0L, lh.getLastAddPushed());
        assertEquals("Only one entry is written for ledger " + lh.getId(),
                0L, readLh.getLastAddConfirmed());
    }

    /**
     * Log Segment Writer should only update last tx id only for user records.
     */
    @Test(timeout = 60000)
    public void testUpdateLastTxIdForUserRecords() throws Exception {
        DistributedLogConfiguration confLocal = newLocalConf();
        confLocal.setImmediateFlushEnabled(false);
        confLocal.setOutputBufferSize(Integer.MAX_VALUE);
        confLocal.setPeriodicFlushFrequencyMilliSeconds(0);
        ZKDistributedLock lock = createLock("/test/lock-" + runtime.getMethodName(), zkc, true);
        BKLogSegmentWriter writer =
                createLogSegmentWriter(confLocal, 0L, -1L, lock);
        // add 10 records
        int numRecords = 10;
        List<Future<DLSN>> futureList = new ArrayList<Future<DLSN>>(numRecords);
        for (int i = 0; i < numRecords; i++) {
            futureList.add(writer.asyncWrite(DLMTestUtil.getLogRecordInstance(i)));
        }
        LogRecord controlRecord = DLMTestUtil.getLogRecordInstance(9999L);
        controlRecord.setControl();
        futureList.add(writer.asyncWrite(controlRecord));
        assertEquals("Last tx id should be " + (numRecords - 1),
                numRecords - 1, writer.getLastTxId());
        assertEquals("Last DLSN should be " + DLSN.InvalidDLSN,
                DLSN.InvalidDLSN, writer.getLastDLSN());
        assertEquals("Position should be " + numRecords,
                numRecords, writer.getPositionWithinLogSegment());

        // close the writer to flush the output buffer
        closeWriterAndLock(writer, lock);

        List<DLSN> dlsns = Await.result(Future.collect(futureList));
        assertEquals("All 11 records should be written",
                numRecords + 1, dlsns.size());
        for (int i = 0; i < numRecords; i++) {
            DLSN dlsn = dlsns.get(i);
            assertEquals("Incorrent ledger sequence number",
                    0L, dlsn.getLogSegmentSequenceNo());
            assertEquals("Incorrent entry id",
                    0L, dlsn.getEntryId());
            assertEquals("Inconsistent slot id",
                    i, dlsn.getSlotId());
        }
        DLSN dlsn = dlsns.get(numRecords);
        assertEquals("Incorrent ledger sequence number",
                0L, dlsn.getLogSegmentSequenceNo());
        assertEquals("Incorrent entry id",
                1L, dlsn.getEntryId());
        assertEquals("Inconsistent slot id",
                0L, dlsn.getSlotId());

        assertEquals("Last tx id should be " + (numRecords - 1),
                numRecords - 1, writer.getLastTxId());
        assertEquals("Last acked tx id should be " + (numRecords - 1),
                numRecords - 1, writer.getLastTxIdAcknowledged());
        assertEquals("Position should be " + numRecords,
                numRecords, writer.getPositionWithinLogSegment());
        assertEquals("Last DLSN should be " + dlsn,
                dlsns.get(numRecords - 1), writer.getLastDLSN());
    }

    /**
     * Non durable write should fail if writer is closed.
     *
     * @throws Exception
     */
    @Test(timeout = 60000)
    public void testNondurableWriteAfterWriterIsClosed() throws Exception {
        DistributedLogConfiguration confLocal = newLocalConf();
        confLocal.setImmediateFlushEnabled(false);
        confLocal.setOutputBufferSize(Integer.MAX_VALUE);
        confLocal.setPeriodicFlushFrequencyMilliSeconds(0);
        confLocal.setDurableWriteEnabled(false);
        ZKDistributedLock lock = createLock("/test/lock-" + runtime.getMethodName(), zkc, true);
        BKLogSegmentWriter writer =
                createLogSegmentWriter(confLocal, 0L, -1L, lock);

        // close the writer
        closeWriterAndLock(writer, lock);
        FutureUtils.result(writer.asyncClose());

        try {
            Await.result(writer.asyncWrite(DLMTestUtil.getLogRecordInstance(1)));
            fail("Should fail the write if the writer is closed");
        } catch (WriteException we) {
            // expected
        }
    }

    /**
     * Non durable write should fail if writer is marked as end of stream.
     *
     * @throws Exception
     */
    @Test(timeout = 60000)
    public void testNondurableWriteAfterEndOfStream() throws Exception {
        DistributedLogConfiguration confLocal = newLocalConf();
        confLocal.setImmediateFlushEnabled(false);
        confLocal.setOutputBufferSize(Integer.MAX_VALUE);
        confLocal.setPeriodicFlushFrequencyMilliSeconds(0);
        confLocal.setDurableWriteEnabled(false);
        ZKDistributedLock lock = createLock("/test/lock-" + runtime.getMethodName(), zkc, true);
        BKLogSegmentWriter writer =
                createLogSegmentWriter(confLocal, 0L, -1L, lock);

        FutureUtils.result(writer.markEndOfStream());

        try {
            Await.result(writer.asyncWrite(DLMTestUtil.getLogRecordInstance(1)));
            fail("Should fail the write if the writer is marked as end of stream");
        } catch (EndOfStreamException we) {
            // expected
        }

        closeWriterAndLock(writer, lock);
    }

    /**
     * Non durable write should fail if the log segment is fenced.
     *
     * @throws Exception
     */
    @Test(timeout = 60000)
    public void testNondurableWriteAfterLedgerIsFenced() throws Exception {
        DistributedLogConfiguration confLocal = newLocalConf();
        confLocal.setImmediateFlushEnabled(false);
        confLocal.setOutputBufferSize(Integer.MAX_VALUE);
        confLocal.setPeriodicFlushFrequencyMilliSeconds(0);
        confLocal.setDurableWriteEnabled(false);
        ZKDistributedLock lock = createLock("/test/lock-" + runtime.getMethodName(), zkc, true);
        BKLogSegmentWriter writer =
                createLogSegmentWriter(confLocal, 0L, -1L, lock);

        // fence the ledger
        fenceLedger(getLedgerHandle(writer));

        LogRecord record = DLMTestUtil.getLogRecordInstance(1);
        record.setControl();
        try {
            Await.result(writer.asyncWrite(record));
            fail("Should fail the writer if the log segment is already fenced");
        } catch (BKTransmitException bkte) {
            // expected
            assertEquals(BKException.Code.LedgerFencedException, bkte.getBKResultCode());
        }

        try {
            Await.result(writer.asyncWrite(DLMTestUtil.getLogRecordInstance(2)));
            fail("Should fail the writer if the log segment is already fenced");
        } catch (WriteException we) {
            // expected
        }

        abortWriterAndLock(writer, lock);
    }

    /**
     * Non durable write should fail if writer is marked as end of stream.
     *
     * @throws Exception
     */
    @Test(timeout = 60000)
    public void testNondurableWrite() throws Exception {
        DistributedLogConfiguration confLocal = newLocalConf();
        confLocal.setImmediateFlushEnabled(false);
        confLocal.setOutputBufferSize(Integer.MAX_VALUE);
        confLocal.setPeriodicFlushFrequencyMilliSeconds(0);
        confLocal.setDurableWriteEnabled(false);
        ZKDistributedLock lock = createLock("/test/lock-" + runtime.getMethodName(), zkc, true);
        BKLogSegmentWriter writer =
                createLogSegmentWriter(confLocal, 0L, -1L, lock);

        assertEquals(DLSN.InvalidDLSN,
                Await.result(writer.asyncWrite(DLMTestUtil.getLogRecordInstance(2))));
        assertEquals(-1L, ((BKLogSegmentEntryWriter) writer.getEntryWriter())
                .getLedgerHandle().getLastAddPushed());

        closeWriterAndLock(writer, lock);
    }
}