bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/AbstractZkLedgerManagerFactory.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.apache.bookkeeper.meta.AbstractZkLedgerManager.isLeadgerIdGeneratorZnode;
import static org.apache.bookkeeper.meta.AbstractZkLedgerManager.isSpecialZnode;
import java.io.IOException;
import java.net.URI;
import java.util.List;
import java.util.Objects;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.common.util.ReflectionUtils;
import org.apache.bookkeeper.conf.AbstractConfiguration;
import org.apache.bookkeeper.meta.LayoutManager.LedgerLayoutExistsException;
import org.apache.bookkeeper.meta.zk.ZKMetadataDriverBase;
import org.apache.commons.configuration.ConfigurationException;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZKUtil;
import org.apache.zookeeper.ZooKeeper;

/**
 * Abstract ledger manager factory based on zookeeper, which provides common
 * methods such as format and validateAndNukeExistingCluster.
 */
@Slf4j
public abstract class AbstractZkLedgerManagerFactory implements LedgerManagerFactory {

    protected ZooKeeper zk;

    @SuppressWarnings("deprecation")
    @Override
    public void format(AbstractConfiguration<?> conf, LayoutManager layoutManager)
            throws InterruptedException, KeeperException, IOException {
        try (AbstractZkLedgerManager ledgerManager = (AbstractZkLedgerManager) newLedgerManager()) {
            String ledgersRootPath = ZKMetadataDriverBase.resolveZkLedgersRootPath(conf);
            List<String> children = zk.getChildren(ledgersRootPath, false);
            for (String child : children) {
                boolean lParentNode = !isSpecialZnode(child) && ledgerManager.isLedgerParentNode(child);
                boolean lIdGenerator = isLeadgerIdGeneratorZnode(child);

                if (lParentNode || lIdGenerator) {
                    ZKUtil.deleteRecursive(zk, ledgersRootPath + "/" + child);
                }
            }
        }

        Class<? extends LedgerManagerFactory> factoryClass;
        try {
            factoryClass = conf.getLedgerManagerFactoryClass();
        } catch (ConfigurationException e) {
            throw new IOException("Failed to get ledger manager factory class from configuration : ", e);
        }

        layoutManager.deleteLedgerLayout();
        // Create new layout information again.
        createNewLMFactory(conf, layoutManager, factoryClass);
    }

    @Override
    public boolean validateAndNukeExistingCluster(AbstractConfiguration<?> conf, LayoutManager layoutManager)
            throws InterruptedException, KeeperException, IOException {
        String zkLedgersRootPath = ZKMetadataDriverBase.resolveZkLedgersRootPath(conf);
        String zkServers = ZKMetadataDriverBase.resolveZkServers(conf);
        AbstractZkLedgerManager zkLedgerManager = (AbstractZkLedgerManager) newLedgerManager();

        /*
         * before proceeding with nuking existing cluster, make sure there
         * are no unexpected znodes under ledgersRootPath
         */
        List<String> ledgersRootPathChildrenList = zk.getChildren(zkLedgersRootPath, false);
        for (String ledgersRootPathChildren : ledgersRootPathChildrenList) {
            if ((!AbstractZkLedgerManager.isSpecialZnode(ledgersRootPathChildren))
                    && (!zkLedgerManager.isLedgerParentNode(ledgersRootPathChildren))) {
                log.error("Found unexpected znode : {} under ledgersRootPath : {} so exiting nuke operation",
                        ledgersRootPathChildren, zkLedgersRootPath);
                return false;
            }
        }

        // formatting ledgermanager deletes ledger znodes
        format(conf, layoutManager);

        // now delete all the special nodes recursively
        ledgersRootPathChildrenList = zk.getChildren(zkLedgersRootPath, false);
        for (String ledgersRootPathChildren : ledgersRootPathChildrenList) {
            if (AbstractZkLedgerManager.isSpecialZnode(ledgersRootPathChildren)) {
                ZKUtil.deleteRecursive(zk, zkLedgersRootPath + "/" + ledgersRootPathChildren);
            } else {
                log.error("Found unexpected znode : {} under ledgersRootPath : {} so exiting nuke operation",
                        ledgersRootPathChildren, zkLedgersRootPath);
                return false;
            }
        }

        // finally deleting the ledgers rootpath
        zk.delete(zkLedgersRootPath, -1);

        log.info("Successfully nuked existing cluster, ZKServers: {} ledger root path: {}",
                zkServers, zkLedgersRootPath);
        return true;
    }

    // v1 layout
    static final int V1 = 1;

