//
// Name
//  $RCSfile: PartitionManager.java,v $
// 
// Copyright
//  Copyright 2010-2015 Cloud Software Group, Inc. ALL RIGHTS RESERVED. 
//  Cloud Software Group, Inc. Confidential Information
//
// History
//  $Revision: 1.1.2.57.2.1 $ $Date: 2015/04/30 19:57:55 $
//
package com.kabira.platform.highavailability;

import com.kabira.platform.ManagedObject;
import com.kabira.platform.ManagedClassError;
import com.kabira.platform.ResourceUnavailableException;
import com.kabira.platform.Transaction;
import com.kabira.platform.Transaction.Rollback;
import com.kabira.platform.disteng.DEPartitionManager;
import com.kabira.platform.disteng.DEPartitionNotifier;
import com.kabira.platform.disteng.DEPartition;
import com.kabira.platform.disteng.DEProperties;
import com.kabira.platform.disteng.DEPartitionNotFound;
import com.kabira.platform.disteng.DENodeMismatch;
import com.kabira.platform.disteng.DENotActiveNode;
import com.kabira.platform.disteng.DEEnableAction;
import com.kabira.platform.disteng.DEDisableAction;
import com.kabira.platform.disteng.DEMapperAudit;
import com.kabira.platform.disteng.PartitionState;
import com.kabira.platform.disteng.PartitionStatus;
import com.kabira.platform.disteng.DEReplicationType;
import com.kabira.platform.disteng.DESparsePartitionAudit;
import com.kabira.platform.disteng.DERemoteEnableAction;

import java.lang.reflect.Field;
import java.util.HashSet;
import java.util.logging.LogManager;
import java.util.logging.Logger;
import java.util.logging.Level;

/**
 * The PartitionManager class. This is the main interface for managing
 * partitions. 
 */
public final class PartitionManager 
{
    /**
     * An enumeration of the possible actions that can be taken when
     * enabling high availability for a node.
     */
    public enum EnableAction
    {
        /**
         * Perform any object migrations needed to activate the
         * partitions defined for this node. This option should be used
         * for newly added nodes.
         */
        JOIN_CLUSTER,
        /**
         * Before migrating the partition data, remove any instances that
         * exist on the local node. This option should be used for nodes
         * where the local data doesn't need to be preserved.
         */
        JOIN_CLUSTER_PURGE,
        /**
         * For all partitions where the node joining the cluster is the
         * current active node, access the remote node where the
         * partition should be restored from, and compare the objects
         * between the local node and remote node. If an object on the
         * local node has been updated, or a new object created, push the
         * update to the remote node. If the remote node has been
         * updated, replicate the change to the local node. If a state
         * conflict is detected, or a duplicate key is found, remove the
         * local object and copy the remote node object to the local
         * node.
         * <p>
         * For all partitions where the node joining the cluster is
         * acting as a replica, the replica objects are deleted and
         * instances migrated from the remote node.
         * <p>
         * Warning: This is performance hostile, and should only be done
         * when resolving a split-brain scenario in a cluster.
         * 
         * @see Partition.Properties#restoreFromNode(String)
         */
        JOIN_CLUSTER_RESTORE
    }

    /**
     * An enumeration of the possible actions that can be taken when
     * disabling high availability for a node.
     */
    public enum DisableAction
    {
        /**
         * Perform any object migrations needed to migrate
         * partitions owned by this node to the first order
         * replica node, and remove this node from all replica
         * lists. Before the disable is executed, an audit is
         * done to insure that no partitions will be moved to the
         * UNAVAILABLE state.
         */
        LEAVE_CLUSTER,
        /**
         * Same as LEAVE_CLUSTER, but with audits disabled.
         */
        LEAVE_CLUSTER_FORCE,
    }

    /**
     * Find all partitions on the local node.
     * 
     * @return Array of Partition instances.
     */
    public static final Partition [] getPartitions()
    {
        DEPartitionManager pm = new DEPartitionManager();
        DEPartition [] dp = pm.getPartitions();

        Partition [] pa = new Partition[dp.length];

        for (int i = 0; i < dp.length; i++)
        {
            pa[i] = new Partition(
                            dp[i].name,
                            dp[i].nodes,
                            dp[i].replicationTypeList,
                            mapState(dp[i].state), 
                            mapStatus(dp[i].status),
                            dp[i].lastUpdated,
                            dp[i].forceReplication,
                            dp[i].objectChunks,
                            dp[i].numThreads,
                            mapDESparseAudit(dp[i].sparseAudit),
                            mapDERemoteEnableAction(dp[i].remoteEnableAction),
                            dp[i].broadcastUpdates);
        }
        return pa;
    }

