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

import java.io.IOException;
import java.net.URI;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import com.google.common.base.Optional;
import com.google.common.collect.Lists;
import com.twitter.distributedlog.annotations.DistributedLogAnnotations;
import com.twitter.distributedlog.config.ConcurrentBaseConfiguration;
import com.twitter.distributedlog.config.ConcurrentConstConfiguration;
import com.twitter.distributedlog.config.DynamicDistributedLogConfiguration;
import com.twitter.distributedlog.exceptions.BKTransmitException;
import com.twitter.distributedlog.exceptions.LockingException;
import com.twitter.distributedlog.io.CompressionCodec;
import com.twitter.distributedlog.util.Utils;
import com.twitter.util.Promise;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.BookKeeperAccessor;
import org.apache.bookkeeper.client.LedgerHandle;
import org.apache.bookkeeper.client.LedgerMetadata;
import org.apache.bookkeeper.feature.FixedValueFeature;
import org.apache.bookkeeper.stats.NullStatsLogger;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestName;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Stopwatch;
import com.twitter.distributedlog.exceptions.DLIllegalStateException;
import com.twitter.distributedlog.exceptions.EndOfStreamException;
import com.twitter.distributedlog.exceptions.IdleReaderException;
import com.twitter.distributedlog.exceptions.LogRecordTooLongException;
import com.twitter.distributedlog.exceptions.OverCapacityException;
import com.twitter.distributedlog.exceptions.ReadCancelledException;
import com.twitter.distributedlog.exceptions.WriteException;
import com.twitter.distributedlog.lock.DistributedLock;
import com.twitter.distributedlog.namespace.DistributedLogNamespace;
import com.twitter.distributedlog.namespace.DistributedLogNamespaceBuilder;
import com.twitter.distributedlog.util.FailpointUtils;
import com.twitter.distributedlog.util.FutureUtils;
import com.twitter.distributedlog.util.SimplePermitLimiter;
import com.twitter.util.Await;
import com.twitter.util.Duration;
import com.twitter.util.Future;
import com.twitter.util.FutureEventListener;

import junit.framework.Assert;
import static com.google.common.base.Charsets.UTF_8;
import static com.twitter.distributedlog.DLMTestUtil.validateFutureFailed;
import static com.twitter.distributedlog.LogRecord.MAX_LOGRECORD_SIZE;
import static org.junit.Assert.*;

public class TestAsyncReaderWriter extends TestDistributedLogBase {
    static final Logger LOG = LoggerFactory.getLogger(TestAsyncReaderWriter.class);

    protected DistributedLogConfiguration testConf;

    public TestAsyncReaderWriter() {
        this.testConf = new DistributedLogConfiguration();
        this.testConf.loadConf(conf);
        this.testConf.setReaderIdleErrorThresholdMillis(1200000);
    }

    @Rule
    public TestName runtime = new TestName();

    /**
     * Test writing control records to writers: writers should be able to write control records, and
     * the readers should skip control records while reading.
     */
    @Test(timeout = 60000)
    public void testWriteControlRecord() throws Exception {
        String name = runtime.getMethodName();
        DistributedLogConfiguration confLocal = new DistributedLogConfiguration();
        confLocal.loadConf(testConf);
        confLocal.setOutputBufferSize(1024);
        DistributedLogManager dlm = createNewDLM(confLocal, name);

        // Write 3 log segments. For each log segments, write one control record and nine user records.
        int txid = 1;
        for (long i = 0; i < 3; i++) {
            final long currentLogSegmentSeqNo = i + 1;
            BKAsyncLogWriter writer = (BKAsyncLogWriter)(dlm.startAsyncLogSegmentNonPartitioned());
            DLSN dlsn = Await.result(writer.writeControlRecord(new LogRecord(txid++, "control".getBytes(UTF_8))));
            assertEquals(currentLogSegmentSeqNo, dlsn.getLogSegmentSequenceNo());
            assertEquals(0, dlsn.getEntryId());
            assertEquals(0, dlsn.getSlotId());
            for (long j = 1; j < 10; j++) {
                final LogRecord record = DLMTestUtil.getLargeLogRecordInstance(txid++);
                Await.result(writer.write(record));
            }
            writer.closeAndComplete();
        }
        dlm.close();

        // Read all the written data: It should skip control records and only return user records.
        DistributedLogManager readDlm = createNewDLM(confLocal, name);
        LogReader reader = readDlm.getInputStream(1);

        long numTrans = 0;
        long expectedTxId = 2;
        LogRecord record = reader.readNext(false);
        while (null != record) {
            DLMTestUtil.verifyLargeLogRecord(record);
            numTrans++;
            assertEquals(expectedTxId, record.getTransactionId());
            if (expectedTxId % 10 == 0) {
                expectedTxId += 2;
            } else {
                ++expectedTxId;
            }
            record = reader.readNext(false);
        }
        assertEquals(3 * 9, numTrans);
        assertEquals(3 * 9, readDlm.getLogRecordCount());
        readDlm.close();
    }

