bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerManagerFactory.java - bookkeeper - Git at Google

 package org.apache.bookkeeper.meta;

 /**
  * 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.
  */

 import java.io.IOException;

 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import org.apache.bookkeeper.replication.ReplicationException;
 import org.apache.bookkeeper.conf.AbstractConfiguration;
 import org.apache.bookkeeper.util.ReflectionUtils;
 import org.apache.commons.configuration.ConfigurationException;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.ZooKeeper;

 public abstract class LedgerManagerFactory {

     static final Logger LOG = LoggerFactory.getLogger(LedgerManagerFactory.class);
     // v1 layout
     static final int V1 = 1;

     /**
      * Return current factory version.
      *
      * @return current version used by factory.
      */
     public abstract int getCurrentVersion();

     /**
      * Initialize a factory.
      *
      * @param conf
      *          Configuration object used to initialize factory
      * @param zk
      *          Available zookeeper handle for ledger manager to use.
      * @param factoryVersion
      *          What version used to initialize factory.
      * @return ledger manager factory instance
      * @throws IOException when fail to initialize the factory.
      */
     public abstract LedgerManagerFactory initialize(final AbstractConfiguration conf,
                                                     final ZooKeeper zk,
                                                     final int factoryVersion)
     throws IOException;

     /**
      * Uninitialize the factory.
      *
      * @throws IOException when fail to uninitialize the factory.
      */
     public abstract void uninitialize() throws IOException;

     /**
      * Return the ledger id generator, which is used for global unique ledger id
      * generation.
      *
      * @return ledger id generator.
      */
     public abstract LedgerIdGenerator newLedgerIdGenerator();

     /**
      * return ledger manager for client-side to manage ledger metadata.
      *
      * @return ledger manager
      * @see LedgerManager
      */
     public abstract LedgerManager newLedgerManager();

     /**
      * Return a ledger underreplication manager, which is used to
      * mark ledgers as unreplicated, and to retrieve a ledger which
      * is underreplicated so that it can be rereplicated.
      *
      * @return ledger underreplication manager
      * @see LedgerUnderreplicationManager
      */
     public abstract LedgerUnderreplicationManager newLedgerUnderreplicationManager()
             throws KeeperException, InterruptedException, ReplicationException.CompatibilityException;

     /**
      * Create new Ledger Manager Factory.
      *
      * @param conf
      *          Configuration Object.
      * @param zk
      *          ZooKeeper Client Handle, talk to zk to know which ledger manager is used.
      * @return new ledger manager factory
      * @throws IOException
      */
     public static LedgerManagerFactory newLedgerManagerFactory(
         final AbstractConfiguration conf, final ZooKeeper zk)
             throws IOException, KeeperException, InterruptedException {
         Class<? extends LedgerManagerFactory> factoryClass;
         try {
             factoryClass = conf.getLedgerManagerFactoryClass();
         } catch (Exception e) {
             throw new IOException("Failed to get ledger manager factory class from configuration : ", e);
         }
         String ledgerRootPath = conf.getZkLedgersRootPath();

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

         // if zk is null, return the default ledger manager
         if (zk == 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 = LedgerLayout.readLayout(zk, ledgerRootPath);
         if (layout == null) { // no existing layout
             lmFactory = createNewLMFactory(conf, zk, factoryClass);
             return lmFactory
                     .initialize(conf, zk, lmFactory.getCurrentVersion());
         }
         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
             @SuppressWarnings("deprecation")
             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, zk, layout.getManagerVersion());
         }

         // handle V2 layout case
         if (factoryClass != null &&
             !layout.getManagerFactoryClass().equals(factoryClass.getName())) {

             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 cnfe) {
                 throw new IOException("Failed to instantiate ledger manager factory " + layout.getManagerFactoryClass());
             }
         }
         // instantiate a factory
         lmFactory = ReflectionUtils.newInstance(factoryClass);
         return lmFactory.initialize(conf, zk, layout.getManagerVersion());
     }

     /**
      * Creates the new layout and stores in zookeeper and returns the
      * LedgerManagerFactory instance.
      */
     private static LedgerManagerFactory createNewLMFactory(
             final AbstractConfiguration conf, final ZooKeeper zk,
             Class<? extends LedgerManagerFactory> factoryClass)
             throws IOException, KeeperException, InterruptedException {

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

         lmFactory = ReflectionUtils.newInstance(factoryClass);

         layout = new LedgerLayout(factoryClass.getName(),
                 lmFactory.getCurrentVersion());
         try {
             layout.store(zk, ledgerRootPath);
         } catch (KeeperException.NodeExistsException nee) {
             LedgerLayout layout2 = LedgerLayout.readLayout(zk, ledgerRootPath);
             if (!layout2.equals(layout)) {
                 throw new IOException(
                         "Contention writing to layout to zookeeper, "
                                 + " other layout " + layout2
                                 + " is incompatible with our " + "layout "
                                 + layout);
             }
         }
         return lmFactory;
     }

     /**
      * Format the ledger metadata for LedgerManager
      *
      * @param conf
      *            Configuration instance
      * @param zk
      *            Zookeeper instance
      */
     public void format(final AbstractConfiguration conf, final ZooKeeper zk)
             throws InterruptedException, KeeperException, IOException {

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

         LedgerLayout layout = LedgerLayout.readLayout(zk,
                 conf.getZkLedgersRootPath());
         layout.delete(zk, conf.getZkLedgersRootPath());
         // Create new layout information again.
         createNewLMFactory(conf, zk, factoryClass);
     }
 }
	package org.apache.bookkeeper.meta;

	/**
	* 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.
	*/

	import java.io.IOException;

	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import org.apache.bookkeeper.replication.ReplicationException;
	import org.apache.bookkeeper.conf.AbstractConfiguration;
	import org.apache.bookkeeper.util.ReflectionUtils;
	import org.apache.commons.configuration.ConfigurationException;
	import org.apache.zookeeper.KeeperException;
	import org.apache.zookeeper.ZooKeeper;

	public abstract class LedgerManagerFactory {

	static final Logger LOG = LoggerFactory.getLogger(LedgerManagerFactory.class);
	// v1 layout
	static final int V1 = 1;

	/**
	* Return current factory version.
	*
	* @return current version used by factory.
	*/
	public abstract int getCurrentVersion();

	/**
	* Initialize a factory.
	*
	* @param conf
	* Configuration object used to initialize factory
	* @param zk
	* Available zookeeper handle for ledger manager to use.
	* @param factoryVersion
	* What version used to initialize factory.
	* @return ledger manager factory instance
	* @throws IOException when fail to initialize the factory.
	*/
	public abstract LedgerManagerFactory initialize(final AbstractConfiguration conf,
	final ZooKeeper zk,
	final int factoryVersion)
	throws IOException;

	/**
	* Uninitialize the factory.
	*
	* @throws IOException when fail to uninitialize the factory.
	*/
	public abstract void uninitialize() throws IOException;

	/**
	* Return the ledger id generator, which is used for global unique ledger id
	* generation.
	*
	* @return ledger id generator.
	*/
	public abstract LedgerIdGenerator newLedgerIdGenerator();

	/**
	* return ledger manager for client-side to manage ledger metadata.
	*
	* @return ledger manager
	* @see LedgerManager
	*/
	public abstract LedgerManager newLedgerManager();

	/**
	* Return a ledger underreplication manager, which is used to
	* mark ledgers as unreplicated, and to retrieve a ledger which
	* is underreplicated so that it can be rereplicated.
	*
	* @return ledger underreplication manager
	* @see LedgerUnderreplicationManager
	*/
	public abstract LedgerUnderreplicationManager newLedgerUnderreplicationManager()
	throws KeeperException, InterruptedException, ReplicationException.CompatibilityException;

	/**
	* Create new Ledger Manager Factory.
	*
	* @param conf
	* Configuration Object.
	* @param zk
	* ZooKeeper Client Handle, talk to zk to know which ledger manager is used.
	* @return new ledger manager factory
	* @throws IOException
	*/
	public static LedgerManagerFactory newLedgerManagerFactory(
	final AbstractConfiguration conf, final ZooKeeper zk)
	throws IOException, KeeperException, InterruptedException {
	Class<? extends LedgerManagerFactory> factoryClass;
	try {
	factoryClass = conf.getLedgerManagerFactoryClass();
	} catch (Exception e) {
	throw new IOException("Failed to get ledger manager factory class from configuration : ", e);
	}
	String ledgerRootPath = conf.getZkLedgersRootPath();

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

	// if zk is null, return the default ledger manager
	if (zk == 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 = LedgerLayout.readLayout(zk, ledgerRootPath);
	if (layout == null) { // no existing layout
	lmFactory = createNewLMFactory(conf, zk, factoryClass);
	return lmFactory
	.initialize(conf, zk, lmFactory.getCurrentVersion());
	}
	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
	@SuppressWarnings("deprecation")
	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, zk, layout.getManagerVersion());
	}

	// handle V2 layout case
	if (factoryClass != null &&
	!layout.getManagerFactoryClass().equals(factoryClass.getName())) {

	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 cnfe) {
	throw new IOException("Failed to instantiate ledger manager factory " + layout.getManagerFactoryClass());
	}
	}
	// instantiate a factory
	lmFactory = ReflectionUtils.newInstance(factoryClass);
	return lmFactory.initialize(conf, zk, layout.getManagerVersion());
	}

	/**
	* Creates the new layout and stores in zookeeper and returns the
	* LedgerManagerFactory instance.
	*/
	private static LedgerManagerFactory createNewLMFactory(
	final AbstractConfiguration conf, final ZooKeeper zk,
	Class<? extends LedgerManagerFactory> factoryClass)
	throws IOException, KeeperException, InterruptedException {

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

	lmFactory = ReflectionUtils.newInstance(factoryClass);

	layout = new LedgerLayout(factoryClass.getName(),
	lmFactory.getCurrentVersion());
	try {
	layout.store(zk, ledgerRootPath);
	} catch (KeeperException.NodeExistsException nee) {
	LedgerLayout layout2 = LedgerLayout.readLayout(zk, ledgerRootPath);
	if (!layout2.equals(layout)) {
	throw new IOException(
	"Contention writing to layout to zookeeper, "
	+ " other layout " + layout2
	+ " is incompatible with our " + "layout "
	+ layout);
	}
	}
	return lmFactory;
	}

	/**
	* Format the ledger metadata for LedgerManager
	*
	* @param conf
	* Configuration instance
	* @param zk
	* Zookeeper instance
	*/
	public void format(final AbstractConfiguration conf, final ZooKeeper zk)
	throws InterruptedException, KeeperException, IOException {

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

	LedgerLayout layout = LedgerLayout.readLayout(zk,
	conf.getZkLedgersRootPath());
	layout.delete(zk, conf.getZkLedgersRootPath());
	// Create new layout information again.
	createNewLMFactory(conf, zk, factoryClass);
	}
	}