distributedlog-core/src/test/java/org/apache/distributedlog/impl/metadata/TestZKLogStreamMetadataStore.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.distributedlog.impl.metadata;

import static com.google.common.base.Charsets.UTF_8;
import static org.apache.distributedlog.DistributedLogConstants.EMPTY_BYTES;
import static org.apache.distributedlog.impl.metadata.ZKLogStreamMetadataStore.*;
import static org.apache.distributedlog.metadata.LogMetadata.*;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyBoolean;
import static org.mockito.Matchers.anyObject;
import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

import com.google.common.collect.Lists;
import java.net.URI;
import java.util.Collections;
import java.util.List;
import org.apache.bookkeeper.stats.NullStatsLogger;
import org.apache.bookkeeper.util.ZkUtils;
import org.apache.bookkeeper.versioning.LongVersion;
import org.apache.bookkeeper.versioning.Versioned;
import org.apache.distributedlog.DLMTestUtil;
import org.apache.distributedlog.DistributedLogConfiguration;
import org.apache.distributedlog.DistributedLogConstants;
import org.apache.distributedlog.LogSegmentMetadata;
import org.apache.distributedlog.TestZooKeeperClientBuilder;
import org.apache.distributedlog.ZooKeeperClient;
import org.apache.distributedlog.ZooKeeperClusterTestCase;
import org.apache.distributedlog.api.MetadataAccessor;
import org.apache.distributedlog.api.namespace.Namespace;
import org.apache.distributedlog.api.namespace.NamespaceBuilder;
import org.apache.distributedlog.common.concurrent.FutureUtils;
import org.apache.distributedlog.exceptions.LockingException;
import org.apache.distributedlog.exceptions.LogExistsException;
import org.apache.distributedlog.exceptions.LogNotFoundException;
import org.apache.distributedlog.exceptions.ZKException;
import org.apache.distributedlog.metadata.DLMetadata;
import org.apache.distributedlog.metadata.LogMetadata;
import org.apache.distributedlog.metadata.LogMetadataForWriter;
import org.apache.distributedlog.util.DLUtils;
import org.apache.distributedlog.util.OrderedScheduler;
import org.apache.distributedlog.util.Utils;
import org.apache.zookeeper.AsyncCallback.Children2Callback;
import org.apache.zookeeper.AsyncCallback.StatCallback;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.Transaction;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestName;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Test {@link ZKLogStreamMetadataStore}.
 */
public class TestZKLogStreamMetadataStore extends ZooKeeperClusterTestCase {

    private static final Logger logger = LoggerFactory.getLogger(TestZKLogStreamMetadataStore.class);

    private static final  int sessionTimeoutMs = 30000;

    @Rule
    public TestName testName = new TestName();

    private ZooKeeperClient zkc;
    private URI uri;
    private OrderedScheduler scheduler;
    private ZKLogStreamMetadataStore metadataStore;

    private static void createLog(ZooKeeperClient zk, URI uri, String logName, String logIdentifier)
            throws Exception {
        final String logRootPath = getLogRootPath(uri, logName, logIdentifier);
        final String logSegmentsPath = logRootPath + LOGSEGMENTS_PATH;
        final String maxTxIdPath = logRootPath + MAX_TXID_PATH;
        final String lockPath = logRootPath + LOCK_PATH;
        final String readLockPath = logRootPath + READ_LOCK_PATH;
        final String versionPath = logRootPath + VERSION_PATH;
        final String allocationPath = logRootPath + ALLOCATION_PATH;

        Utils.zkCreateFullPathOptimistic(zk, logRootPath, new byte[0],
                zk.getDefaultACL(), CreateMode.PERSISTENT);
        Transaction txn = zk.get().transaction();
        txn.create(logSegmentsPath, DLUtils.serializeLogSegmentSequenceNumber(
                        DistributedLogConstants.UNASSIGNED_LOGSEGMENT_SEQNO),
                zk.getDefaultACL(), CreateMode.PERSISTENT);
        txn.create(maxTxIdPath, DLUtils.serializeTransactionId(0L),
                zk.getDefaultACL(), CreateMode.PERSISTENT);
        txn.create(lockPath, EMPTY_BYTES,
                zk.getDefaultACL(), CreateMode.PERSISTENT);
        txn.create(readLockPath, EMPTY_BYTES,
                zk.getDefaultACL(), CreateMode.PERSISTENT);
        txn.create(versionPath, intToBytes(LAYOUT_VERSION),
                zk.getDefaultACL(), CreateMode.PERSISTENT);
        txn.create(allocationPath, EMPTY_BYTES,
                zk.getDefaultACL(), CreateMode.PERSISTENT);
        txn.commit();
    }