    @Test(timeout = 60000)
    public void testAsyncWritePendingWritesAbortedWhenLedgerRollTriggerFails() throws Exception {
        String name = runtime.getMethodName();
        DistributedLogConfiguration confLocal = new DistributedLogConfiguration();
        confLocal.loadConf(testConf);
        confLocal.setOutputBufferSize(1024);
        confLocal.setMaxLogSegmentBytes(1024);
        confLocal.setLogSegmentRollingIntervalMinutes(0);
        DistributedLogManager dlm = createNewDLM(confLocal, name);
        BKAsyncLogWriter writer = (BKAsyncLogWriter)(dlm.startAsyncLogSegmentNonPartitioned());

        // Write one record larger than max seg size. Ledger doesn't roll until next write.
        int txid = 1;
        LogRecord record = DLMTestUtil.getLogRecordInstance(txid++, 2048);
        Future<DLSN> result = writer.write(record);
        DLSN dlsn = Await.result(result, Duration.fromSeconds(10));
        assertEquals(1, dlsn.getLogSegmentSequenceNo());

        record = DLMTestUtil.getLogRecordInstance(txid++, MAX_LOGRECORD_SIZE + 1);
        result = writer.write(record);
        validateFutureFailed(result, LogRecordTooLongException.class);

        record = DLMTestUtil.getLogRecordInstance(txid++, MAX_LOGRECORD_SIZE + 1);
        result = writer.write(record);
        validateFutureFailed(result, WriteException.class);

        record = DLMTestUtil.getLogRecordInstance(txid++, MAX_LOGRECORD_SIZE + 1);
        result = writer.write(record);
        validateFutureFailed(result, WriteException.class);

        writer.closeAndComplete();
        dlm.close();
    }

    /**
     * Test Case: Simple Async Writes. Writes 30 records. They should be written correctly.
     * @throws Exception
     */
    @Test(timeout = 60000)
    public void testSimpleAsyncWrite() throws Exception {
        String name = runtime.getMethodName();
        DistributedLogConfiguration confLocal = new DistributedLogConfiguration();
        confLocal.loadConf(testConf);
        confLocal.setOutputBufferSize(1024);

        int numLogSegments = 3;
        int numRecordsPerLogSegment = 10;

        DistributedLogManager dlm = createNewDLM(confLocal, name);

        final CountDownLatch syncLatch = new CountDownLatch(numLogSegments * numRecordsPerLogSegment);
        final AtomicBoolean errorsFound = new AtomicBoolean(false);
        final AtomicReference<DLSN> maxDLSN = new AtomicReference<DLSN>(DLSN.InvalidDLSN);
        int txid = 1;
        for (long i = 0; i < numLogSegments; i++) {
            final long currentLogSegmentSeqNo = i + 1;
            BKAsyncLogWriter writer = (BKAsyncLogWriter)(dlm.startAsyncLogSegmentNonPartitioned());
            for (long j = 0; j < numRecordsPerLogSegment; j++) {
                final long currentEntryId = j;
                final LogRecord record = DLMTestUtil.getLargeLogRecordInstance(txid++);
                Future<DLSN> dlsnFuture = writer.write(record);
                dlsnFuture.addEventListener(new FutureEventListener<DLSN>() {
                    @Override
                    public void onSuccess(DLSN value) {
                        if(value.getLogSegmentSequenceNo() != currentLogSegmentSeqNo) {
                            LOG.debug("LogSegmentSequenceNumber: {}, Expected {}", value.getLogSegmentSequenceNo(), currentLogSegmentSeqNo);
                            errorsFound.set(true);
                        }

                        if(value.getEntryId() != currentEntryId) {
                            LOG.debug("EntryId: {}, Expected {}", value.getEntryId(), currentEntryId);
                            errorsFound.set(true);
                        }

                        if (value.compareTo(maxDLSN.get()) > 0) {
                            maxDLSN.set(value);
                        }

                        syncLatch.countDown();
                        LOG.debug("SyncLatch: {}", syncLatch.getCount());
                    }
                    @Override
                    public void onFailure(Throwable cause) {
                        LOG.error("Encountered exception on writing record {} in log segment {}", currentEntryId, currentLogSegmentSeqNo);
                        errorsFound.set(true);
                    }
                });
            }
            writer.closeAndComplete();
        }

        syncLatch.await();
        assertFalse("Should not encounter any errors for async writes", errorsFound.get());

        LogRecordWithDLSN last = dlm.getLastLogRecord();
        assertEquals("Last DLSN" + last.getDlsn() + " isn't the maximum DLSN " + maxDLSN.get(),
                last.getDlsn(), maxDLSN.get());
        assertEquals(last.getDlsn(), dlm.getLastDLSN());
        assertEquals(last.getDlsn(), Await.result(dlm.getLastDLSNAsync()));
        DLMTestUtil.verifyLargeLogRecord(last);

        dlm.close();
    }