    /**
     * Find a partition by name.
     * 
     * @param partitionName Name of partition to find.
     *
     * @return Partition instance.
     *
     * @exception PartitionNotFound
     *  The given partition doesn't exist.
     * @exception IllegalArgumentException
     *  The partitionName was empty.
     */
    public static final Partition getPartition(
        final String partitionName) throws
            IllegalArgumentException, PartitionNotFound
    {
        if (partitionName.isEmpty())
        {
            throw new IllegalArgumentException("partitionName cannot be empty");
        }

        return _getPartition(partitionName);
    }

    /**
     * Get the partition an object is in.
     * 
     * @param obj Object to examine.
     *
     * @return Partition instance.
     *
     * @exception ManagedClassError
     *  The object is not a Managed object.
     * @exception PartitionNotFound
     *  The given object isn't in a partition. This can happen if
     *  object instances were created before a {@link PartitionMapper} was
     *  installed for a managed class, or no mapper was ever defined.
     */
    public static final Partition getObjectPartition(
        final Object obj) throws ManagedClassError, PartitionNotFound
    {
        checkManagedObject(obj);

        DEPartitionManager pm = new DEPartitionManager();

        String  partitionName = pm.getObjectPartition(obj);
        if (partitionName.length() == 0)
        {
            throw new PartitionNotFound(
                "No Partition exists for object '" + obj + "'");
        }

        return _getPartition(partitionName);
    }

    /**
     * Move the instance from the current partition (if any) to
     * the partition defined by the installed mapper for the object class.
     * The interactions between installed mappers and partitions when an
     * object is repartitioned is enumerated below:
     * <ul>
     * <li>
     * If the object was not associated with a partition, and a mapper
     * maps it to a new partition, the instance is replicated to all
     * nodes defined for the new partition.
     * </li>
     * <li>
     * If the object was mapped to a pre-existing partition, and a mapper
     * maps it to a new partition, the instance is replicated to all
     * nodes defined for the new partition. The instance is removed from
     * all nodes that were defined in the old partition but are not
     * defined with the new partition.
     * </li>
     * <li>
     * If the object was associated with a partition, and a mapper no
     * longer exists, the instance is removed from all nodes that that
     * were defined in the old partition, and the instance is left in
     * shared memory on the local node as un-partitioned.
     * </li>
     * <li>
     * If the object was not associated with a partition, and a mapper
     * does not exist, this operation does nothing.
     * </li>
     * </ul>
     *
     * @param obj Object to repartition.
     *
     * @exception ManagedClassError
     *  The object is not a Managed object.
     * @exception ResourceUnavailableException
     *  An illegal partition name was returned from the mapper, or the
     *  partition that is returned in not in the <b>ACTIVE</b> state.
     */
    public static final void repartitionInstance(final Object obj)
          throws ManagedClassError
    {
        checkManagedObject(obj);

        // First, clear the cached location in the Java proxy for this object.
        //
        // FIX THIS, DS: make this an undocumented native entry point
        //               on ManagedObject?
        try
        {
            Field localityField = obj.getClass().getField("$m_locality");
            localityField.setByte(obj, (byte)0);
        }
        catch (NoSuchFieldException ex)
        {
            throw new ManagedClassError(
                    "Clearing locality failed: " + ex.getMessage());
        }
        catch (SecurityException ex)
        {
            throw new ManagedClassError(
                    "Clearing locality failed: " + ex.getMessage());
        }
        catch (IllegalArgumentException ex)
        {
            throw new ManagedClassError(
                    "Clearing locality failed: " + ex.getMessage());
        }
        catch (IllegalAccessException ex)
        {
            throw new ManagedClassError(
                    "Clearing locality failed: " + ex.getMessage());
        }

        DEPartitionManager pm = new DEPartitionManager();
        pm.repartitionObject(obj);
    }

    /**
     * Use {@link #definePartition(String, Partition.Properties, String, ReplicaNode [])}
     * @param partitionName Name of partition to create.
     * @param nodes An ordered list of nodes for the partition.
     */
    @Deprecated
    public static final void definePartition(
        final String partitionName,
        final String [] nodes) throws IllegalArgumentException
    {
        if (partitionName.isEmpty())
        {
            throw new IllegalArgumentException("partitionName cannot be empty");
        }
        validateNodeList(nodes);

        DEReplicationType [] rtl = new DEReplicationType[nodes.length];
        for (int i = 0; i < rtl.length; i++)
        {
            rtl[i] = DEReplicationType.DataSynchronous;
        }

        DEProperties props = Partition.defaultDEProperties();

        definePartition(partitionName, props, nodes, rtl);
    }

