/*
 * 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 java.nio.charset.StandardCharsets.UTF_8;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;

import io.netty.buffer.ByteBuf;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.bookkeeper.bookie.Bookie;
import org.apache.bookkeeper.bookie.BookieException;
import org.apache.bookkeeper.bookie.TestBookieImpl;
import org.apache.bookkeeper.client.AsyncCallback.AddCallback;
import org.apache.bookkeeper.client.BookKeeper.DigestType;
import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.net.BookieId;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.WriteCallback;
import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
import org.apache.bookkeeper.test.TestCallbacks.AddCallbackFuture;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * This class tests the ledger close logic.
 */
@SuppressWarnings("deprecation")
public class LedgerCloseTest extends BookKeeperClusterTestCase {

    private static final Logger LOG = LoggerFactory.getLogger(LedgerCloseTest.class);

    static final int READ_TIMEOUT = 1;

    final DigestType digestType;

    public LedgerCloseTest() {
        super(6);
        this.digestType = DigestType.CRC32;
        // set timeout to a large value which disable it.
        baseClientConf.setReadTimeout(99999);
        baseConf.setGcWaitTime(999999);
    }

    @Test
    public void testLedgerCloseWithConsistentLength() throws Exception {
        ClientConfiguration conf = new ClientConfiguration();
        conf.setMetadataServiceUri(zkUtil.getMetadataServiceUri());
        conf.setReadTimeout(1);

        BookKeeper bkc = new BookKeeper(conf);
        LedgerHandle lh = bkc.createLedger(6, 3, DigestType.CRC32, new byte[] {});
        final CountDownLatch latch = new CountDownLatch(1);
        stopBKCluster();
        final AtomicInteger i = new AtomicInteger(0xdeadbeef);
        AsyncCallback.AddCallback cb = new AsyncCallback.AddCallback() {
            @Override
            public void addComplete(int rc, LedgerHandle lh, long entryId, Object ctx) {
                i.set(rc);
                latch.countDown();
            }
        };
        lh.asyncAddEntry("Test Entry".getBytes(), cb, null);
        latch.await();
        assertEquals(i.get(), BKException.Code.NotEnoughBookiesException);
        assertEquals(0, lh.getLength());
        assertEquals(LedgerHandle.INVALID_ENTRY_ID, lh.getLastAddConfirmed());
        startBKCluster(zkUtil.getMetadataServiceUri());
        LedgerHandle newLh = bkc.openLedger(lh.getId(), DigestType.CRC32, new byte[] {});
        assertEquals(0, newLh.getLength());
        assertEquals(LedgerHandle.INVALID_ENTRY_ID, newLh.getLastAddConfirmed());
    }

    @Test
    public void testLedgerCloseDuringUnrecoverableErrors() throws Exception {
        int numEntries = 3;
        LedgerHandle lh = bkc.createLedger(3, 3, 3, digestType, "".getBytes());
        verifyMetadataConsistency(numEntries, lh);
    }

    @Test
    public void testLedgerCheckerShouldnotSelectInvalidLastFragments() throws Exception {
        int numEntries = 10;
        LedgerHandle lh = bkc.createLedger(3, 3, 3, digestType, "".getBytes());
        // Add some entries before bookie failures
        for (int i = 0; i < numEntries; i++) {
            lh.addEntry("data".getBytes());
        }
        numEntries = 4; // add n*ensemleSize+1 entries async after bookies
                        // failed.
        verifyMetadataConsistency(numEntries, lh);

        LedgerChecker checker = new LedgerChecker(bkc);
        CheckerCallback cb = new CheckerCallback();
        checker.checkLedger(lh, cb);
        Set<LedgerFragment> result = cb.waitAndGetResult();
        assertEquals("No fragments should be selected", 0, result.size());
    }

    class CheckerCallback implements GenericCallback<Set<LedgerFragment>> {
        private Set<LedgerFragment> result = null;
        private CountDownLatch latch = new CountDownLatch(1);

        public void operationComplete(int rc, Set<LedgerFragment> result) {
            this.result = result;
            latch.countDown();
        }

        Set<LedgerFragment> waitAndGetResult() throws InterruptedException {
            latch.await();
            return result;
        }
    }