    /**
     * Write records into <i>numLogSegments</i> log segments. Each log segment has <i>numRecordsPerLogSegment</i> records.
     *
     * @param dlm
     *          distributedlog manager
     * @param numLogSegments
     *          number of log segments
     * @param numRecordsPerLogSegment
     *          number records per log segment
     * @param startTxId
     *          start tx id
     * @return next tx id
     */
    private static long writeRecords(DistributedLogManager dlm,
                                     int numLogSegments,
                                     int numRecordsPerLogSegment,
                                     long startTxId,
                                     boolean emptyRecord) throws IOException {
        long txid = startTxId;
        for (long i = 0; i < numLogSegments; i++) {
            BKSyncLogWriter writer = (BKSyncLogWriter)dlm.startLogSegmentNonPartitioned();
            for (long j = 1; j <= numRecordsPerLogSegment; j++) {
                if (emptyRecord) {
                    writer.write(DLMTestUtil.getEmptyLogRecordInstance(txid++));
                } else {
                    writer.write(DLMTestUtil.getLargeLogRecordInstance(txid++));
                }
            }
            writer.closeAndComplete();
        }
        return txid;
    }

    /**
     * Write <code>numRecords</code> records to the log, starting with <code>startTxId</code>.
     * It flushes every <code>flushPerNumRecords</code> records.
     *
     * @param dlm
     *          distributedlog manager
     * @param numRecords
     *          num records to write
     * @param startTxId
     *          start tx id
     * @param flushPerNumRecords
     *          number records to flush
     * @return next tx id
     * @throws IOException
     */
    private static long writeLogSegment(DistributedLogManager dlm,
                                        int numRecords,
                                        long startTxId,
                                        int flushPerNumRecords,
                                        boolean emptyRecord) throws IOException {
        long txid = startTxId;
        LogWriter writer = dlm.startLogSegmentNonPartitioned();
        for (long j = 1; j <= numRecords; j++) {
            if (emptyRecord) {
                writer.write(DLMTestUtil.getEmptyLogRecordInstance(txid++));
            } else {
                writer.write(DLMTestUtil.getLargeLogRecordInstance(txid++));
            }
            if (j % flushPerNumRecords == 0 ) {
                writer.setReadyToFlush();
                writer.flushAndSync();
            }
        }
        writer.setReadyToFlush();
        writer.flushAndSync();
        writer.close();
        return txid;
    }