    /**
     * Use {@link #definePartition(String, Partition.Properties, String, ReplicaNode [])}
     * instead.
     * @param partitionName Name of partition to create.
     * @param restoreFromNode The remote node to use when restoring the
     *  partition's objects.
     * @param nodes An ordered list of nodes for the partition.
     */
    @Deprecated
    public static final void definePartition(
        final String partitionName,
        final String restoreFromNode,
        final String [] nodes) throws IllegalArgumentException, NotActiveNode
    {
        if (restoreFromNode.isEmpty())
        {
            throw new IllegalArgumentException(
                "restoreFromNode cannot be empty");
        }

        if (partitionName.isEmpty())
        {
            throw new IllegalArgumentException("partitionName cannot be empty");
        }
        validateNodeList(nodes);

        DEReplicationType [] rtl = new DEReplicationType[nodes.length];
        for (int i = 0; i < rtl.length; i++)
        {
            rtl[i] = DEReplicationType.DataSynchronous;
        }

        DEProperties props = Partition.defaultDEProperties();
        props.restoreFromNode = restoreFromNode;

        definePartition(partitionName, props, nodes, rtl);
    }

    /**
     * Define a partition.
     * <p>
     * This method defines a partition. This method can be called on any
     * node in the cluster to define a new partition, or change the   
     * definition of a pre-existing partition. Once all partitions have
     * been defined for a node, the enablePartitions() method should be
     * called to move the partitions to the active state. It is a runtime
     * error to attempt to create or modify instances in a newly defined
     * partition until the enablePartitions() method is called.
     * <p>
     * The activeNode is the node that will be the active node for the
     * partition after it is enabled.
     * <p>
     * Optional partition properties can be used by passing in an
     * instance of the {@link Partition.Properties} class. User defined
     * properties such as forceReplication are saved and subsequently 
     * used during failover processing. 
     * <p>
     * The replicas array contains an ordered list of replica nodes for
     * the partition. A failure of the active node results in the
     * partition automatically migrating to the the next entry in the
     * replicas array. Objects are sent to each replica node based on
     * the configuration in the ReplicaNode entry.
     *
     * @param partitionName Name of partition to create.
     * @param partitionProperties Optional properties for the partition.
     *  If null, the default property values are used.
     * @param activeNode The active node for the partition
     * @param replicas An ordered list of replica nodes for the partition.
     *  Should be passed in as a null instance or a zero length array if
     *  no replicas are desired.
     *
     * @exception IllegalArgumentException
     *  The partitionName or activeNode was empty, or the replicas array
     *  was invalid.
     * @exception NotActiveNode
     *  The current node is not the active node for the partition and 
     *  restoreFromNode was defined in partitionProperties.
     * @exception ResourceUnavailableException
     *  The operation timed out waiting for the active node to become
     *  available.
     */
    public static final void definePartition(
        final String partitionName,
        final Partition.Properties partitionProperties,
        final String activeNode,
        final ReplicaNode [] replicas)
            throws IllegalArgumentException, NotActiveNode, ResourceUnavailableException
    {
        if (partitionName.isEmpty())
        {
            throw new IllegalArgumentException("partitionName cannot be empty");
        }

        if (activeNode.isEmpty())
        {
            throw new IllegalArgumentException("activeNode cannot be empty");
        }

        validateReplicaList(replicas);

        int nodeLength = (replicas == null) ? 1 : replicas.length + 1;

        String [] nodes = new String[nodeLength];

        nodes[0] = activeNode;
        if (replicas != null)
        {
            for (int i = 0; i < replicas.length; i++)
            {
                assert( i + 1 < nodeLength );
                nodes[i + 1] = replicas[i].nodeName;
            }
        }
        validateNodeList(nodes);

        DEReplicationType [] rtl = new DEReplicationType[nodes.length];
        rtl[0] = DEReplicationType.DataSynchronous;
        if (replicas != null)
        {
            for (int i = 0; i < replicas.length; i++)
            {
                rtl[i + 1] = mapReplicationType(replicas[i].replicationType);
            }
        }

        DEProperties props = Partition.copyDEProperties(partitionProperties);

        definePartition(partitionName, props, nodes, rtl);
    }

    /**
     * Associate a mapper to a given managed class.
     *
     * This method uses {@link PartitionMapper.Properties.Audit#IGNORE_PARTITIONING}
     * for audit.
     * 
     * @param managedClass Name of a Managed object.
     * @param partitionMapper User defined PartitionMapper instance.
     *
     * @exception ManagedClassError
     *  The managedClass is not a Managed object, or has a non-abstract
     *  parent class with a key defined but no mapper installed.
     *
     * @see PartitionManager#setMapper(Class, PartitionMapper, PartitionMapper.Properties)
     */
    public static final void setMapper(
        final Class<?> managedClass,
        final PartitionMapper partitionMapper) throws ManagedClassError
    {
        setMapper(managedClass, partitionMapper, null);
    }

