//
// Name
//  $RCSfile: Partition.java,v $
// 
// Copyright
//  Copyright 2010-2015 Cloud Software Group, Inc. ALL RIGHTS RESERVED. 
//  Cloud Software Group, Inc. Confidential Information
//
// History
//  $Revision: 1.1.2.30.2.1 $ $Date: 2015/08/05 20:33:21 $
//
package com.kabira.platform.highavailability;

import com.kabira.platform.ResourceUnavailableException;
import com.kabira.platform.disteng.DEPartitionManager;
import com.kabira.platform.disteng.DENodeMismatch;
import com.kabira.platform.disteng.DEReplicationType;
import com.kabira.platform.disteng.DEProperties;
import com.kabira.platform.disteng.DEPartitionStats;
import com.kabira.platform.disteng.DESparsePartitionAudit;
import com.kabira.platform.disteng.DEReplicaAudit;
import com.kabira.platform.disteng.DERemoteEnableAction;
import com.kabira.platform.disteng.DEPartitionBroadcast;
import java.util.Date;

/**
 * The Partition class. This non-managed class contains a snapshot of a
 * shared memory partition. Direct access to the Partition instance in
 * shared memory is not allowed to avoid transaction locking issues.
 * <p>
 * Note that partitions are only visible on their active and replica
 * nodes, not all nodes in the cluster.
 */
public final class Partition
{
    /**
     * The Partition states
     */
    public enum State
    {
        /** The partition has been defined, but not enabled */
        INITIAL,
        /** The partition is active */
        ACTIVE,
        /**
         * The partition definition is being updated. Entered when 
         * partition membership (i.e. the objects associated with the
         * partition) is being updated to split or merge partitions.
         */
        UPDATING,
        /**
         * The partition owner is being updated. Each object in the
         * partition is sent to the new owner and the location definition
         * of all instances are updated.
         */
        MIGRATING,
        /**
         * The partition replicas are being updated. Each object in the
         * partition is sent to the new replica list.
         */
        REPLICATING,
        /** 
         * The partition is not active because all associated nodes have
         * failed, the node was taken off-line because it is in a
         * non-quorum state, or the node was restarted.
         */
        UNAVAILABLE
    }

    /**
     * The Partition status
     */
    public enum Status
    {
        /** The partition has been disabled */
        LOCAL_DISABLED,
        /** The partition has been defined, but not enabled */
        LOCAL_DEFINED,
        /** The partition has been defined and enabled. */
        LOCAL_ENABLED,
        /** The partition has been defined remotely. */
        REMOTE_DEFINED,
        /**
         * The partition has been defined and enabled on a remote node
         * but not defined on this node.
         */
        REMOTE_ENABLED
    }

    /**
     * The Partition properties used when defining a partition
     */
    public final static class Properties
    {
        /**
         * An enumeration of the possible audits that can be taken when
         * enabling a sparse Partition.
         * <p>
         * A sparse Partition is a partition that doesn't have the local
         * node as the active node, or in the replica node list. When
         * enabled, no data migration takes place; instead the node list
         * is verified against the node list that is present on the
         * active node to verify they match.
         * <p>
         * When restoring multiple nodes, it may be necessary to disable
         * the auditing of node lists, since the active node may have a
         * node list that is the result of a previous failover.
         * <p>
         * The default audit is defined by {@link #DefaultSparseAudit},
         */
        public enum SparseAudit
        {
            /**
             * Verify that the defined node list matches the active node's
             * node list.
             */
            VERIFY_NODE_LIST,
            /**
             * Do not perform any verification of node lists.
             */
            IGNORE_NODE_LIST
        }

        /**
         * An enumeration of the possible audits done for each replica when
         * enabling a Partition.
         * <p>
         * The default audit is defined by {@link #DefaultReplicaAudit},
         */
        public enum ReplicaAudit
        {
            /**
             * Block until all replica nodes are active. The runtime will
             * wait until each replica node is active before enabling the
             * partition. If the wait times out, a ResourceUnavailableException
             * exception is thrown.
             */
            WAIT_ACTIVE,
            /**
             * Discard the replica if not active. If the replica node state
             * is not Active, the replica node is removed from the node list,
             * and a log message is generated.
             */
            DISCARD_REPLICA,
            /**
             * The replica node state is ignored.
             */
            @Deprecated
            IGNORE_REPLICA
        }