    private static void readNext(final AsyncLogReader reader,
                                 final DLSN startPosition,
                                 final long startSequenceId,
                                 final boolean monotonic,
                                 final CountDownLatch syncLatch,
                                 final CountDownLatch completionLatch,
                                 final AtomicBoolean errorsFound) {
        Future<LogRecordWithDLSN> record = reader.readNext();
        record.addEventListener(new FutureEventListener<LogRecordWithDLSN>() {
            @Override
            public void onSuccess(LogRecordWithDLSN value) {
                try {
                    if (monotonic) {
                        assertEquals(startSequenceId, value.getSequenceId());
                    } else {
                        assertTrue(value.getSequenceId() < 0);
                        assertTrue(value.getSequenceId() > startSequenceId);
                    }
                    LOG.debug("Recevied record {} from {}", value.getDlsn(), reader.getStreamName());
                    assertTrue(!value.isControl());
                    assertTrue(value.getDlsn().getSlotId() == 0);
                    assertTrue(value.getDlsn().compareTo(startPosition) >= 0);
                    DLMTestUtil.verifyLargeLogRecord(value);
                } catch (Exception exc) {
                    LOG.debug("Exception Encountered when verifying log record {} : ", value.getDlsn(), exc);
                    errorsFound.set(true);
                    completionLatch.countDown();
                    return;
                }
                syncLatch.countDown();
                if (syncLatch.getCount() <= 0) {
                    completionLatch.countDown();
                } else {
                    TestAsyncReaderWriter.readNext(
                            reader,
                            value.getDlsn().getNextDLSN(),
                            monotonic ? value.getSequenceId() + 1 : value.getSequenceId(),
                            monotonic,
                            syncLatch,
                            completionLatch,
                            errorsFound);
                }
            }
            @Override
            public void onFailure(Throwable cause) {
                LOG.debug("Encountered Exception on reading {}", reader.getStreamName(), cause);
                errorsFound.set(true);
                completionLatch.countDown();
            }
        });
    }

    void simpleAsyncReadTest(String name, DistributedLogConfiguration confLocal) throws Exception {
        confLocal.setOutputBufferSize(1024);
        confLocal.setReadAheadWaitTime(10);
        confLocal.setReadAheadBatchSize(10);
        DistributedLogManager dlm = createNewDLM(confLocal, name);

        int numLogSegments = 3;
        int numRecordsPerLogSegment = 10;

        // Write 30 records: 3 log segments, 10 records per log segment
        long txid = 1L;
        txid = writeRecords(dlm, numLogSegments, numRecordsPerLogSegment, txid, false);
        // Write another log segment with 5 records and flush every 2 records
        txid = writeLogSegment(dlm, 5, txid, 2, false);

        final AsyncLogReader reader = dlm.getAsyncLogReader(DLSN.InvalidDLSN);
        final CountDownLatch syncLatch = new CountDownLatch((int) (txid - 1));
        final CountDownLatch completionLatch = new CountDownLatch(1);
        final AtomicBoolean errorsFound = new AtomicBoolean(false);

        boolean monotonic = LogSegmentMetadata.supportsSequenceId(confLocal.getDLLedgerMetadataLayoutVersion());
        TestAsyncReaderWriter.readNext(
                reader,
                DLSN.InvalidDLSN,
                monotonic ? 0L : Long.MIN_VALUE,
                monotonic,
                syncLatch,
                completionLatch,
                errorsFound);

        completionLatch.await();
        assertFalse("Errors encountered on reading records", errorsFound.get());
        syncLatch.await();

        Utils.close(reader);
        dlm.close();
    }

    @Test(timeout = 60000)
    public void testSimpleAsyncRead() throws Exception {
        String name = runtime.getMethodName();
        DistributedLogConfiguration confLocal = new DistributedLogConfiguration();
        confLocal.loadConf(testConf);
        simpleAsyncReadTest(name, confLocal);
    }

    @Test(timeout = 60000)
    public void testSimpleAsyncReadWriteWithMonitoredFuturePool() throws Exception {
        String name = runtime.getMethodName();
        DistributedLogConfiguration confLocal = new DistributedLogConfiguration();
        confLocal.loadConf(testConf);
        confLocal.setTaskExecutionWarnTimeMicros(1000);
        confLocal.setEnableTaskExecutionStats(true);
        simpleAsyncReadTest(name, confLocal);
    }