    /**
     * Associate a mapper to a given managed class.
     * <p>
     * This method is used to associate a user defined PartitionMapper
     * instance with a managed class. All managed types that are replicated 
     * must have a PartitionMapper associated with the class. This method must
     * be called on all nodes that has the partition definition.
     * <p>
     * The same PartitionMapper instance can be associated with multiple
     * classes. If a PartitionMapper is already associated with a class,
     * the new value overwrites the previous one.
     * <p>
     * Applications should only delete a PartitionMapper instance after a
     * new instance is installed. If a PartitionMapper instance is
     * deleted, any subsequent object creates for the class will throw a
     * runtime exception.
     * <p>
     * Mapper instances are inherited by all classes that extend the
     * managed class. If the managed class extends another class, that
     * class must not have any keys defined.
     * <p>
     * If auditing is enabled in the mapperProperties instance, this routine
     * will verify that no unpartitioned instances of the given class
     * already exist in shared memory. This can happen if instances are
     * created before a mapper is installed.
     * 
     * @param managedClass Managed class.
     * @param partitionMapper User defined PartitionMapper instance.
     * @param mapperProperties Optional mapper properties. If null,
     * {@link PartitionMapper.Properties.Audit#IGNORE_PARTITIONING} is used.
     *
     * @return Old mapper instance associated with managedClass, or null if
     *  no mapper was set.
     *
     * @exception ManagedClassError
     *  The managedClass is not a Managed object, or has a non-abstract
     *  parent class with a key defined but no mapper installed.
     */
    public static final PartitionMapper setMapper(
        final Class<?> managedClass,
        final PartitionMapper partitionMapper,
        final PartitionMapper.Properties mapperProperties)
            throws ManagedClassError
    {
        try
        {
            checkManagedClass(managedClass);
            return setMapper(managedClass.getName(),
                partitionMapper, mapperProperties);
        }
        catch (ResourceUnavailableException ex)
        {
            //
            // FIX THIS: Probably  untestable because of the checkManagedClass()
            //
            throw new ManagedClassError(ex.getMessage());
        }
    }

    /**
     * Associate a mapper to a given shared memory type.
     * <p>
     * This method is used to associate a user defined PartitionMapper
     * instance with shared memory type. All types that are replicated 
     * must have a PartitionMapper associated with the class. This method must
     * be called on all nodes that has the partition definition.
     * <p>
     * The same PartitionMapper instance can be associated with multiple
     * type. If a PartitionMapper is already associated with a type,
     * the new value overwrites the previous one.
     * <p>
     * Applications should only delete a PartitionMapper instance after a
     * new instance is installed. If a PartitionMapper instance is
     * deleted, any subsequent object creates for the type will throw a
     * runtime exception.
     * <p>
     * Mapper instances are inherited by all subtypes. If the type
     * extends another type, that class must not have any keys defined.
     * <p>
     * If auditing is enabled in the mapperProperties instance, this routine
     * will verify that no unpartitioned instances of the given type
     * already exist in shared memory. This can happen if instances are
     * created before a mapper is installed.
     * 
     * @param typeName Name of a shared memory type.
     * @param partitionMapper User defined PartitionMapper instance.
     * @param mapperProperties Optional mapper properties. If null,
     * {@link PartitionMapper.Properties.Audit#IGNORE_PARTITIONING} is used.
     *
     * @return Old mapper instance associated with managedClass, or null if
     *  no mapper was set.
     *
     * @exception ResourceUnavailableException
     *  The typeName is not a shared memory type, or has a non-abstract
     *  parent class with a key defined but no mapper installed.
     */
    public static final PartitionMapper setMapper(
        final String typeName,
        final PartitionMapper partitionMapper,
        final PartitionMapper.Properties mapperProperties)
            throws ResourceUnavailableException
    {
        if (partitionMapper == null)
        {
            throw new NullPointerException("partitionMapper cannot be null");
        }
        DEPartitionManager pm = new DEPartitionManager();
        DEMapperAudit audit = (mapperProperties == null)
            ? DEMapperAudit.MapperIgnore
            : mapAudit(mapperProperties.getAudit());

        return (PartitionMapper)pm.setMapperAudit(
                        typeName, partitionMapper, audit);
    }