    private static void createLog(ZooKeeperClient zk,
                                  URI uri,
                                  String logName,
                                  String logIdentifier,
                                  int numSegments)
            throws Exception {
        final String logRootPath = getLogRootPath(uri, logName, logIdentifier);
        final String logSegmentsPath = logRootPath + LOGSEGMENTS_PATH;
        final String maxTxIdPath = logRootPath + MAX_TXID_PATH;
        final String lockPath = logRootPath + LOCK_PATH;
        final String readLockPath = logRootPath + READ_LOCK_PATH;
        final String versionPath = logRootPath + VERSION_PATH;
        final String allocationPath = logRootPath + ALLOCATION_PATH;

        Utils.zkCreateFullPathOptimistic(zk, logRootPath, new byte[0],
                zk.getDefaultACL(), CreateMode.PERSISTENT);
        Transaction txn = zk.get().transaction();
        txn.create(logSegmentsPath, DLUtils.serializeLogSegmentSequenceNumber(
                        DistributedLogConstants.UNASSIGNED_LOGSEGMENT_SEQNO),
                zk.getDefaultACL(), CreateMode.PERSISTENT);
        txn.create(maxTxIdPath, DLUtils.serializeTransactionId(0L),
                zk.getDefaultACL(), CreateMode.PERSISTENT);
        txn.create(lockPath, EMPTY_BYTES,
                zk.getDefaultACL(), CreateMode.PERSISTENT);
        txn.create(readLockPath, EMPTY_BYTES,
                zk.getDefaultACL(), CreateMode.PERSISTENT);
        txn.create(versionPath, intToBytes(LAYOUT_VERSION),
                zk.getDefaultACL(), CreateMode.PERSISTENT);
        txn.create(allocationPath, EMPTY_BYTES,
                zk.getDefaultACL(), CreateMode.PERSISTENT);

        for (int i = 0; i < numSegments; i++) {
            LogSegmentMetadata segment = DLMTestUtil.completedLogSegment(
                logSegmentsPath,
                i + 1L,
                1L + i * 1000L,
                (i + 1) * 1000L,
                1000,
                i + 1L,
                999L,
                0L);
            txn.create(
                segment.getZkPath(),
                segment.getFinalisedData().getBytes(UTF_8),
                zk.getDefaultACL(),
                CreateMode.PERSISTENT);
        }

        txn.commit();
    }

    @Before
    public void setup() throws Exception {
        zkc = TestZooKeeperClientBuilder.newBuilder()
                .name("zkc")
                .uri(DLMTestUtil.createDLMURI(zkPort, "/"))
                .sessionTimeoutMs(sessionTimeoutMs)
                .build();
        uri = DLMTestUtil.createDLMURI(zkPort, "");
        try {
            ZkUtils.createFullPathOptimistic(
                    zkc.get(),
                    uri.getPath(),
                    new byte[0],
                    ZooDefs.Ids.OPEN_ACL_UNSAFE,
                    CreateMode.PERSISTENT);
        } catch (KeeperException.NodeExistsException nee) {
            logger.debug("The namespace uri already exists.");
        }
        scheduler = OrderedScheduler.newBuilder()
            .name("test-scheduler")
            .corePoolSize(1)
            .build();
        metadataStore = new ZKLogStreamMetadataStore(
            "test-logstream-metadata-store",
            new DistributedLogConfiguration(),
            zkc,
            scheduler,
            NullStatsLogger.INSTANCE);
    }

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

    @Test(timeout = 60000)
    public void testCheckLogMetadataPathsWithAllocator() throws Exception {
        String logRootPath = "/" + testName.getMethodName();
        List<Versioned<byte[]>> metadatas =
                Utils.ioResult(checkLogMetadataPaths(
                        zkc.get(), logRootPath, true));
        assertEquals("Should have 8 paths",
                8, metadatas.size());
        for (Versioned<byte[]> path : metadatas.subList(2, metadatas.size())) {
            assertNull(path.getValue());
            assertNull(path.getVersion());
        }
    }

    @Test(timeout = 60000)
    public void testCheckLogMetadataPathsWithoutAllocator() throws Exception {
        String logRootPath = "/" + testName.getMethodName();
        List<Versioned<byte[]>> metadatas =
                Utils.ioResult(checkLogMetadataPaths(
                        zkc.get(), logRootPath, false));
        assertEquals("Should have 7 paths",
                7, metadatas.size());
        for (Versioned<byte[]> path : metadatas.subList(2, metadatas.size())) {
            assertNull(path.getValue());
            assertNull(path.getVersion());
        }
    }