    @Test(timeout = 60000)
    public void testBulkAsyncRead() throws Exception {
        String name = "distrlog-bulkasyncread";
        DistributedLogConfiguration confLocal = new DistributedLogConfiguration();
        confLocal.loadConf(conf);
        confLocal.setOutputBufferSize(0);
        confLocal.setImmediateFlushEnabled(true);
        confLocal.setReadAheadWaitTime(10);
        confLocal.setReadAheadMaxRecords(10000);
        confLocal.setReadAheadBatchSize(10);

        int numLogSegments = 3;
        int numRecordsPerLogSegment = 20;

        DistributedLogManager dlm = createNewDLM(confLocal, name);
        writeRecords(dlm, numLogSegments, numRecordsPerLogSegment, 1L, false);

        final AsyncLogReader reader = dlm.getAsyncLogReader(DLSN.InitialDLSN);
        int expectedTxID = 1;
        int numReads = 0;
        while (expectedTxID <= numLogSegments * numRecordsPerLogSegment) {
            if (expectedTxID == numLogSegments * numRecordsPerLogSegment) {
                break;
            }
            List<LogRecordWithDLSN> records = Await.result(reader.readBulk(20));
            LOG.info("Bulk read {} entries.", records.size());

            assertTrue(records.size() >= 1);
            for (LogRecordWithDLSN record : records) {
                assertEquals(expectedTxID, record.getTransactionId());
                ++expectedTxID;
            }
            ++numReads;
        }

        // we expect bulk read works
        assertTrue(numReads < 60);

        Utils.close(reader);
        dlm.close();
    }

    @Test(timeout = 60000)
    public void testBulkAsyncReadWithWriteBatch() throws Exception {
        String name = "distrlog-bulkasyncread-with-writebatch";
        DistributedLogConfiguration confLocal = new DistributedLogConfiguration();
        confLocal.loadConf(conf);
        confLocal.setOutputBufferSize(1024000);
        confLocal.setReadAheadWaitTime(10);
        confLocal.setReadAheadMaxRecords(10000);
        confLocal.setReadAheadBatchSize(10);

        DistributedLogManager dlm = createNewDLM(confLocal, name);

        int numLogSegments = 3;
        int numRecordsPerLogSegment = 20;

        writeRecords(dlm, numLogSegments, numRecordsPerLogSegment, 1L, false);

        final AsyncLogReader reader = dlm.getAsyncLogReader(DLSN.InitialDLSN);
        int expectedTxID = 1;
        for (long i = 0; i < 3; i++) {
            // since we batched 20 entries into single bookkeeper entry
            // we should be able to read 20 entries as a batch.
            List<LogRecordWithDLSN> records = Await.result(reader.readBulk(20));
            assertEquals(20, records.size());
            for (LogRecordWithDLSN record : records) {
                assertEquals(expectedTxID, record.getTransactionId());
                ++expectedTxID;
            }
        }

        Utils.close(reader);
        dlm.close();
    }

    @Test(timeout = 60000)
    public void testAsyncReadEmptyRecords() throws Exception {
        String name = "distrlog-simpleasyncreadempty";
        DistributedLogConfiguration confLocal = new DistributedLogConfiguration();
        confLocal.loadConf(testConf);
        confLocal.setOutputBufferSize(0);
        confLocal.setReadAheadWaitTime(10);
        confLocal.setReadAheadBatchSize(10);
        DistributedLogManager dlm = createNewDLM(confLocal, name);

        int numLogSegments = 3;
        int numRecordsPerLogSegment = 10;

        long txid = 1L;
        // write 3 log segments, 10 records per log segment
        txid = writeRecords(dlm, numLogSegments, numRecordsPerLogSegment, txid, true);
        // write another log segment with 5 records and flush every 2 records
        txid = writeLogSegment(dlm, 5, txid, 2, true);

        AsyncLogReader asyncReader = dlm.getAsyncLogReader(DLSN.InvalidDLSN);
        assertEquals("Expected stream name = " + name + " but " + asyncReader.getStreamName() + " found",
                name, asyncReader.getStreamName());
        long numTrans = 0;
        DLSN lastDLSN = DLSN.InvalidDLSN;
        LogRecordWithDLSN record = Await.result(asyncReader.readNext());
        while (null != record) {
            DLMTestUtil.verifyEmptyLogRecord(record);
            assertEquals(0, record.getDlsn().getSlotId());
            assertTrue(record.getDlsn().compareTo(lastDLSN) > 0);
            lastDLSN = record.getDlsn();
            numTrans++;
            if (numTrans >= (txid - 1)) {
                break;
            }
            record = Await.result(asyncReader.readNext());
        }
        assertEquals((txid - 1), numTrans);
        Utils.close(asyncReader);
        dlm.close();
    }