    private void verifyMetadataConsistency(int numEntries, LedgerHandle lh)
            throws Exception {
        final CountDownLatch addDoneLatch = new CountDownLatch(1);
        final CountDownLatch deadIOLatch = new CountDownLatch(1);
        final CountDownLatch recoverDoneLatch = new CountDownLatch(1);
        final CountDownLatch failedLatch = new CountDownLatch(1);
        // kill first bookie to replace with a unauthorize bookie
        BookieId bookie = lh.getCurrentEnsemble().get(0);
        ServerConfiguration conf = killBookie(bookie);
        // replace a unauthorize bookie
        startUnauthorizedBookie(conf, addDoneLatch);
        // kill second bookie to replace with a dead bookie
        bookie = lh.getCurrentEnsemble().get(1);
        conf = killBookie(bookie);
        // replace a slow dead bookie
        startDeadBookie(conf, deadIOLatch);

        // tried to add entries
        for (int i = 0; i < numEntries; i++) {
            lh.asyncAddEntry("data".getBytes(), new AddCallback() {
                @Override
                public void addComplete(int rc, LedgerHandle lh, long entryId, Object ctx) {
                    if (BKException.Code.OK != rc) {
                        failedLatch.countDown();
                        deadIOLatch.countDown();
                    }
                    if (0 == entryId) {
                        try {
                            recoverDoneLatch.await();
                        } catch (InterruptedException ie) {
                            Thread.currentThread().interrupt();
                        }
                    }
                }
            }, null);
        }
        // add finished
        addDoneLatch.countDown();
        // wait until entries failed due to UnauthorizedAccessException
        failedLatch.await();
        // simulate the ownership of this ledger is transfer to another host
        LOG.info("Recover ledger {}.", lh.getId());
        ClientConfiguration newConf = new ClientConfiguration();
        newConf.addConfiguration(baseClientConf);
        BookKeeper newBkc = new BookKeeperTestClient(newConf.setReadTimeout(1));
        LedgerHandle recoveredLh = newBkc.openLedger(lh.getId(), digestType, "".getBytes());
        LOG.info("Recover ledger {} done.", lh.getId());
        recoverDoneLatch.countDown();
        // wait a bit until add operations failed from second bookie due to IOException
        TimeUnit.SECONDS.sleep(5);
        // open the ledger again to make sure we ge the right last confirmed.
        LedgerHandle newLh = newBkc.openLedger(lh.getId(), digestType, "".getBytes());
        assertEquals("Metadata should be consistent across different opened ledgers",
                recoveredLh.getLastAddConfirmed(), newLh.getLastAddConfirmed());
    }

    private void startUnauthorizedBookie(ServerConfiguration conf, final CountDownLatch latch)
            throws Exception {
        Bookie sBookie = new TestBookieImpl(conf) {
            @Override
            public void addEntry(ByteBuf entry, boolean ackBeforeSync, WriteCallback cb, Object ctx, byte[] masterKey)
                    throws IOException, BookieException {
                try {
                    latch.await();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                throw BookieException.create(BookieException.Code.UnauthorizedAccessException);
            }

            @Override
            public void recoveryAddEntry(ByteBuf entry, WriteCallback cb, Object ctx, byte[] masterKey)
                    throws IOException, BookieException {
                throw new IOException("Dead bookie for recovery adds.");
            }
        };
        startAndAddBookie(conf, sBookie);
    }

    // simulate slow adds, then become normal when recover,
    // so no ensemble change when recovering ledger on this bookie.
    private void startDeadBookie(ServerConfiguration conf, final CountDownLatch latch) throws Exception {
        Bookie dBookie = new TestBookieImpl(conf) {
            @Override
            public void addEntry(ByteBuf entry, boolean ackBeforeSync, WriteCallback cb, Object ctx, byte[] masterKey)
                    throws IOException, BookieException {
                try {
                    latch.await();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                // simulate slow adds.
                throw new IOException("Dead bookie");
            }
        };
        startAndAddBookie(conf, dBookie);
    }

    @Test
    public void testAllWritesAreCompletedOnClosedLedger() throws Exception {
        for (int i = 0; i < 100; i++) {
            LOG.info("Iteration {}", i);

            List<AddCallbackFuture> futures = new ArrayList<AddCallbackFuture>();
            LedgerHandle w = bkc.createLedger(DigestType.CRC32, new byte[0]);
            AddCallbackFuture f = new AddCallbackFuture(0L);
            w.asyncAddEntry("foobar".getBytes(UTF_8), f, null);
            f.get();

            LedgerHandle r = bkc.openLedger(w.getId(), DigestType.CRC32, new byte[0]);
            for (int j = 0; j < 100; j++) {
                AddCallbackFuture f1 = new AddCallbackFuture(1L + j);
                w.asyncAddEntry("foobar".getBytes(), f1, null);
                futures.add(f1);
            }

            for (AddCallbackFuture f2: futures) {
                try {
                    f2.get(10, TimeUnit.SECONDS);
                } catch (ExecutionException ee) {
                    // we don't care about errors
                } catch (TimeoutException te) {
                    LOG.error("Error on waiting completing entry {} : ", f2.getExpectedEntryId(), te);
                    fail("Should succeed on waiting completing entry " + f2.getExpectedEntryId());
                }
            }
        }
    }
}