    private void testCreateLogMetadataWithMissingPaths(URI uri,
                                                       String logName,
                                                       String logIdentifier,
                                                       List<String> pathsToDelete,
                                                       boolean ownAllocator,
                                                       boolean createLogFirst)
            throws Exception {
        if (createLogFirst) {
            createLog(zkc, uri, logName, logIdentifier);
        }
        // delete a path
        for (String path : pathsToDelete) {
            zkc.get().delete(path, -1);
        }

        LogMetadataForWriter logMetadata =
                Utils.ioResult(getLog(uri, logName, logIdentifier, zkc, ownAllocator, true));

        final String logRootPath = getLogRootPath(uri, logName, logIdentifier);

        List<Versioned<byte[]>> metadatas =
                Utils.ioResult(checkLogMetadataPaths(zkc.get(), logRootPath, ownAllocator));

        if (ownAllocator) {
            assertEquals("Should have 8 paths : ownAllocator = " + ownAllocator,
                    8, metadatas.size());
        } else {
            assertEquals("Should have 7 paths : ownAllocator = " + ownAllocator,
                    7, metadatas.size());
        }

        for (Versioned<byte[]> metadata : metadatas) {
            assertTrue(pathExists(metadata));
            assertTrue(((LongVersion) metadata.getVersion()).getLongVersion() >= 0L);
        }

        Versioned<byte[]> logSegmentsData = logMetadata.getMaxLSSNData();

        assertEquals(DistributedLogConstants.UNASSIGNED_LOGSEGMENT_SEQNO,
                DLUtils.deserializeLogSegmentSequenceNumber(logSegmentsData.getValue()));

        Versioned<byte[]> maxTxIdData = logMetadata.getMaxTxIdData();

        assertEquals(0L, DLUtils.deserializeTransactionId(maxTxIdData.getValue()));

        if (ownAllocator) {
            Versioned<byte[]> allocationData = logMetadata.getAllocationData();
            assertEquals(0, allocationData.getValue().length);
        }
    }

    @Test(timeout = 60000)
    public void testCreateLogMetadataMissingLogSegmentsPath() throws Exception {
        String logName = testName.getMethodName();
        String logIdentifier = "<default>";
        String logRootPath = getLogRootPath(uri, logName, logIdentifier);
        List<String> pathsToDelete = Lists.newArrayList(
                logRootPath + LOGSEGMENTS_PATH);
        testCreateLogMetadataWithMissingPaths(uri, logName, logIdentifier, pathsToDelete, false, true);
    }

    @Test(timeout = 60000)
    public void testCreateLogMetadataMissingMaxTxIdPath() throws Exception {
        String logName = testName.getMethodName();
        String logIdentifier = "<default>";
        String logRootPath = getLogRootPath(uri, logName, logIdentifier);
        List<String> pathsToDelete = Lists.newArrayList(
                logRootPath + MAX_TXID_PATH);
        testCreateLogMetadataWithMissingPaths(uri, logName, logIdentifier, pathsToDelete, false, true);
    }

    @Test(timeout = 60000)
    public void testCreateLogMetadataMissingLockPath() throws Exception {
        String logName = testName.getMethodName();
        String logIdentifier = "<default>";
        String logRootPath = getLogRootPath(uri, logName, logIdentifier);
        List<String> pathsToDelete = Lists.newArrayList(
                logRootPath + LOCK_PATH);
        testCreateLogMetadataWithMissingPaths(uri, logName, logIdentifier, pathsToDelete, false, true);
    }

    @Test(timeout = 60000)
    public void testCreateLogMetadataMissingReadLockPath() throws Exception {
        String logName = testName.getMethodName();
        String logIdentifier = "<default>";
        String logRootPath = getLogRootPath(uri, logName, logIdentifier);
        List<String> pathsToDelete = Lists.newArrayList(
                logRootPath + READ_LOCK_PATH);
        testCreateLogMetadataWithMissingPaths(uri, logName, logIdentifier, pathsToDelete, false, true);
    }

    @Test(timeout = 60000)
    public void testCreateLogMetadataMissingVersionPath() throws Exception {
        String logName = testName.getMethodName();
        String logIdentifier = "<default>";
        String logRootPath = getLogRootPath(uri, logName, logIdentifier);
        List<String> pathsToDelete = Lists.newArrayList(
                logRootPath + VERSION_PATH);
        testCreateLogMetadataWithMissingPaths(uri, logName, logIdentifier, pathsToDelete, false, true);
    }