    /**
     * Test Async Read by positioning to a given position in the log
     * @throws Exception
     */
    @Test(timeout = 60000)
    public void testSimpleAsyncReadPosition() throws Exception {
        String name = runtime.getMethodName();
        DistributedLogConfiguration confLocal = new DistributedLogConfiguration();
        confLocal.loadConf(testConf);
        confLocal.setOutputBufferSize(1024);
        confLocal.setReadAheadWaitTime(10);
        confLocal.setReadAheadBatchSize(10);
        DistributedLogManager dlm = createNewDLM(confLocal, name);

        int numLogSegments = 3;
        int numRecordsPerLogSegment = 10;

        long txid = 1L;
        // write 3 log segments, 10 records per log segment
        txid = writeRecords(dlm, numLogSegments, numRecordsPerLogSegment, txid, false);
        // write another log segment with 5 records
        txid = writeLogSegment(dlm, 5, txid, Integer.MAX_VALUE, false);

        final CountDownLatch syncLatch = new CountDownLatch((int)(txid - 14));
        final CountDownLatch doneLatch = new CountDownLatch(1);
        final AtomicBoolean errorsFound = new AtomicBoolean(false);
        final AsyncLogReader reader = dlm.getAsyncLogReader(new DLSN(2, 2, 4));
        assertEquals(name, reader.getStreamName());

        boolean monotonic = LogSegmentMetadata.supportsSequenceId(confLocal.getDLLedgerMetadataLayoutVersion());
        TestAsyncReaderWriter.readNext(
                reader,
                new DLSN(2, 3, 0),
                monotonic ? 13L : Long.MIN_VALUE,
                monotonic,
                syncLatch,
                doneLatch,
                errorsFound);

        doneLatch.await();
        assertFalse("Errors found on reading records", errorsFound.get());
        syncLatch.await();

        Utils.close(reader);
        dlm.close();
    }

    /**
     * Test write/read entries when immediate flush is disabled.
     * @throws Exception
     */
    @Test(timeout = 60000)
    public void testSimpleAsyncReadWrite() throws Exception {
        testSimpleAsyncReadWriteInternal(runtime.getMethodName(), false);
    }

    /**
     * Test write/read entries when immediate flush is enabled.
     *
     * @throws Exception
     */
    @Test(timeout = 60000)
    public void testSimpleAsyncReadWriteImmediateFlush() throws Exception {
        testSimpleAsyncReadWriteInternal(runtime.getMethodName(), true);
    }

    /**
     * Test if entries written using log segment metadata that doesn't support enveloping
     * can be read correctly by a reader supporting both.
     *
     * NOTE: An older reader cannot read enveloped entry, so we don't have a test case covering
     *       the other scenario.
     *
     * @throws Exception
     */
    @Test(timeout = 60000)
    public void testNoEnvelopeWriterEnvelopeReader() throws Exception {
        testSimpleAsyncReadWriteInternal(runtime.getMethodName(), true,
                LogSegmentMetadata.LogSegmentMetadataVersion.VERSION_V4_ENVELOPED_ENTRIES.value - 1);
    }

    static class WriteFutureEventListener implements FutureEventListener<DLSN> {
        private final LogRecord record;
        private final long currentLogSegmentSeqNo;
        private final long currentEntryId;
        private final CountDownLatch syncLatch;
        private final AtomicBoolean errorsFound;
        private final boolean verifyEntryId;

        WriteFutureEventListener(LogRecord record,
                                 long currentLogSegmentSeqNo,
                                 long currentEntryId,
                                 CountDownLatch syncLatch,
                                 AtomicBoolean errorsFound,
                                 boolean verifyEntryId) {
            this.record = record;
            this.currentLogSegmentSeqNo = currentLogSegmentSeqNo;
            this.currentEntryId = currentEntryId;
            this.syncLatch = syncLatch;
            this.errorsFound = errorsFound;
            this.verifyEntryId = verifyEntryId;
        }

        /**
         * Invoked if the computation completes successfully
         */
        @Override
        public void onSuccess(DLSN value) {
            if(value.getLogSegmentSequenceNo() != currentLogSegmentSeqNo) {
                LOG.error("Ledger Seq No: {}, Expected: {}", value.getLogSegmentSequenceNo(), currentLogSegmentSeqNo);
                errorsFound.set(true);
            }

            if(verifyEntryId && value.getEntryId() != currentEntryId) {
                LOG.error("EntryId: {}, Expected: {}", value.getEntryId(), currentEntryId);
                errorsFound.set(true);
            }
            syncLatch.countDown();
        }