    /**
     * Create new Ledger Manager Factory.
     *
     * @param conf
     *          Configuration Object.
     * @param layoutManager
     *          layout manager
     * @return new ledger manager factory
     * @throws IOException
     */
    @SuppressWarnings("deprecation")
    public static LedgerManagerFactory newLedgerManagerFactory(
        final AbstractConfiguration<?> conf, LayoutManager layoutManager)
            throws IOException, InterruptedException {
        String metadataServiceUriStr;
        try {
            metadataServiceUriStr = conf.getMetadataServiceUri();
        } catch (ConfigurationException e) {
            log.error("Failed to retrieve metadata service uri from configuration", e);
            throw new IOException(
                "Failed to retrieve metadata service uri from configuration", e);
        }

        Class<? extends LedgerManagerFactory> factoryClass;
        String ledgerRootPath;
        // `metadataServiceUri` can be null when constructing bookkeeper client using an external zookeeper client.
        if (null == metadataServiceUriStr) { //
            try {
                factoryClass = conf.getLedgerManagerFactoryClass();
            } catch (ConfigurationException e) {
                log.error("Failed to get ledger manager factory class when using an external zookeeper client", e);
                throw new IOException(
                    "Failed to get ledger manager factory class when using an external zookeeper client", e);
            }
            ledgerRootPath = conf.getZkLedgersRootPath();
        } else {
            URI metadataServiceUri = URI.create(metadataServiceUriStr);
            factoryClass = ZKMetadataDriverBase.resolveLedgerManagerFactory(metadataServiceUri);
            ledgerRootPath = metadataServiceUri.getPath();
        }

        if (null == ledgerRootPath || ledgerRootPath.length() == 0) {
            throw new IOException("Empty Ledger Root Path.");
        }

        // if layoutManager is null, return the default ledger manager
        if (layoutManager == null) {
            return new FlatLedgerManagerFactory()
                   .initialize(conf, null, FlatLedgerManagerFactory.CUR_VERSION);
        }

        LedgerManagerFactory lmFactory;

        // check that the configured ledger manager is
        // compatible with the existing layout
        LedgerLayout layout = layoutManager.readLedgerLayout();

        if (layout == null) { // no existing layout
            lmFactory = createNewLMFactory(conf, layoutManager, factoryClass);
            return lmFactory
                    .initialize(conf, layoutManager, lmFactory.getCurrentVersion());
        }
        if (log.isDebugEnabled()) {
            log.debug("read ledger layout {}", layout);
        }

        // there is existing layout, we need to look into the layout.
        // handle pre V2 layout
        if (layout.getLayoutFormatVersion() <= V1) {
            // pre V2 layout we use type of ledger manager
            String lmType = conf.getLedgerManagerType();
            if (lmType != null && !layout.getManagerFactoryClass().equals(lmType)) {
                throw new IOException("Configured layout " + lmType
                        + " does not match existing layout "  + layout.getManagerFactoryClass());
            }

            // create the ledger manager
            if (FlatLedgerManagerFactory.NAME.equals(layout.getManagerFactoryClass())) {
                lmFactory = new FlatLedgerManagerFactory();
            } else if (HierarchicalLedgerManagerFactory.NAME.equals(layout.getManagerFactoryClass())) {
                lmFactory = new HierarchicalLedgerManagerFactory();
            } else {
                throw new IOException("Unknown ledger manager type: " + lmType);
            }
            return lmFactory.initialize(conf, layoutManager, layout.getManagerVersion());
        }

        // handle V2 layout case
        if (factoryClass != null
                && !isSameLedgerManagerFactory(
                    conf, layout.getManagerFactoryClass(), factoryClass.getName())
                && conf.getProperty(AbstractConfiguration.LEDGER_MANAGER_FACTORY_DISABLE_CLASS_CHECK) == null) {
                // Disable should ONLY happen during compatibility testing.

            throw new IOException("Configured layout " + factoryClass.getName()
                                + " does not match existing layout "  + layout.getManagerFactoryClass());
        }
        if (factoryClass == null) {
            // no factory specified in configuration
            try {
                Class<?> theCls = Class.forName(layout.getManagerFactoryClass());
                if (!LedgerManagerFactory.class.isAssignableFrom(theCls)) {
                    throw new IOException("Wrong ledger manager factory " + layout.getManagerFactoryClass());
                }
                factoryClass = theCls.asSubclass(LedgerManagerFactory.class);
            } catch (ClassNotFoundException | IOException e) {
                factoryClass = attemptToResolveShadedLedgerManagerFactory(
                    conf,
                    layout.getManagerFactoryClass(),
                    e);
            }
        }
        // instantiate a factory
        lmFactory = ReflectionUtils.newInstance(factoryClass);
        return lmFactory.initialize(conf, layoutManager, layout.getManagerVersion());
    }