    /**
     * Clear a mapper for a given managed class.
     * <p>
     * This method is used to remove a user defined PartitionMapper
     * instance for a managed class. This method must be called on all
     * nodes where the mapper was previously defined.
     * <p>
     * If no PartitionMapper is associated with a class, this operation
     * is a noop.
     *
     * @param managedClass Managed class.
     *
     * @exception ManagedClassError
     *  The managedClass is not a Managed object.
     *
     * @see PartitionManager#setMapper(Class, PartitionMapper, PartitionMapper.Properties)
     */
    public static final void clearMapper(
        final Class<?> managedClass) throws ManagedClassError
    {
        try
        {
            clearMapper(managedClass.getName());
        }
        catch (ResourceUnavailableException ex)
        {
            throw new ManagedClassError(ex.getMessage());
        }
    }

    /**
     * Clear a mapper for a given type.
     * <p>
     * This method is used to remove a user defined PartitionMapper
     * instance for a shared memory type. This method must be called on all
     * nodes where the mapper was previously defined.
     * <p>
     * If no PartitionMapper is associated with a class, this operation
     * is a noop.
     *
     * @param typeName Name of a Managed object.
     *
     * @exception ResourceUnavailableException
     *  The typeName does not exist in shared memory.
     *
     * @see PartitionManager#setMapper(String, PartitionMapper, PartitionMapper.Properties)
     */
    public static final void clearMapper(
        final String typeName) throws ResourceUnavailableException
    {
        DEPartitionManager pm = new DEPartitionManager();
        pm.setMapperAudit(typeName, null, DEMapperAudit.MapperIgnore);
    }

    /**
     * Use {@link com.kabira.platform.ManagedObject#isPartitioned}
      * @param managedClass Class to check.
      * @return true if class is partitioned.
     */
    @Deprecated
    public static final boolean isPartitioned(final Class<?> managedClass)
    {
        return ManagedObject.isPartitioned(managedClass);
    }

    /**
     * Enable high availability for the all partitions defined on the node.
     * <p>
     * This method is used to enable all partitions that have been
     * defined for this node. This method should be called each time new
     * partitions are defined.
     * This method blocks until all required object migration
     * is complete.
     *
     * @param enableAction The action to take when enabling partitions.
     *
     * @exception NodeMismatch
     *  The partition does not have the local node in the node list, and the
     *  active node for the partition has a different node list.
     * @exception ResourceUnavailableException
     *  The minimum number of nodes needed to establish a quorum has not been
     *  seen.
     */
    public static final void enablePartitions(
        final EnableAction enableAction)
            throws NodeMismatch, ResourceUnavailableException
    {
        DEPartitionManager pm = new DEPartitionManager();
        try
        {
            pm.enablePartitions(mapEnable(enableAction));
        }
        catch (DENodeMismatch ex)
        {
            throw new NodeMismatch(ex.errMsg);
        }
    }

    /**
     * Disable high availability for the all partitions defined on the node.
     * <p>
     * This method is used to remove the local node from the node list of
     * all partitions that have been defined for this node. If called
     * multiple times, the additional calls have no effect.
     *
     * @param disableAction  The action to take when disabling partitions.
     *
     * @exception ResourceUnavailableException
     *  One or more partitions would be in the UNAVAILABLE state after the
     *  disable executes, not thrown if LEAVE_CLUSTER_FORCE is used.
     */
    public static final void disablePartitions(
        final DisableAction disableAction) throws ResourceUnavailableException
    {
        DEPartitionManager pm = new DEPartitionManager();
        pm.disablePartitions(mapDisable(disableAction));
    }

    /**
     * Wait for a remote node to become available.
     *
     * @param remoteNode name of remote node to wait for.
     *
     * @exception ResourceUnavailableException
     *  The operation timed out waiting for the remote node to become
     *  available.
     */
    public static final void waitForNode(
        final String remoteNode) throws ResourceUnavailableException
    {
        DEPartitionManager pm = new DEPartitionManager();
        pm.waitForNode(remoteNode);
    }

    /**
     * Create local copies of all partition defined by the remote node.
     * <p>
     * After execution of this method completes, all partitions defined
     * on the remote node will exist on the local node, and will be
     * accessible using the {@link #getPartition(String)} method.
     * 
     * @param remoteNode name of remote node to access.
     *
     * @exception ResourceUnavailableException
     *  The operation timed out waiting for the remote node to become
     *  available.
     */
    public static final void getRemotePartitions(
        final String remoteNode) throws ResourceUnavailableException
    {
        DEPartitionManager pm = new DEPartitionManager();
        pm.getRemotePartitions(remoteNode);
    }