        /**
         * An enumeration of the possible actions to be taken when a remote
         * node enables a Partition.
         * <p>
         * When a remote node enables a partition, it will enable the
         * partition on all nodes in the node list, and update the partition
         * status and state. This is done even on nodes that have explicitly
         * disabled the partition. This behaviour can be changed by setting
         * the RemoteEnableAction to LEAVE_DISABLED. Note that this only
         * controls the behaviour on the local node, it does not affect
         * partitions defined on remote nodes.
         * <p>
         * The default action is defined by {@link #DefaultRemoteEnableAction},
         */
        public enum RemoteEnableAction
        {
            /**
             * Allow remote nodes to enable the partition on the local node.
             */
            ENABLE_PARTITION,
            /**
             * Do not allow remote nodes to enable the partition on the
             * local node if it has been explicitly disabled.
             */
            LEAVE_DISABLED
        }

        /**
         * Default constructor.
         */
        public Properties()
        {
            m_restoreFromNode = null;
            m_forceReplication = false;
            m_objectsLockedPerTransaction = DefaultObjectsLockedPerTransaction;
            m_numberOfThreads = DefaultNumberOfThreads;
            m_sparseAudit = DefaultSparseAudit;
            m_replicaAudit = DefaultReplicaAudit;
            m_remoteEnableAction = DefaultRemoteEnableAction;
            m_broadcastUpdates = DEPartitionBroadcast.BroadcastDefault;
        }

        /**
         * Define the node that the partition should be restored from.
         * <p>
         * When this property is set, the partition defined on the given
         * remote node is loaded to the local node. This should be done
         * when restoring a node from a split-brain situation, where
         * <b>nodeName</b> is the node in the cluster where all objects
         * should be preserved, and the local node is the node being
         * restored. Any conflicts during restore will preserve the
         * objects on <b>nodeName</b>, and remove the conflicting objects
         * on the local node.
         * <p>
         * A restore is needed when multiple nodes are currently the active
         * node for a partition in a cluster due to a split-brain scenario.
         * In this case, the application needs to decide which active node
         * will be the node where the objects are preserved during a
         * restore. Note that the <b>nodeName</b> does not necessarily have
         * to be the node which becomes the partition's active node after the
         * restore completes.
         * <p>
         * The actual restore of the partition is done in the
         * enable() or enablePartitions() method when the JOIN_CLUSTER_RESTORE
         * EnableAction is used. If any other EnableAction is used,
         * object data isn't preserved, and no restoration of partition
         * objects is done.
         * <p>
         * If restoreFromNode isn't set after a split-brain scenario, the
         * runtime will perform a cluster wide broadcast to find the
         * current active node, and use that node to restore instances in
         * the partition. If multiple active nodes are found, the first
         * responder is chosen.
         * <p>
         * @param nodeName The remote node to use when restoring the
         *  partition's objects.
         *
         * @exception IllegalArgumentException
         * The nodeName was empty.
         * @see PartitionManager#definePartition(String,
         *              Partition.Properties, String, ReplicaNode [])
         * @see PartitionManager.EnableAction
         */
        public void restoreFromNode(String nodeName)
            throws IllegalArgumentException
        {
            if (nodeName.isEmpty())
            {
                throw new IllegalArgumentException(
                    "nodeName cannot be empty");
            }
            m_restoreFromNode = nodeName;
        }

        /**
         * Get the current restoreFromNode property value.
         * @return String containing value.
         */
        public final String getRestoreFromNode()
        {
            return m_restoreFromNode;
        }

        /**
         * Determine how objects are replicated during a migrate of
         * an object partition.
         * <p>
         * When set to true, a migrate() or
         * definePartition()/enablePartitions() will force the copy of
         * partitioned objects to all pre-existing replica nodes. The
         * default value for this property is false, objects are only
         * copied to new replicas as they are added since the objects
         * should already exist on the pre-existing replica nodes.
         * <p>
         * Normally, a migrate will skip the replication of objects to
         * pre-existing nodes in the partition's replica node list. This
         * allows applications to incrementally add replica nodes without
         * having to copy the objects to replicas that already exist in
         * the partition. However, if one or more replicas have gone
         * offline, or were not discovered when the partition was first
         * enabled, this property can be set to insure that objects are
         * pushed to all replicas in the cluster.
         * <p>
         * Warning: This is performance hostile, and should only be done
         * if the replica can't be manually taken offline and restored.
         * <p>
         * The value passed into definePartition() is stored and used in
         * failover. The value passed to migrate() overrides
         * the value passed to definePartition().
         *
         * @param enabled If true, force the copy of objects to all replicas
         *  when a migrate() or enablePartition() is executed.
         *
         */
        public void forceReplication(boolean enabled)
        {
            m_forceReplication = enabled;
        }