    @Test(timeout = 60000)
    public void testCreateLogMetadataMissingAllocatorPath() throws Exception {
        URI uri = DLMTestUtil.createDLMURI(zkPort, "");
        String logName = testName.getMethodName();
        String logIdentifier = "<default>";
        String logRootPath = getLogRootPath(uri, logName, logIdentifier);
        List<String> pathsToDelete = Lists.newArrayList(
                logRootPath + ALLOCATION_PATH);
        testCreateLogMetadataWithMissingPaths(uri, logName, logIdentifier, pathsToDelete, true, true);
    }

    @Test(timeout = 60000)
    public void testCreateLogMetadataMissingAllPath() throws Exception {
        String logName = testName.getMethodName();
        String logIdentifier = "<default>";
        String logRootPath = getLogRootPath(uri, logName, logIdentifier);
        List<String> pathsToDelete = Lists.newArrayList(
                logRootPath + LOGSEGMENTS_PATH,
                logRootPath + MAX_TXID_PATH,
                logRootPath + LOCK_PATH,
                logRootPath + READ_LOCK_PATH,
                logRootPath + VERSION_PATH,
                logRootPath + ALLOCATION_PATH);
        testCreateLogMetadataWithMissingPaths(uri, logName, logIdentifier, pathsToDelete, true, true);
    }

    @Test(timeout = 60000)
    public void testCreateLogMetadataOnExistedLog() throws Exception {
        String logName = testName.getMethodName();
        String logIdentifier = "<default>";
        List<String> pathsToDelete = Lists.newArrayList();
        testCreateLogMetadataWithMissingPaths(uri, logName, logIdentifier, pathsToDelete, true, true);
    }

    @Test(timeout = 60000)
    public void testCreateLogMetadata() throws Exception {
        String logName = testName.getMethodName();
        String logIdentifier = "<default>";
        List<String> pathsToDelete = Lists.newArrayList();

        testCreateLogMetadataWithMissingPaths(uri, logName, logIdentifier, pathsToDelete, true, false);
    }

    @Test(timeout = 60000, expected = LogNotFoundException.class)
    public void testCreateLogMetadataWithCreateIfNotExistsSetToFalse() throws Exception {
        String logName = testName.getMethodName();
        String logIdentifier = "<default>";
        Utils.ioResult(getLog(uri, logName, logIdentifier, zkc, true, false));
    }

    @Test(timeout = 60000)
    public void testCreateLogMetadataWithCustomMetadata() throws Exception {
        String logName = testName.getMethodName();
        String logIdentifier = "<default>";
        List<String> pathsToDelete = Lists.newArrayList();

        DLMetadata.create(new BKDLConfig(zkServers, "/ledgers")).update(uri);

        Namespace namespace = NamespaceBuilder.newBuilder()
            .conf(new DistributedLogConfiguration())
            .uri(uri)
            .build();

        MetadataAccessor accessor = namespace.getNamespaceDriver().getMetadataAccessor(logName);
        accessor.createOrUpdateMetadata(logName.getBytes("UTF-8"));
        accessor.close();

        testCreateLogMetadataWithMissingPaths(uri, logName, logIdentifier, pathsToDelete, true, false);
    }

    @Test(timeout = 60000, expected = LogNotFoundException.class)
    public void testGetLogSegmentsLogNotFound() throws Exception {
        String logName = testName.getMethodName();
        String logIdentifier = "<default>";

        String logSegmentsPath = LogMetadata.getLogSegmentsPath(uri, logName, logIdentifier);
        FutureUtils.result(getLogSegments(zkc, logSegmentsPath));
    }

    @Test(timeout = 60000)
    public void testGetLogSegmentsZKExceptions() throws Exception {
        String logName = testName.getMethodName();
        String logIdentifier = "<default>";

        ZooKeeper mockZk = mock(ZooKeeper.class);
        ZooKeeperClient mockZkc = mock(ZooKeeperClient.class);
        when(mockZkc.get()).thenReturn(mockZk);
        doAnswer(invocationOnMock -> {
            String path = (String) invocationOnMock.getArguments()[0];
            Children2Callback callback = (Children2Callback) invocationOnMock.getArguments()[2];
            callback.processResult(Code.BADVERSION.intValue(), path, null, null, null);
            return null;
        }).when(mockZk).getChildren(anyString(), anyBoolean(), any(Children2Callback.class), anyObject());

        String logSegmentsPath = LogMetadata.getLogSegmentsPath(uri, logName, logIdentifier);
        try {
            FutureUtils.result(getLogSegments(mockZkc, logSegmentsPath));
            fail("Should fail to get log segments when encountering zk exceptions");
        } catch (ZKException zke) {
            assertEquals(Code.BADVERSION, zke.getKeeperExceptionCode());
        }
    }