    //
    // This routine performs initialization of the DistributionLogger
    // logger, and runtime resources used in the highavailability
    // package. It normally is called from a static initializer defined
    // in this class, but can also be called from ManagedObject via
    // reflection.
    //
    // Note: We assume caller synchronizes this call.
    //
    private static final void initialize()
    {
        if (m_logger == null)
        {
            m_logger = new DistributionLogger();
            LogManager lm = LogManager.getLogManager();
            lm.addLogger(m_logger);
            ManagedObject.setDistributedQuery(new DistributedQuery());
        }
    }

    //
    // Package private methods. These don't bother with handling
    // PartitionNotFound exceptions, since they can only be called after
    // getPartition() was executed, and we never delete partitions.
    //

    static final void updatePartition(
        final String existingPartition,
        DEProperties props) throws NodeMismatch, NotActiveNode
    {
        DEPartitionManager pm = new DEPartitionManager();
        try
        {
            pm.updatePartition(existingPartition, props);
        }
        catch (DENodeMismatch ex)
        {
            throw new NodeMismatch(ex.errMsg);
        }
        catch (DENotActiveNode ex)
        {
            throw new NotActiveNode(ex.errMsg);
        }
    }

    static final void migratePartition(
        final String partitionName,
        DEProperties props,
        final String [] nodes,
        DEReplicationType [] rtl) throws NotActiveNode
    {
        DEPartitionManager pm = new DEPartitionManager();
        try
        {
            pm.migratePartition(partitionName, props, nodes, rtl);
        }
        catch (DENotActiveNode ex)
        {
            throw new NotActiveNode(ex.errMsg);
        }
    }

    static final void setNotifier(
        final String partitionName,
        final PartitionNotifier partitionNotifier) throws PartitionNotFound
    {
        Notifier notifier = new Notifier(partitionNotifier);
        DEPartitionManager pm = new DEPartitionManager();
        pm.setNotifier(partitionName, notifier);
    }

    //
    // Validate the replica list. We just check for null replicas here, the
    // rest of the checks are done in validateNodeList()
    //
    static void validateReplicaList(
        final ReplicaNode [] replicas) throws IllegalArgumentException
    {
        if (replicas != null)
        {
            for (int i = 0; i < replicas.length; i++)
            {
                ReplicaNode r = replicas[i];
                if (r == null)
                {
                    throw new IllegalArgumentException(
                        "Replica entry at index " + i + " cannot be empty");
                }
            }
        }
    }

    //
    // Validate the node list
    //
    static void validateNodeList(
        final String [] nodes) throws IllegalArgumentException
    {
        HashSet<String>     nodeSet = new HashSet<String>();
        if (nodes.length == 0)
        {
            throw new IllegalArgumentException(
                "The node array must have at least one entry");
        }
        for (int i = 0; i < nodes.length; i++)
        {
            String node = nodes[i];
            if (node == null || node.isEmpty())
            {
                throw new IllegalArgumentException("Node entry at index " +
                    i + " cannot be empty");
            }
            if (nodeSet.contains(node))
            {
                throw new IllegalArgumentException("Node entry " + node +
                    " at index " + i + " is a duplicate. " +
                    "All nodes must be unique");
            }
            nodeSet.add(node);
        }
    }

    //
    // Internal getPartition() method
    //
    private static Partition _getPartition(
        final String partitionName) throws PartitionNotFound
    {
        assert( !partitionName.isEmpty() );

        DEPartitionManager pm = new DEPartitionManager();

        try
        {
            DEPartition p = pm.getPartition(partitionName);
            return new Partition(
                        p.name,
                        p.nodes,
                        p.replicationTypeList,
                        mapState(p.state),
                        mapStatus(p.status),
                        p.lastUpdated,
                        p.forceReplication,
                        p.objectChunks,
                        p.numThreads,
                        mapDESparseAudit(p.sparseAudit),
                        mapDERemoteEnableAction(p.remoteEnableAction),
                        p.broadcastUpdates);
        }
        catch (DEPartitionNotFound ex)
        {
            throw new PartitionNotFound(ex.errMsg);
        }
    }

    //
    // Wrapper class to map notifiers.
    //
    private static class Notifier extends DEPartitionNotifier
    {
        public void stateChange(
                final String partitionName,
                final PartitionState oldState,
                final PartitionState newState)
        {
            if (ManagedObject.isEmpty(m_partitionNotifier))
            {
                ManagedObject.delete(this);
            }
            else
            {
                m_partitionNotifier.stateChange(
                    partitionName,
                    mapState(oldState),
                    mapState(newState));
            }
        }

        Notifier(PartitionNotifier partitionNotifier)
        {
            m_partitionNotifier = partitionNotifier;
        }

        PartitionNotifier   m_partitionNotifier;
    }