        /**
         * Get the current forceReplication property value.
         * @return boolean containing value.
         */
        public final boolean getForceReplication()
        {
            return m_forceReplication;
        }

        /**
         * Define the number of objects locked in a transaction when performing
         * a migrate() or update().
         * <p>
         * When distribution performs a migrate() or update(), or when it
         * performs a migrate() as part of failover, the work is done in
         * units of work defined by objectsLockedPerTransaction. This
         * allows applications to concurrently run while the work is in
         * progress, otherwise all partitioned instances would be locked
         * in the calling transaction, preventing application code from
         * establishing a write lock on instances on the active node, or
         * establishing a read lock on instances on replica nodes.
         * <p>
         * The default is defined by {@link #DefaultObjectsLockedPerTransaction}
         * <p>
         * If objectsLockedPerTransaction is set to zero, all instances will
         * be processed in the caller's transaction.
         * <p>
         * If the calling transaction has one or more partitioned
         * instances locked in the transaction,
         * objectsLockedPerTransaction is ignored, and all work is done
         * in the caller's transaction. To insure that migrate() or update()
         * minimizes the number of locks taken, they should be run in separate
         * transactions that have no partitioned instances locked. 
         * <p>
         * The value passed into definePartition() is stored and used in
         * failover. The value passed to migrate() and update() override
         * the value passed to definePartition().
         * 
         * @param objectsLockedPerTransaction Number of objects locked
         *  per transaction when sending data to remote nodes.
         * @exception IllegalArgumentException The objectsLockedPerTransaction
         *  value was negative.
         */
        public void setObjectsLockedPerTransaction(
            long objectsLockedPerTransaction)
        {
            if (objectsLockedPerTransaction < 0)
            {
                throw new IllegalArgumentException(
                    "invalid objectsLockedPerTransaction: " +
                    objectsLockedPerTransaction);
            }
            m_objectsLockedPerTransaction = objectsLockedPerTransaction;
        }

        /**
         * Get the current objectsLockedPerTransaction property value.
         * @return long containing value.
         */
        public final long getObjectsLockedPerTransaction()
        {
            return m_objectsLockedPerTransaction;
        }

        /**
         * Define the number of threads used when performing a migrate().
         * <p>
         * When distribution performs a migrate(), either explicitly or
         * as part of the enablePartitions() call, the work is done in
         * multiple threads in order to scale to the number of CPUs
         * available in the system. 
         * <p>
         * The default is defined by {@link #DefaultNumberOfThreads}
         * <p>
         * If the number of partitioned instances is less that
         * objectsLockedPerTransaction, only one thread will be used.
         * <p>
         * If objectsLockedPerTransaction is zero, or the calling
         * transaction has one or more partitioned instances locked in
         * the transaction, numberOfThreads is ignored, and all work is
         * done in the caller's transaction.
         * <p>
         * @param numberOfThreads Number of threads to use
         *  when sending data to remote nodes.
         * @exception IllegalArgumentException The numberOfThreads
         *  value was less than one.
         */
        public void setNumberOfThreads(long numberOfThreads)
        {
            if (numberOfThreads < 1)
            {
                throw new IllegalArgumentException(
                    "invalid numberOfThreads: " + numberOfThreads);
            }
            m_numberOfThreads = numberOfThreads;
        }

        /**
         * Get the current numberOfThreads property value.
         * @return long containing value.
         */
        public final long getNumberOfThreads()
        {
            return m_numberOfThreads;
        }

        /**
         * Set the sparseAudit property value.
         * <p>
         * The default is defined by {@link #DefaultSparseAudit}
         *
         * @param sparseAudit What auditing should be done.
         *
         */
        public void setSparseAudit(final SparseAudit sparseAudit)
        {  
            m_sparseAudit = sparseAudit;
        }

        /**
         * Get the current sparseAudit property value.
         * @return SparseAudit enum.
         */
        public final SparseAudit getSparseAudit()
        {
            return m_sparseAudit;
        }

        /**
         * Set the replicaAudit property value.
         * <p>
         * The default is defined by {@link #DefaultReplicaAudit}
         *
         * @param replicaAudit What auditing should be done.
         *
         */
        public void setReplicaAudit(final ReplicaAudit replicaAudit)
        {  
            m_replicaAudit = replicaAudit;
        }