        /**
         * Invoked if the computation completes unsuccessfully
         */
        @Override
        public void onFailure(Throwable cause) {
            LOG.error("Encountered failures on writing record as (lid = {}, eid = {}) :",
                    new Object[]{currentLogSegmentSeqNo, currentEntryId, cause});
            errorsFound.set(true);
            syncLatch.countDown();
        }
    }

    void testSimpleAsyncReadWriteInternal(String name, boolean immediateFlush)
            throws Exception {
        testSimpleAsyncReadWriteInternal(name, immediateFlush,
                                         LogSegmentMetadata.LEDGER_METADATA_CURRENT_LAYOUT_VERSION);
    }

    void testSimpleAsyncReadWriteInternal(String name, boolean immediateFlush,
                                          int logSegmentVersion) throws Exception {
        DistributedLogConfiguration confLocal = new DistributedLogConfiguration();
        confLocal.loadConf(testConf);
        confLocal.setReadAheadWaitTime(10);
        confLocal.setReadAheadBatchSize(10);
        confLocal.setOutputBufferSize(1024);
        confLocal.setDLLedgerMetadataLayoutVersion(logSegmentVersion);
        confLocal.setImmediateFlushEnabled(immediateFlush);
        DistributedLogManager dlm = createNewDLM(confLocal, name);

        int numLogSegments = 3;
        int numRecordsPerLogSegment = 10;

        final CountDownLatch readLatch = new CountDownLatch(numLogSegments * numRecordsPerLogSegment);
        final CountDownLatch readDoneLatch = new CountDownLatch(1);
        final AtomicBoolean readErrors = new AtomicBoolean(false);
        final CountDownLatch writeLatch = new CountDownLatch(numLogSegments * numRecordsPerLogSegment);
        final AtomicBoolean writeErrors = new AtomicBoolean(false);
        final AsyncLogReader reader = dlm.getAsyncLogReader(DLSN.InvalidDLSN);
        assertEquals(name, reader.getStreamName());

        int txid = 1;
        for (long i = 0; i < 3; i++) {
            final long currentLogSegmentSeqNo = i + 1;
            BKAsyncLogWriter writer = (BKAsyncLogWriter)(dlm.startAsyncLogSegmentNonPartitioned());
            for (long j = 0; j < 10; j++) {
                final long currentEntryId = j;
                final LogRecord record = DLMTestUtil.getLargeLogRecordInstance(txid++);
                Future<DLSN> dlsnFuture = writer.write(record);
                dlsnFuture.addEventListener(new WriteFutureEventListener(
                        record, currentLogSegmentSeqNo, currentEntryId, writeLatch, writeErrors, true));
                if (i == 0 && j == 0) {
                    boolean monotonic = LogSegmentMetadata.supportsSequenceId(logSegmentVersion);
                    TestAsyncReaderWriter.readNext(
                            reader,
                            DLSN.InvalidDLSN,
                            monotonic ? 0L : Long.MIN_VALUE,
                            monotonic,
                            readLatch,
                            readDoneLatch,
                            readErrors);
                }
            }
            writer.closeAndComplete();
        }

        writeLatch.await();
        assertFalse("All writes should succeed", writeErrors.get());

        readDoneLatch.await();
        assertFalse("All reads should succeed", readErrors.get());
        readLatch.await();

        Utils.close(reader);
        dlm.close();
    }


    /**
     * Test Case: starting reading when the streams don't exist.
     *
     * @throws Exception
     */
    @Test(timeout = 60000)
    public void testSimpleAsyncReadWriteStartEmpty() throws Exception {
        String name = runtime.getMethodName();
        DistributedLogConfiguration confLocal = new DistributedLogConfiguration();
        confLocal.loadConf(testConf);
        confLocal.setReadAheadWaitTime(10);
        confLocal.setReadAheadBatchSize(10);
        confLocal.setOutputBufferSize(1024);

        int numLogSegments = 3;
        int numRecordsPerLogSegment = 10;

        DistributedLogManager dlm = createNewDLM(confLocal, name);

        final CountDownLatch readerReadyLatch = new CountDownLatch(1);
        final CountDownLatch readerDoneLatch = new CountDownLatch(1);
        final CountDownLatch readerSyncLatch = new CountDownLatch(numLogSegments * numRecordsPerLogSegment);

        final TestReader reader = new TestReader(
                "test-reader",
                dlm,
                DLSN.InitialDLSN,
                false,
                0,
                readerReadyLatch,
                readerSyncLatch,
                readerDoneLatch);