    //
    // Wrapper method to map definePartition exceptions.
    //
    private static void definePartition(
        final String partitionName,
        final DEProperties props,
        final String [] nodes,
        DEReplicationType [] rtl)
            throws NotActiveNode
    {
        DEPartitionManager pm = new DEPartitionManager();

        try
        {
            pm.definePartition(partitionName, props, nodes, rtl);
        }
        catch (DENotActiveNode ex)
        {
            throw new NotActiveNode(ex.errMsg);
        }
    }

    //
    // Utility function to map PartitionState to Partition.State
    //
    private static Partition.State mapState(PartitionState state)
    {
        Partition.State s;

        switch (state)
        {
        case PartitionInitial:
            s = Partition.State.INITIAL;
            break;
        case PartitionActive:
            s = Partition.State.ACTIVE;
            break;
        case PartitionUpdating:
            s = Partition.State.UPDATING;
            break;
        case PartitionMigrating:
            s = Partition.State.MIGRATING;
            break;
        case PartitionReplicating:
            s = Partition.State.REPLICATING;
            break;
        default:
            assert( state == PartitionState.PartitionUnavailable );
            s = Partition.State.UNAVAILABLE;
            break;
        }
        return s;
    }

    //
    // Utility function to map PartitionStatus to Partition.Status
    //
    private static Partition.Status mapStatus(PartitionStatus status)
    {
        Partition.Status s;

        switch (status)
        {
        case LocalDisabled:
            s = Partition.Status.LOCAL_DISABLED;
            break;
        case LocalDefined:
            s = Partition.Status.LOCAL_DEFINED;
            break;
        case LocalEnabled:
            s = Partition.Status.LOCAL_ENABLED;
            break;
        case RemoteDefined:
            s = Partition.Status.REMOTE_DEFINED;
            break;
        default:
            assert( status == PartitionStatus.RemoteEnabled );
            s = Partition.Status.REMOTE_ENABLED;
            break;
        }
        return s;
    }

    //
    // Utility function to map PartitionManager.EnableAction to EnableAction
    //
    static DEEnableAction mapEnable(
        final PartitionManager.EnableAction enableAction)
    {
        DEEnableAction e;

        switch (enableAction)
        {
        case JOIN_CLUSTER:
            e = DEEnableAction.JoinCluster;
            break;
        case JOIN_CLUSTER_PURGE:
            e = DEEnableAction.JoinClusterPurge;
            break;
        default:
            assert( enableAction == EnableAction.JOIN_CLUSTER_RESTORE );
            e = DEEnableAction.JoinClusterRestore;
            break;
        }
        return e;
    }

    //
    // Utility function to map PartitionManager.DisableAction to DisableAction
    //
    static DEDisableAction mapDisable(
        final PartitionManager.DisableAction disableAction)
    {
        DEDisableAction e;

        switch (disableAction)
        {
        case LEAVE_CLUSTER:
            e = DEDisableAction.LeaveCluster;
            break;
        default:
            assert( disableAction == DisableAction.LEAVE_CLUSTER_FORCE );
            e = DEDisableAction.LeaveClusterForce;
            break;
        }
        return e;
    }

    //
    // Utility function to map ReplicaNode.ReplicationType to DEReplicationType
    //
    static DEReplicationType mapReplicationType(
        final ReplicaNode.ReplicationType replicationType)
    {
        DEReplicationType rt;

        switch (replicationType)
        {
        case SYNCHRONOUS:
            rt = DEReplicationType.DataSynchronous;
            break;
        default:
            assert( replicationType ==
                ReplicaNode.ReplicationType.ASYNCHRONOUS );
            rt = DEReplicationType.DataAsynchronous;
            break;
        }
        return rt;
    }

    //
    // Utility function to map DEReplicationType to ReplicaNode.ReplicationType
    //
    static ReplicaNode.ReplicationType mapDEReplicationType(
        final DEReplicationType rt)
    {
        ReplicaNode.ReplicationType replicationType;

        switch (rt)
        {
        case DataSynchronous:
            replicationType = ReplicaNode.ReplicationType.SYNCHRONOUS;
            break;
        default:
            assert( rt == DEReplicationType.DataAsynchronous );
            replicationType = ReplicaNode.ReplicationType.ASYNCHRONOUS;
            break;
        }
        return replicationType;
    }

    //
    // Utility function to map SparseAudit to DESparsePartitionAudit
    //
    static DESparsePartitionAudit mapSparseAudit(
        final Partition.Properties.SparseAudit sparseAudit)
    {
        DESparsePartitionAudit spa;

        switch (sparseAudit)
        {
        case VERIFY_NODE_LIST:
            spa = DESparsePartitionAudit.VerifyNodeList;
            break;
        default:
            assert( sparseAudit ==
                Partition.Properties.SparseAudit.IGNORE_NODE_LIST );
            spa = DESparsePartitionAudit.IgnoreNodeList;
            break;
        }
        return spa;
    }