        /**
         * Get the current replicaAudit property value.
         * @return ReplicaAudit enum.
         */
        public final ReplicaAudit getReplicaAudit()
        {
            return m_replicaAudit;
        }

        /**
         * Set the remoteEnableAction property value.
         * <p>
         * The default is defined by {@link #DefaultRemoteEnableAction}
         *
         * @param remoteEnableAction The action to take when remote nodes enable
         * a partition.
         *
         */
        public void setRemoteEnableAction(final RemoteEnableAction remoteEnableAction)
        {  
            m_remoteEnableAction = remoteEnableAction;
        }

        /**
         * Get the current remoteEnableAction property value.
         * @return RemoteEnableAction enum.
         */
        public final RemoteEnableAction getRemoteEnableAction()
        {
            return m_remoteEnableAction;
        }

        /**
         * Set the broadcastUpdates property value.
         * <p>
         * By default, this property is enabled. Updates to the partition's 
         * definition are broadcast to all nodes in the cluster so that
         * the partition definition is consistent across the cluster.
         * <p>
         * Warning: The ability to disable this property is only provided
         * to support the testing of multi-master scenarios.
         * Disabling this property for any other reason is not recommended.
         *
         * @param enabled If true, broadcast partition definition updates
         *  to all nodes in the cluster.
         */
        public void broadcastUpdates(boolean enabled)
        {
            m_broadcastUpdates = (enabled == true)
                ? DEPartitionBroadcast.BroadcastEnabled 
                : DEPartitionBroadcast.BroadcastDisabled;
        }

        /**
         * Get the current broadcastUpdates property value.
         * @return boolean containing value.
         */
        public final boolean getBroadcastUpdates()
        {
            // Map both BroadcastDefault and BroadcastEnabled to true.
            return (m_broadcastUpdates == DEPartitionBroadcast.BroadcastDisabled)
                ? false : true;
        }

        //
        // package private constructor
        //
        Properties(
            boolean forceReplication,
            long objectChunks,
            long numThreads,
            SparseAudit sparseAudit,
            ReplicaAudit replicaAudit,
            RemoteEnableAction remoteEnableAction,
            DEPartitionBroadcast broadcastUpdates)
        {
            m_restoreFromNode = null;
            m_forceReplication = forceReplication;
            m_objectsLockedPerTransaction = objectChunks;
            m_numberOfThreads = numThreads;
            m_sparseAudit = sparseAudit;
            m_replicaAudit = replicaAudit;
            m_remoteEnableAction = remoteEnableAction;
            m_broadcastUpdates = broadcastUpdates;
        }

        String m_restoreFromNode;
        boolean m_forceReplication;
        long m_objectsLockedPerTransaction;
        long m_numberOfThreads;
        SparseAudit m_sparseAudit;
        ReplicaAudit m_replicaAudit;
        RemoteEnableAction m_remoteEnableAction;
        DEPartitionBroadcast m_broadcastUpdates;
    }

    /**
     * Partition statistics
     * <p>
     * Partition statistics are captured for the local node only, any
     * operations done on remote nodes are not reflected in the stats. As
     * an example, if the local node is a replica, and the node executes
     * 10 object creates, the createCount will be set to 10. If objects
     * are created on the active node, or on other replica nodes, the
     * createCount will not be updated.
     * <p>
     * All partition statistics are reset each time an engine containing
     * the highavailability component is restarted. They do not persist
     * across engine restarts.
     */
    public final static class Statistics
    {
        /**
         * The number of instances created in this partition.
         */
        public long createCount;
        /**
         * The number of updates that have been applied to instance in this
         * partition.
         */
        public long updateCount;
        /**
         * The number of instances deleted in this partition.
         */
        public long deleteCount;
        /**
         * The number of synchronous create failures that occurred in
         * this partition.
         */
        public long createFailures;
        /**
         * The number of synchronous update failures that occurred in
         * this partition.
         */
        public long updateFailures;
        /**
         * The number of synchronous delete failures that occurred in
         * this partition.
         */
        public long deleteFailures;
        /**
         * The number of asynchronous create failures that occurred in
         * this partition.
         */
        public long asyncCreateFailures;
        /**
         * The number of asynchronous update failures that occurred in
         * this partition.
         */
        public long asyncUpdateFailures;
        /**
         * The number of asynchronous delete failures that occurred in
         * this partition.
         */
        public long asyncDeleteFailures;

