bookkeeper-server/src/test/java/org/apache/bookkeeper/meta/LedgerMetadataCreationTest.java

/*
 *
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 *
 */

package org.apache.bookkeeper.meta;

import static org.junit.Assert.assertTrue;

import java.util.Random;
import java.util.Set;
import java.util.Vector;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.BookKeeper.DigestType;
import org.apache.bookkeeper.meta.zk.ZKMetadataDriverBase;
import org.apache.zookeeper.ZooKeeper;
import org.junit.Assume;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Test the creation of ledger metadata.
 */
public class LedgerMetadataCreationTest extends LedgerManagerTestCase {
    static final Logger LOG = LoggerFactory.getLogger(LedgerMetadataCreationTest.class);

    public LedgerMetadataCreationTest(Class<? extends LedgerManagerFactory> lmFactoryCls) {
        super(lmFactoryCls, 4);
        baseConf.setGcWaitTime(100000);
    }

    @Test
    public void testLedgerCreationAndDeletionWithRandomLedgerIds() throws Exception {
        testExecution(true);
    }

    @Test
    public void testLedgerCreationAndDeletion() throws Exception{
        testExecution(false);
    }

    public void testExecution(boolean randomLedgerId) throws Exception {
        Set<Long> createRequestsLedgerIds = ConcurrentHashMap.newKeySet();
        ConcurrentLinkedDeque<Long> existingLedgerIds = new ConcurrentLinkedDeque<Long>();

        Vector<Long> failedCreates = new Vector<Long>();
        Vector<Long> failedDeletes = new Vector<Long>();
        BookKeeper bookKeeper = new BookKeeper(baseClientConf);

        ExecutorService executor = Executors.newFixedThreadPool(300);
        Random rand = new Random();
        int numberOfOperations = 20000;
        for (int i = 0; i < numberOfOperations; i++) {
            int iteration = i;
            if (rand.nextBoolean() || existingLedgerIds.isEmpty()) {
                executor.submit(() -> {
                    long ledgerId = -1;
                    try {
                        if (randomLedgerId) {
                            do {
                                ledgerId = Math.abs(rand.nextLong());
                                if (!baseClientConf.getLedgerManagerFactoryClass()
                                        .equals(LongHierarchicalLedgerManagerFactory.class)) {
                                    /*
                                     * since LongHierarchicalLedgerManager
                                     * supports ledgerIds of decimal length upto
                                     * 19 digits but other LedgerManagers only
                                     * upto 10 decimals
                                     */
                                    ledgerId %= 9999999999L;
                                }
                            } while (!createRequestsLedgerIds.add(ledgerId));
                        } else {
                            ledgerId = iteration;
                        }
                        bookKeeper.createLedgerAdv(ledgerId, 3, 2, 2, DigestType.CRC32, "passwd".getBytes(), null);
                        existingLedgerIds.add(ledgerId);
                    } catch (Exception e) {
                        LOG.error("Got Exception while creating Ledger with ledgerId " + ledgerId, e);
                        failedCreates.add(ledgerId);
                    }
                });
            } else {
                executor.submit(() -> {
                    Long ledgerId = null;
                    if (rand.nextBoolean()) {
                        ledgerId = existingLedgerIds.pollFirst();
                    } else {
                        ledgerId = existingLedgerIds.pollLast();
                    }
                    if (ledgerId == null) {
                        return;
                    }
                    try {
                        bookKeeper.deleteLedger(ledgerId);
                    } catch (Exception e) {
                        LOG.error("Got Exception while deleting Ledger with ledgerId " + ledgerId, e);
                        failedDeletes.add(ledgerId);
                    }
                });
            }
        }
        executor.shutdown();
        assertTrue("All the ledger create/delete operations should have'been completed",
                executor.awaitTermination(120, TimeUnit.SECONDS));
        assertTrue("There should be no failed creates. But there are " + failedCreates.size() + " failedCreates",
                failedCreates.isEmpty());
        assertTrue("There should be no failed deletes. But there are " + failedDeletes.size() + " failedDeletes",
                failedDeletes.isEmpty());
        bookKeeper.close();
    }

    @Test
    public void testParentNodeDeletion() throws Exception {
        /*
         * run this testcase only for HierarchicalLedgerManager and
         * LongHierarchicalLedgerManager, since we do recursive zNode deletes
         * only for HierarchicalLedgerManager
         */
        Assume.assumeTrue((baseClientConf.getLedgerManagerFactoryClass().equals(HierarchicalLedgerManagerFactory.class)
                || baseClientConf.getLedgerManagerFactoryClass().equals(LongHierarchicalLedgerManagerFactory.class)));

        ZooKeeper zkc = new ZooKeeper(zkUtil.getZooKeeperConnectString(), 10000, null);
        BookKeeper bookKeeper = new BookKeeper(baseClientConf);
        bookKeeper.createLedgerAdv(1, 3, 2, 2, DigestType.CRC32, "passwd".getBytes(), null);
        String ledgersRootPath = ZKMetadataDriverBase.resolveZkLedgersRootPath(baseClientConf);
        String parentZnodePath;
        if (baseClientConf.getLedgerManagerFactoryClass().equals(HierarchicalLedgerManagerFactory.class)) {
            /*
             * in HierarchicalLedgerManager (ledgersRootPath)/00/0000/L0001
             * would be the path of the znode for ledger - 1. So when ledger - 1
             * is deleted, (ledgersRootPath)/00 should also be deleted since
             * there are no other children znodes
             */
            parentZnodePath = ledgersRootPath + "/00";

        } else {
            /*
             * in LongHierarchicalLedgerManager
             * (ledgersRootPath)/000/0000/0000/0000/L0001 would be the path of
             * the znode for ledger - 1. So when ledger - 1 is deleted,
             * (ledgersRootPath)/000 should also be deleted since there are no
             * other children znodes
             */
            parentZnodePath = ledgersRootPath + "/000";
        }
        assertTrue(parentZnodePath + " zNode should exist", null != zkc.exists(parentZnodePath, false));
        bookKeeper.deleteLedger(1);
        assertTrue(parentZnodePath + " zNode should not exist anymore", null == zkc.exists(parentZnodePath, false));
        bookKeeper.close();
        zkc.close();
    }
}