    private static String normalizedLedgerManagerFactoryClassName(String factoryClass,
                                                                  String shadedClassPrefix,
                                                                  boolean isShadedClassAllowed) {
        if (isShadedClassAllowed) {
            if (null == factoryClass || null == shadedClassPrefix) {
                return factoryClass;
            } else {
                return factoryClass.replace(shadedClassPrefix, "");
            }
        } else {
            return factoryClass;
        }
    }

    private static boolean isSameLedgerManagerFactory(
            AbstractConfiguration<?> conf, String leftFactoryClass, String rightFactoryClass) {
        leftFactoryClass = normalizedLedgerManagerFactoryClassName(
            leftFactoryClass,
            conf.getShadedLedgerManagerFactoryClassPrefix(),
            conf.isShadedLedgerManagerFactoryClassAllowed());
        rightFactoryClass = normalizedLedgerManagerFactoryClassName(
            rightFactoryClass,
            conf.getShadedLedgerManagerFactoryClassPrefix(),
            conf.isShadedLedgerManagerFactoryClassAllowed());
        return Objects.equals(leftFactoryClass, rightFactoryClass);
    }

    private static Class<? extends LedgerManagerFactory> attemptToResolveShadedLedgerManagerFactory(
            AbstractConfiguration<?> conf, String lmfClassName, Throwable cause) throws IOException {
        if (conf.isShadedLedgerManagerFactoryClassAllowed()) {
            String shadedPrefix = conf.getShadedLedgerManagerFactoryClassPrefix();
            log.warn("Failed to instantiate ledger manager factory {}, trying its shaded class {}{}",
                lmfClassName, shadedPrefix, lmfClassName);
            // try shading class
            try {
                return resolveShadedLedgerManagerFactory(lmfClassName, shadedPrefix);
            } catch (ClassNotFoundException cnfe) {
                throw new IOException("Failed to instantiate ledger manager factory "
                    + lmfClassName + " and its shaded class " + shadedPrefix + lmfClassName, cnfe);
            }
        } else {
            throw new IOException("Failed to instantiate ledger manager factory "
                + lmfClassName, cause);
        }
    }

    private static Class<? extends LedgerManagerFactory> resolveShadedLedgerManagerFactory(String lmfClassName,
                                                                                           String shadedClassPrefix)
            throws ClassNotFoundException, IOException {
        if (null == lmfClassName) {
            return null;
        } else {
            // this is to address the issue when using the dlog shaded jar
            Class<?> theCls = Class.forName(shadedClassPrefix + lmfClassName);
            if (!LedgerManagerFactory.class.isAssignableFrom(theCls)) {
                throw new IOException("Wrong shaded ledger manager factory : " + shadedClassPrefix + lmfClassName);
            }
            return theCls.asSubclass(LedgerManagerFactory.class);
        }
    }

    /**
     * Creates the new layout and stores in zookeeper and returns the
     * LedgerManagerFactory instance.
     */
    @SuppressWarnings("deprecation")
    protected static LedgerManagerFactory createNewLMFactory(
            final AbstractConfiguration conf, final LayoutManager layoutManager,
            Class<? extends LedgerManagerFactory> factoryClass)
            throws IOException, InterruptedException {

        LedgerManagerFactory lmFactory;
        LedgerLayout layout;
        // use default ledger manager factory if no one provided
        if (factoryClass == null) {
            // for backward compatibility, check manager type
            String lmType = conf.getLedgerManagerType();
            if (lmType == null) {
                factoryClass = HierarchicalLedgerManagerFactory.class;
            } else {
                if (FlatLedgerManagerFactory.NAME.equals(lmType)) {
                    factoryClass = FlatLedgerManagerFactory.class;
                } else if (HierarchicalLedgerManagerFactory.NAME.equals(lmType)) {
                    factoryClass = HierarchicalLedgerManagerFactory.class;
                } else if (LongHierarchicalLedgerManagerFactory.NAME.equals(lmType)) {
                    factoryClass = LongHierarchicalLedgerManagerFactory.class;
                } else {
                    throw new IOException("Unknown ledger manager type: "
                            + lmType);
                }
            }
        }

        lmFactory = ReflectionUtils.newInstance(factoryClass);

        layout = new LedgerLayout(factoryClass.getName(),
                lmFactory.getCurrentVersion());
        try {
            layoutManager.storeLedgerLayout(layout);
        } catch (LedgerLayoutExistsException e) {
            LedgerLayout layout2 = layoutManager.readLedgerLayout();
            if (!layout2.equals(layout)) {
                throw new IOException(
                        "Contention writing to layout to zookeeper, "
                                + " other layout " + layout2
                                + " is incompatible with our " + "layout "
                                + layout);
            }
        }
        return lmFactory;
    }

}