        // package private constructor
        Statistics(DEPartitionStats stats)
        {
            this.createCount = stats.createCount;
            this.updateCount = stats.updateCount;
            this.deleteCount = stats.deleteCount;
            this.createFailures = stats.createFailures;
            this.updateFailures = stats.updateFailures;
            this.deleteFailures = stats.deleteFailures;
            this.asyncCreateFailures = stats.asyncCreateFailures;
            this.asyncUpdateFailures = stats.asyncUpdateFailures;
            this.asyncDeleteFailures = stats.asyncDeleteFailures;
        }
        // private constructor
        private Statistics() { }
    }

    /** Index to the active node in the nodes array */
    public final static int     ActiveNodeIndex = 0;

    /**
     * The default number of objects locked per transaction when
     * performing a migrate or update.
     */
    public final static long    DefaultObjectsLockedPerTransaction = 1000;

    /**
     * The default number of threads to use when performing a migrate.
     */
    public final static long    DefaultNumberOfThreads = 1;

    /**
     * The default audit used for sparse partitions.
     * The value is {@link Properties.SparseAudit#VERIFY_NODE_LIST}
     */
    public final static Properties.SparseAudit DefaultSparseAudit =
                                    Properties.SparseAudit.VERIFY_NODE_LIST;

    /**
     * The default audit used for replicas.
     * The value is {@link Properties.ReplicaAudit#WAIT_ACTIVE}
     */
    public final static Properties.ReplicaAudit DefaultReplicaAudit =
                                    Properties.ReplicaAudit.WAIT_ACTIVE;

    /**
     * The default action used for remote partition enables.
     * The value is {@link Properties.RemoteEnableAction#ENABLE_PARTITION}
     */
    public final static Properties.RemoteEnableAction DefaultRemoteEnableAction =
                                    Properties.RemoteEnableAction.ENABLE_PARTITION;

    /**
     * Get the partition name.
     * @return Partition name.
     */
    public final String getName()
    {
        return m_name;
    }

    /**
     * Get the active node for the partition.
     * @return Active node name.
     */
    public final String getActiveNode()
    {
        assert( m_nodes.length >= 1 );
        return m_nodes[ActiveNodeIndex];
    }

    /**
     * Get the replica node list for the partition.
     * @return Array of replica node names.
     */
    public final String [] getReplicaNodes()
    {
        assert( m_nodes.length >= 1 );
        String [] replicas = new String[m_nodes.length - 1];
        for (int i = 0; i < m_nodes.length - 1; i++)
        {
            replicas[i] = m_nodes[i + 1];
        }
        return replicas;
    }

    /**
     * Get the complete node list for the partition.
     * <p>
     * The node list is a String array of node names, with the active
     * node being at ActiveNodeIndex, followed by a prioritized list of
     * replica nodes.
     * @return Array of node names.
     */
    public final String [] getNodeList()
    {
        return m_nodes;
    }

    /**
     * Get the current state for the partition.
     * @return Current state of partition.
     */
    public final State getCurrentState()
    {
        return m_currentState;
    }

    /**
     * Get the current status for the partition.
     * @return Current status of partition.
     */
    public final Status getCurrentStatus()
    {
        return m_currentStatus;
    }

    /**
     * Get the ReplicaNode instance for the given replica node index.
     * @param idx Index into the ReplicaNode array.
     * @return ReplicaNode instance.
     * @exception IndexOutOfBoundsException
     *  The idx isn't valid.
     */
    public final ReplicaNode getReplicaNode(int idx)
        throws IndexOutOfBoundsException
    {
        // m_nodes[0] is active node, so we need to disallow -1
        if (idx < 0) throw new IndexOutOfBoundsException(Integer.toString(idx));
        return new ReplicaNode(m_nodes[idx + 1],
            PartitionManager.mapDEReplicationType(
                m_replicationTypes[idx + 1]));
    }

    /**
     * Get the ReplicaNode instance for the given replica node name.
     * @param nodeName Name of replica node.
     * @return ReplicaNode instance, or null if not found.
     * @exception IllegalArgumentException The nodeName was invalid.
     */
    public final ReplicaNode getReplicaNode(final String nodeName)
        throws IllegalArgumentException
    {
        if (nodeName.isEmpty())
        {
            throw new IllegalArgumentException("nodeName cannot be empty");
        }
        for (int idx = 1; idx < m_nodes.length; idx++)
        {
            if (m_nodes[idx].equals(nodeName))
            {
                return getReplicaNode(idx - 1);
            }
        }
        return null;
    }