    @Test(timeout = 60000)
    public void testGetLogSegments() throws Exception {
        String logName = testName.getMethodName();
        String logIdentifier = "<default>";

        // create log
        createLog(
            zkc,
            uri,
            logName,
            logIdentifier,
            5);

        List<LogSegmentMetadata> segments = FutureUtils.result(
            getLogSegments(zkc, LogMetadata.getLogSegmentsPath(uri, logName, logIdentifier)));
        assertEquals(5, segments.size());
        for (int i = 0; i < 5; i++) {
            assertEquals(1L + i, segments.get(i).getLogSegmentSequenceNumber());
        }
    }

    @Test(timeout = 60000)
    public void testGetMissingPathsRecursive() throws Exception {
        List<String> missingPaths = FutureUtils.result(
            getMissingPaths(zkc, uri, "path/to/log"));

        assertEquals(
            Lists.newArrayList(
                uri.getPath() + "/path/to/log",
                uri.getPath() + "/path/to",
                uri.getPath() + "/path"
            ),
            missingPaths);
    }

    @Test(timeout = 60000)
    public void testGetMissingPathsRecursive2() throws Exception {
        String path = uri.getPath() + "/path/to/log";
        ZkUtils.createFullPathOptimistic(
            zkc.get(), path, EMPTY_BYTES, zkc.getDefaultACL(), CreateMode.PERSISTENT);

        List<String> missingPaths = FutureUtils.result(
            getMissingPaths(zkc, uri, "path/to/log"));

        assertEquals(
            Collections.emptyList(),
            missingPaths);
    }

    @Test(timeout = 60000)
    public void testGetMissingPathsFailure() throws Exception {
        ZooKeeper mockZk = mock(ZooKeeper.class);
        ZooKeeperClient mockZkc = mock(ZooKeeperClient.class);
        when(mockZkc.get()).thenReturn(mockZk);
        doAnswer(invocationOnMock -> {
            String path = (String) invocationOnMock.getArguments()[0];
            StatCallback callback = (StatCallback) invocationOnMock.getArguments()[2];
            callback.processResult(Code.BADVERSION.intValue(), path, null, null);
            return null;
        }).when(mockZk).exists(anyString(), anyBoolean(), any(StatCallback.class), anyObject());

        try {
            FutureUtils.result(getMissingPaths(mockZkc, uri, "path/to/log"));
            fail("Should fail on getting missing paths on zookeeper exceptions.");
        } catch (ZKException zke) {
            assertEquals(Code.BADVERSION, zke.getKeeperExceptionCode());
        }
    }

    @Test(timeout = 60000)
    public void testRenameLog() throws Exception {
        String logName = testName.getMethodName();
        String logIdentifier = "<default>";
        int numSegments = 5;

        createLog(
            zkc,
            uri,
            logName,
            logIdentifier,
            numSegments);

        String newLogName = "path/to/new/" + logName;
        FutureUtils.result(metadataStore.renameLog(uri, logName, newLogName));
    }

    @Test(timeout = 60000, expected = LogExistsException.class)
    public void testRenameLogExists() throws Exception {
        String logName = testName.getMethodName();
        String logIdentifier = "<default>";
        int numSegments = 5;
        createLog(
            zkc,
            uri,
            logName,
            logIdentifier,
            numSegments);

        String newLogName = "path/to/new/" + logName;
        createLog(
            zkc,
            uri,
            newLogName,
            logIdentifier,
            3);

        FutureUtils.result(metadataStore.renameLog(uri, logName, newLogName));
    }

    @Test(timeout = 60000, expected = LockingException.class)
    public void testRenameLockedLog() throws Exception {
        String logName = testName.getMethodName();
        String logIdentifier = "<default>";
        int numSegments = 5;
        createLog(
            zkc,
            uri,
            logName,
            logIdentifier,
            numSegments);

        // create a lock
        String logRootPath = getLogRootPath(uri, logName, logIdentifier);
        String lockPath = logRootPath + LOCK_PATH;
        zkc.get().create(lockPath + "/test", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

        String newLogName = "path/to/new/" + logName;
        FutureUtils.result(metadataStore.renameLog(uri, logName, newLogName));
    }

}