    //
    // Utility function to map DESparsePartitionAudit to SparseAudit
    //
    static Partition.Properties.SparseAudit mapDESparseAudit(
        final DESparsePartitionAudit sparseAudit)
    {
        Partition.Properties.SparseAudit spa;

        switch (sparseAudit)
        {
        case VerifyNodeList:
            spa = Partition.Properties.SparseAudit.VERIFY_NODE_LIST;
            break;
        default:
            assert( sparseAudit == DESparsePartitionAudit.IgnoreNodeList );
            spa = Partition.Properties.SparseAudit.IGNORE_NODE_LIST;
            break;
        }
        return spa;
    }

    //
    // Utility function to map RemoteEnableAction to DERemoteEnableAction
    //
    static DERemoteEnableAction mapRemoteEnableAction(
        final Partition.Properties.RemoteEnableAction remoteEnableAction)
    {
        DERemoteEnableAction rea;

        switch (remoteEnableAction)
        {
        case ENABLE_PARTITION:
            rea = DERemoteEnableAction.EnablePartition;
            break;
        default:
            assert( remoteEnableAction ==
                Partition.Properties.RemoteEnableAction.LEAVE_DISABLED );
            rea = DERemoteEnableAction.LeaveDisabled;
            break;
        }
        return rea;
    }

    //
    // Utility function to map DERemoteEnableAction to RemoteEnableAction
    //
    static Partition.Properties.RemoteEnableAction mapDERemoteEnableAction(
        final DERemoteEnableAction remoteEnableAction)
    {
        Partition.Properties.RemoteEnableAction rea;

        switch (remoteEnableAction)
        {
        case EnablePartition:
            rea = Partition.Properties.RemoteEnableAction.ENABLE_PARTITION;
            break;
        default:
            assert( remoteEnableAction == DERemoteEnableAction.LeaveDisabled );
            rea = Partition.Properties.RemoteEnableAction.LEAVE_DISABLED;
            break;
        }
        return rea;
    }

    //
    // Utility function to map MapperProperties.Audit to DEMapperAudit
    //
    private static DEMapperAudit mapAudit(
        final PartitionMapper.Properties.Audit audit)
    {
        DEMapperAudit e;

        switch (audit)
        {
        case VERIFY_PARTITIONING:
        case VERIFY_PARTIONING:
            e = DEMapperAudit.MapperVerify;
            break;
        case VERIFY_DEFER_INSTALL:
            e = DEMapperAudit.MapperDefer;
            break;
        default:
            assert( audit == PartitionMapper.Properties.Audit.IGNORE_PARTITIONING );
            e = DEMapperAudit.MapperIgnore;
            break;
        }
        return e;
    }

    //
    // Validates a given object is managed.
    //
    private static void checkManagedObject(Object obj)
    {
        if (!ManagedObject.isManaged(obj))
        {
            throw new ManagedClassError("class " + obj.getClass().getName() +
                " is not a Managed class.");
        }
    }

    //
    // Validates a given class is managed.
    //
    private static void checkManagedClass(Class<?> klass)
    {
        if (!ManagedObject.isManagedClass(klass))
        {
            throw new ManagedClassError("class " + klass.getName() +
                " is not a Managed class.");
        }
    }

    //
    // Never should be called
    //
    private PartitionManager()
    {
    }

    //
    // Private logger class
    //
    private static class DistributionLogger extends Logger
    {
        DistributionLogger()
        {
            super("com.kabira.platform.highavailability", null);
            m_enabled = false;
        }

        public void setLevel(Level level)
        {
            super.setLevel(level);
            if (level != null)
            {
                m_enabled = (level.intValue() <= Level.FINEST.intValue())
                    ? true : false;
                if (Transaction.isActive())
                {
                    debugTrace();
                }
                else
                {
                    new Transaction("DistributionLogger")
                    {
                        @Override
                        protected void run() throws Rollback
                        {
                            debugTrace();
                        }
                    }.execute();
                }
            }
        }

        private boolean m_enabled;

        private void debugTrace()
        {
                DEPartitionManager pm = new DEPartitionManager();
                pm.debugTrace(m_enabled);
        }

        @Override
        public String toString()
        {
            StringBuilder sb =
                new StringBuilder("DistributionLogger: debug tracing is ");
            sb.append(m_enabled ? "enabled" : "disabled");
            return sb.toString();
        }
    }

    private static DistributionLogger    m_logger;

    //
    // Create and register logger and DistributedQuery instance at
    // start of world.
    //
    static
    {
        initialize();
    }
}