    /**
     * Get the last time the state for the partition was updated
     * @return Date of last state change for partition.
     */
    public final Date getLastStateChangeTime()
    {
        return m_lastUpdated;
    }

    /**
     * Get the properties currently defined for the partition.
     * <p>
     * The restoreFromNode property is not stored in the runtime since it
     * has no meaning outside of the initial definePartition(). So this method
     * will return a Properties instance with a null restoreFromNode value.
     *
     * @return Properties instance for partition.
     */
    public final Properties getProperties()
    {
        return new Properties(m_forceReplication, m_objectChunks,
                m_numThreads, m_sparseAudit, m_replicaAudit,
                m_remoteEnableAction, m_broadcastUpdates);
    }

    /**
     * Get the statistics for the partition.
     *
     * @return A Statistics instance for the partition.
     */
    public final Statistics getStatistics()
    {
        DEPartitionManager pm = new DEPartitionManager();
        return new Statistics(pm.getPartitionStats(m_name));
    }

    /**
     * Clear the statistics for the partition.
     */
    public final void clearStatistics()
    {
        DEPartitionManager pm = new DEPartitionManager();
        pm.clearPartitionStats(m_name);
    }

    /**
     * Use {@link #update(Properties)} instead.
     */
    @Deprecated
    public final void update() throws NodeMismatch, NotActiveNode
    {
        update(null);
    }

    /**
     * Update the partition.
     * <p>
     * This method is used to re-partition all instances that exist in a
     * partition. For each instance in the partition, the {@link
     * PartitionMapper} defined for that type is accessed, and the
     * instance re-assigned to the partition that the {@link
     * PartitionMapper#getPartition} method returns. The update() method
     * must be called on the current active node for the partition.
     * <p>
     * To split a partition, the applications should create the new
     * partition, install a new PartitionMapper for all types managed by
     * the partition, and call the update() method for the existing
     * partition.
     * <p>
     * To merge two or more partitions, the applications should install a
     * new PartitionMapper for all types managed by the partition(s), and
     * call the update() method for all the partitions that need to
     * be merged.
     * <p>
     * It is important that all partitions returned when executing the
     * PartitionMapper's getPartition() call for a type all contain
     * identical node lists. If the partition returned contains a
     * different node list than this partition, a NodeMismatch exception
     * is thrown, and the update() terminates. To fix this, perform a
     * migrate() on all partitions that will be split or merged to insure
     * that they have identical node lists before performing an update().
     *
     * @param partitionProperties Optional properties for the partition.
     *  If null, the default property values are used.
     * 
     * @exception NodeMismatch
     *  The re-partitioning of instances was done using partitions that
     *  have different node lists.
     * @exception NotActiveNode
     *  The current node is not the active node for the partition.
     */
    public final void update(final Properties partitionProperties)
            throws NodeMismatch, NotActiveNode
    {
        DEProperties props = copyDEProperties(partitionProperties);

        PartitionManager.updatePartition(m_name, props);
    }

    /**
     * Use {@link #migrate(Properties, String, ReplicaNode [])} instead.
     * @param nodes An ordered list of nodes for the partition.
     */
    @Deprecated
    public final void migrate(String [] nodes)
        throws NotActiveNode, IllegalArgumentException
    {
        PartitionManager.validateNodeList(nodes);
        DEReplicationType [] drl = new DEReplicationType[nodes.length];
        for (int i = 0; i < drl.length; i++)
        {
            drl[i] = DEReplicationType.DataSynchronous;
        }

        DEProperties props = defaultDEProperties();

        PartitionManager.migratePartition(m_name, props, nodes, drl);
    }

    /**
     * Migrate the partition.
     * <p>
     * This method is used to migrate all instances that exist in a
     * partition. For each instance in the partition, the object is
     * migrated as needed to all nodes in the replicas array, and to the
     * new activeNode if it is different from the current active node. This
     * method must be called on the currently active node for the partition.
     * <p>
     * Any properties passed in only apply to the current migrate() command,
     * properties passed into definePartition() are saved and used during
     * failover. 
     * 
     * @param partitionProperties Optional properties for the partition.
     *  If null, the default property values are used.
     * @param activeNode Active node after migrate completes.
     * @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 exists.
     *
     * @exception NotActiveNode
     *  The current node is not the active node for the partition.
     * @exception IllegalArgumentException
     * The activeNode or replica array was invalid.
     */
    public final void migrate(
        final Properties partitionProperties,
        String activeNode,
        ReplicaNode [] replicas)
            throws NotActiveNode, IllegalArgumentException
    {
        PartitionManager.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 < nodes.length );
                nodes[i + 1] = replicas[i].nodeName;
            }
        }
        PartitionManager.validateNodeList(nodes);

        DEReplicationType [] drl = new DEReplicationType[nodes.length];
        drl[0] = DEReplicationType.DataSynchronous;
        if (replicas != null)
        {
            for (int i = 0; i < replicas.length; i++)
            {
                assert( i + 1 < nodes.length );
                drl[i + 1] = PartitionManager.mapReplicationType(
                    replicas[i].replicationType);
            }
        }
        boolean forceReplication = (partitionProperties != null)
            ? partitionProperties.m_forceReplication : false;

        DEProperties props = copyDEProperties(partitionProperties);

        PartitionManager.migratePartition(m_name, props, nodes, drl);
    }

    /**
     * Enable high availability for this partition.
     * <p>
     * This method is used to enable this partition. If the partition is
     * already in the <b>ACTIVE</b> state, no action is taken.
     * This method blocks until all required object migration
     * is complete.
     *
     * @param enableAction The action to take when enabling partitions.
     *
     * @exception ResourceUnavailableException
     *  The minimum number of nodes needed to establish a quorum has not been
     *  seen, or the partition has not been defined.
     * @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.
     * @see PartitionManager#enablePartitions(PartitionManager.EnableAction)
     */
    public final void enable(
        final PartitionManager.EnableAction enableAction)
            throws NodeMismatch, ResourceUnavailableException
    {
        DEPartitionManager pm = new DEPartitionManager();
        try
        {
            pm.enablePartition(m_name,
                PartitionManager.mapEnable(enableAction));
        }
        catch (DENodeMismatch ex)
        {
            throw new NodeMismatch(ex.errMsg);
        }
    }

    /**
     * Disable high availability for this partition.
     * <p>
     * This method is used to remove the local node from the node list 
     * of this partition, and move the partition state to <b>UNAVAILABLE</b>.
     * If called multiple times, the additional calls have no effect.
     * <p>
     * If the local node does not exist in the partition, no action is taken.
     *
     * @param disableAction  The action to take when disabling partitions.
     *
     * @exception ResourceUnavailableException
     *  The partition would be in the <b>UNAVAILABLE</b> state after the
     *  disable executes, not thrown if <b>LEAVE_CLUSTER_FORCE</b> is used.
     * @see PartitionManager#disablePartitions(PartitionManager.DisableAction)
     */
    public final void disable(
        final PartitionManager.DisableAction disableAction)
            throws ResourceUnavailableException
    {
        DEPartitionManager pm = new DEPartitionManager();
        pm.disablePartition(m_name, PartitionManager.mapDisable(disableAction));
    }

    /**
     * Associated a notifier with the partition.
     * <p>
     * This method is used to associate a user defined {@link PartitionNotifier}
     * instance with a partition.
     * <p>
     * PartitionNotifier instances are local to a node, this method
     * should be executed on all nodes which need to determine if a
     * partition state change occurs.
     * <p>
     * The same PartitionNotifier instance can be associated with multiple
     * partitions. Multiple PartitionNotifier instances can be installed for
     * a given Partition. When a state change occurs, all instances are
     * executed, no order is guaranteed when executing the notifiers.
     * To remove a notifier, the PartitionNotifier instance should be deleted.
     * 
     * @param partitionNotifier User defined notifier instance.
     *
     */
    public final void setNotifier(PartitionNotifier partitionNotifier)
    {
        PartitionManager.setNotifier(m_name, partitionNotifier);
    }

    /**
     * Returns the number of instances in this partition.
     * <p>
     * This method does not establish transaction locks on the
     * partition or any of the individual objects in the partition.
     * This means the number returned by cardinality() may 
     * change even as observed within a single transaction.
     * <p>
     * Warning: This method requires a scan of all objects for each class in 
     * the partition to determine the count, which is computationally
     * expensive.
     *
     * @return Number of instances found.
     */
    public final long cardinality()
    {
        DEPartitionManager pm = new DEPartitionManager();
        return pm.getCardinality(m_name);
    }

    /** The name of the partition */
    private final String         m_name;

    /** The list of nodes for a partition */
    private final String []      m_nodes;

    /** The list of nodes for a partition */
    private final DEReplicationType [] m_replicationTypes;

    /** The current state of the partition */
    private final State          m_currentState;

    /** The current status of the partition */
    private final Status        m_currentStatus;

    /** The time it was updated */
    private final Date          m_lastUpdated;

    /** stored property */
    private final boolean       m_forceReplication;

    /** stored property */
    private final long          m_objectChunks;

    /** stored property */
    private final long          m_numThreads;

    /** stored property */
    private final Properties.SparseAudit m_sparseAudit;

    /** stored property */
    private final Properties.ReplicaAudit m_replicaAudit;

    /** stored property */
    private final Properties.RemoteEnableAction m_remoteEnableAction;

    /** stored property */
    private final DEPartitionBroadcast   m_broadcastUpdates;

    //
    // FIX THIS: Add stats attributes...
    //

    //
    // Package private constructor
    //
    Partition(
        String name,
        String [] nodes,
        DEReplicationType [] replicationTypes,
        State currentState,
        Status currentStatus,
        Date lastUpdated,
        boolean forceReplication,
        long objectChunks,
        long numThreads,
        Properties.SparseAudit sparseAudit,
        Properties.ReplicaAudit replicaAudit,
        Properties.RemoteEnableAction remoteEnableAction,
        DEPartitionBroadcast broadcastUpdates)
    {
        this.m_name = name;
        this.m_nodes = nodes;
        this.m_replicationTypes = replicationTypes;
        this.m_currentState = currentState;
        this.m_currentStatus = currentStatus;
        this.m_lastUpdated = lastUpdated;
        this.m_forceReplication = forceReplication;
        this.m_objectChunks = objectChunks;
        this.m_numThreads = numThreads;
        this.m_sparseAudit = sparseAudit;
        this.m_replicaAudit = replicaAudit;
        this.m_remoteEnableAction = remoteEnableAction;
        this.m_broadcastUpdates = broadcastUpdates;
    }
    //
    // Never should be called
    //
    private Partition()
    {
        this.m_name = null;
        this.m_nodes = null;
        this.m_replicationTypes = null;
        this.m_currentState = null;
        this.m_currentStatus = null;
        this.m_lastUpdated = null;
        this.m_forceReplication = false;
        this.m_objectChunks = 0;
        this.m_numThreads = 0;
        this.m_sparseAudit = null;
        this.m_replicaAudit = null;
        this.m_remoteEnableAction = null;
        this.m_broadcastUpdates = DEPartitionBroadcast.BroadcastDefault;
    }

    //
    // Copy properties to DEProperties.
    //
    static DEProperties copyDEProperties(Properties partitionProperties)
    {
        if (partitionProperties == null)
        {
            return defaultDEProperties();
        }

        DEProperties props = new DEProperties();

        props.forceReplication = partitionProperties.m_forceReplication;
        props.objectChunks =
            partitionProperties.m_objectsLockedPerTransaction;
        props.numThreads = partitionProperties.m_numberOfThreads;
        props.sparseAudit =
            PartitionManager.mapSparseAudit(partitionProperties.m_sparseAudit);
        props.replicaAudit =
            PartitionManager.mapReplicaAudit(partitionProperties.m_replicaAudit);
        props.remoteEnableAction =
            PartitionManager.mapRemoteEnableAction(partitionProperties.m_remoteEnableAction);
        props.restoreFromNode = partitionProperties.m_restoreFromNode;
        props.broadcastUpdates = partitionProperties.m_broadcastUpdates;

        return props;
    }

    //
    // Allocate and initialize default DEProperties.
    //
    static DEProperties defaultDEProperties()
    {
        DEProperties props = new DEProperties();

        //
        // Note: This needs to match the Properties() constructor.
        //
        props.forceReplication = false;
        props.objectChunks = DefaultObjectsLockedPerTransaction;
        props.numThreads = DefaultNumberOfThreads;
        props.sparseAudit = DESparsePartitionAudit.VerifyNodeList;
        props.replicaAudit = DEReplicaAudit.ReplicaIgnore;
        props.remoteEnableAction = DERemoteEnableAction.EnablePartition;
        props.restoreFromNode = null;
        props.broadcastUpdates = DEPartitionBroadcast.BroadcastDefault;

        return props;
    }
}