QPID-4006: add support for using BDB HA to form an active-passive cluster for persistent messaging

- Includes support for setting BDB configuration parameters via the store configuration, both for the existing store and the new HA variant.
- Removes the MessageStoreFactory and reverts store configuration to historical values.

Applied patch from Keith Wall, Andrew MacBean <andymacbean@gmail.com>, Oleksandr Rudyy <orudyy@gmail.com>, Philip Harvey <phil@philharveyonline.com>, and myself.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1339728 13f79535-47bb-0310-9956-ffa450edef68

17 files changed, 2832 insertions, 341 deletions

diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/AbstractBDBMessageStore.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/AbstractBDBMessageStore.java
index a9ae4eb16d..789d5714c8 100644
--- a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/AbstractBDBMessageStore.java
+++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/AbstractBDBMessageStore.java
@@ -33,11 +33,12 @@ import com.sleepycat.je.EnvironmentConfig;
 import com.sleepycat.je.LockConflictException;
 import com.sleepycat.je.LockMode;
 import com.sleepycat.je.OperationStatus;
-import com.sleepycat.je.TransactionConfig;
 import java.io.File;
 import java.lang.ref.SoftReference;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
@@ -45,6 +46,7 @@ import java.util.Random;
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicLong;
 import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.ConfigurationException;
 import org.apache.log4j.Logger;
 import org.apache.qpid.AMQStoreException;
 import org.apache.qpid.framing.FieldTable;
@@ -128,8 +130,6 @@ public abstract class AbstractBDBMessageStore implements MessageStore
 
     protected final StateManager _stateManager;
 
-    protected TransactionConfig _transactionConfig = new TransactionConfig();
-
     private MessageStoreRecoveryHandler _messageRecoveryHandler;
 
     private TransactionLogRecoveryHandler _tlogRecoveryHandler;
@@ -146,6 +146,8 @@ public abstract class AbstractBDBMessageStore implements MessageStore
 
     private ConfiguredObjectHelper _configuredObjectHelper = new ConfiguredObjectHelper();
 
+    private Map<String, String> _envConfigMap;
+
     public AbstractBDBMessageStore()
     {
         _stateManager = new StateManager(_eventManager);
@@ -161,7 +163,7 @@ public abstract class AbstractBDBMessageStore implements MessageStore
                                      ConfigurationRecoveryHandler recoveryHandler,
                                      Configuration storeConfiguration) throws Exception
     {
-        _stateManager.attainState(State.CONFIGURING);
+        _stateManager.attainState(State.INITIALISING);
 
         _configRecoveryHandler = recoveryHandler;
 
@@ -179,12 +181,12 @@ public abstract class AbstractBDBMessageStore implements MessageStore
         _messageRecoveryHandler = messageRecoveryHandler;
         _tlogRecoveryHandler = tlogRecoveryHandler;
 
-        _stateManager.attainState(State.CONFIGURED);
+        _stateManager.attainState(State.INITIALISED);
     }
 
-    public void activate() throws Exception
+    public synchronized void activate() throws Exception
     {
-        _stateManager.attainState(State.RECOVERING);
+        _stateManager.attainState(State.ACTIVATING);
 
         recoverConfig(_configRecoveryHandler);
         recoverMessages(_messageRecoveryHandler);
@@ -231,11 +233,30 @@ public abstract class AbstractBDBMessageStore implements MessageStore
 
         _storeLocation = storeLocation;
 
+        _envConfigMap = getConfigMap(storeConfig, "envConfig");
+
         LOGGER.info("Configuring BDB message store");
 
         setupStore(environmentPath, name);
     }
 
+    protected Map<String,String> getConfigMap(Configuration config, String prefix) throws ConfigurationException
+    {
+        final List<Object> argumentNames = config.getList(prefix + ".name");
+        final List<Object> argumentValues = config.getList(prefix + ".value");
+        final Map<String,String> attributes = new HashMap<String,String>(argumentNames.size());
+
+        for (int i = 0; i < argumentNames.size(); i++)
+        {
+            final String argName = argumentNames.get(i).toString();
+            final String argValue = argumentValues.get(i).toString();
+
+            attributes.put(argName, argValue);
+        }
+
+        return Collections.unmodifiableMap(attributes);
+    }
+
     @Override
     public String getStoreLocation()
     {
@@ -251,9 +272,9 @@ public abstract class AbstractBDBMessageStore implements MessageStore
      */
     void startWithNoRecover() throws AMQStoreException
     {
-        _stateManager.attainState(State.CONFIGURING);
-        _stateManager.attainState(State.CONFIGURED);
-        _stateManager.attainState(State.RECOVERING);
+        _stateManager.attainState(State.INITIALISING);
+        _stateManager.attainState(State.INITIALISED);
+        _stateManager.attainState(State.ACTIVATING);
         _stateManager.attainState(State.ACTIVE);
     }
 
@@ -268,51 +289,7 @@ public abstract class AbstractBDBMessageStore implements MessageStore
         _totalStoreSize = getSizeOnDisk();
     }
 
-    protected Environment createEnvironment(File environmentPath) throws DatabaseException
-    {
-        LOGGER.info("BDB message store using environment path " + environmentPath.getAbsolutePath());
-        EnvironmentConfig envConfig = new EnvironmentConfig();
-        // This is what allows the creation of the store if it does not already exist.
-        envConfig.setAllowCreate(true);
-        envConfig.setTransactional(true);
-        envConfig.setConfigParam("je.lock.nLockTables", "7");
-
-        // Added to help diagnosis of Deadlock issue
-        // http://www.oracle.com/technology/products/berkeley-db/faq/je_faq.html#23
-        if (Boolean.getBoolean("qpid.bdb.lock.debug"))
-        {
-            envConfig.setConfigParam("je.txn.deadlockStackTrace", "true");
-            envConfig.setConfigParam("je.txn.dumpLocks", "true");
-        }
-
-        // Set transaction mode
-        _transactionConfig.setReadCommitted(true);
-
-        //This prevents background threads running which will potentially update the store.
-        envConfig.setReadOnly(false);
-        try
-        {
-            return new Environment(environmentPath, envConfig);
-        }
-        catch (DatabaseException de)
-        {
-            if (de.getMessage().contains("Environment.setAllowCreate is false"))
-            {
-                //Allow the creation this time
-                envConfig.setAllowCreate(true);
-                if (_environment != null )
-                {
-                    _environment.cleanLog();
-                    _environment.close();
-                }
-                return new Environment(environmentPath, envConfig);
-            }
-            else
-            {
-                throw de;
-            }
-        }
-    }
+    protected abstract Environment createEnvironment(File environmentPath) throws DatabaseException;
 
     public Environment getEnvironment()
     {
@@ -352,14 +329,9 @@ public abstract class AbstractBDBMessageStore implements MessageStore
      */
     public void close() throws Exception
     {
-        if (_stateManager.isInState(State.ACTIVE) || _stateManager.isInState(State.QUIESCED))
-        {
-            _stateManager.stateTransition(State.ACTIVE, State.CLOSING);
-
-            closeInternal();
-
-            _stateManager.stateTransition(State.CLOSING, State.CLOSED);
-        }
+        _stateManager.attainState(State.CLOSING);
+        closeInternal();
+        _stateManager.attainState(State.CLOSED);
     }
 
     protected void closeInternal() throws Exception
@@ -1504,6 +1476,7 @@ public abstract class AbstractBDBMessageStore implements MessageStore
         return true;
     }
 
+    @SuppressWarnings("unchecked")
     public <T extends StorableMessageMetaData> StoredMessage<T> addMessage(T metaData)
     {
         if(metaData.isPersistent())
@@ -1914,4 +1887,25 @@ public abstract class AbstractBDBMessageStore implements MessageStore
     {
         return _persistentSizeHighThreshold;
     }
+
+    private void setEnvironmentConfigProperties(EnvironmentConfig envConfig)
+    {
+        for (Map.Entry<String, String> configItem : _envConfigMap.entrySet())
+        {
+            LOGGER.debug("Setting EnvironmentConfig key " + configItem.getKey() + " to '" + configItem.getValue() + "'");
+            envConfig.setConfigParam(configItem.getKey(), configItem.getValue());
+        }
+    }
+
+    protected EnvironmentConfig createEnvironmentConfig()
+    {
+        EnvironmentConfig envConfig = new EnvironmentConfig();
+        envConfig.setAllowCreate(true);
+        envConfig.setTransactional(true);
+        envConfig.setConfigParam(EnvironmentConfig.LOCK_N_LOCK_TABLES, "7");
+    
+        setEnvironmentConfigProperties(envConfig);
+    
+        return envConfig;
+    }
 }
diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBHAMessageStore.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBHAMessageStore.java
new file mode 100644
index 0000000000..f887c8ce36
--- /dev/null
+++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBHAMessageStore.java
@@ -0,0 +1,490 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.store.berkeleydb;
+
+import java.io.File;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.log4j.Logger;
+import org.apache.qpid.AMQStoreException;
+import org.apache.qpid.server.logging.RootMessageLogger;
+import org.apache.qpid.server.logging.actors.AbstractActor;
+import org.apache.qpid.server.logging.actors.CurrentActor;
+import org.apache.qpid.server.store.HAMessageStore;
+import org.apache.qpid.server.store.MessageStore;
+import org.apache.qpid.server.store.MessageStoreRecoveryHandler;
+import org.apache.qpid.server.store.State;
+import org.apache.qpid.server.store.StoreFuture;
+import org.apache.qpid.server.store.TransactionLogRecoveryHandler;
+
+import com.sleepycat.je.DatabaseException;
+import com.sleepycat.je.Durability;
+import com.sleepycat.je.Durability.ReplicaAckPolicy;
+import com.sleepycat.je.Durability.SyncPolicy;
+import com.sleepycat.je.Environment;
+import com.sleepycat.je.EnvironmentConfig;
+import com.sleepycat.je.OperationFailureException;
+import com.sleepycat.je.Transaction;
+import com.sleepycat.je.rep.InsufficientLogException;
+import com.sleepycat.je.rep.NetworkRestore;
+import com.sleepycat.je.rep.NetworkRestoreConfig;
+import com.sleepycat.je.rep.ReplicatedEnvironment;
+import com.sleepycat.je.rep.ReplicationConfig;
+import com.sleepycat.je.rep.ReplicationMutableConfig;
+import com.sleepycat.je.rep.ReplicationNode;
+import com.sleepycat.je.rep.StateChangeEvent;
+import com.sleepycat.je.rep.StateChangeListener;
+import com.sleepycat.je.rep.util.ReplicationGroupAdmin;
+
+public class BDBHAMessageStore extends AbstractBDBMessageStore implements HAMessageStore
+{
+    private static final String MUTLI_SYNC = "MUTLI_SYNC";
+    private static final String DEFAULT_REPLICATION_POLICY =
+        MUTLI_SYNC + "," + SyncPolicy.NO_SYNC.name() + "," + ReplicaAckPolicy.SIMPLE_MAJORITY.name();
+
+    private static final Logger LOGGER = Logger.getLogger(BDBHAMessageStore.class);
+
+    private String _groupName;
+    private String _nodeName;
+    private String _nodeHostPort;
+    private String _helperHostPort;
+    private String _replicationPolicy;
+    private Durability _replicationDurability;
+
+    private String _name;
+
+    private BDBHAMessageStoreManagerMBean _managedObject;
+
+    public static final String GRP_MEM_COL_NODE_HOST_PORT = "NodeHostPort";
+
+    public static final String GRP_MEM_COL_NODE_NAME = "NodeName";
+
+    private CommitThreadWrapper _commitThreadWrapper;
+    private boolean _localMultiSyncCommits;
+    private boolean _autoDesignatedPrimary;
+    private Map<String, String> _repConfigMap;
+
+    @Override
+    public void configure(String name, Configuration storeConfig) throws Exception
+    {
+        //Mandatory configuration
+        _groupName = getValidatedPropertyFromConfig("highAvailability.groupName", storeConfig);
+        _nodeName = getValidatedPropertyFromConfig("highAvailability.nodeName", storeConfig);
+        _nodeHostPort = getValidatedPropertyFromConfig("highAvailability.nodeHostPort", storeConfig);
+        _helperHostPort = getValidatedPropertyFromConfig("highAvailability.helperHostPort", storeConfig);
+        _name = name;
+
+        //Optional configuration
+        _replicationPolicy = storeConfig.getString("highAvailability.replicationPolicy", DEFAULT_REPLICATION_POLICY).trim();
+        _autoDesignatedPrimary = storeConfig.getBoolean("highAvailability.autoDesignatedPrimary", Boolean.TRUE);
+
+        if(_replicationPolicy.startsWith(MUTLI_SYNC))
+        {
+            _replicationDurability = Durability.parse(_replicationPolicy.replaceFirst(MUTLI_SYNC, SyncPolicy.SYNC.name()));
+            _localMultiSyncCommits = true;
+        }
+        else
+        {
+            _replicationDurability = Durability.parse(_replicationPolicy);
+            _localMultiSyncCommits = false;
+        }
+
+        _repConfigMap = getConfigMap(storeConfig, "repConfig");
+
+        _managedObject = new BDBHAMessageStoreManagerMBean(this);
+        _managedObject.register();
+
+        super.configure(name, storeConfig);
+    }
+
+    @Override
+    protected void setupStore(File storePath, String name) throws DatabaseException, AMQStoreException
+    {
+        super.setupStore(storePath, name);
+
+        if(_localMultiSyncCommits)
+        {
+            _commitThreadWrapper = new CommitThreadWrapper("Commit-Thread-" + name, getEnvironment());
+            _commitThreadWrapper.startCommitThread();
+        }
+    }
+
+    private String getValidatedPropertyFromConfig(String key, Configuration config) throws ConfigurationException
+    {
+        if (!config.containsKey(key))
+        {
+            throw new ConfigurationException("BDB HA configuration key not found. Please specify configuration key with XPath: "
+                                                + key.replace('.', '/'));
+        }
+        return config.getString(key);
+    }
+
+    @Override
+    protected Environment createEnvironment(File environmentPath) throws DatabaseException
+    {
+        if (LOGGER.isInfoEnabled())
+        {
+            LOGGER.info("Environment path " + environmentPath.getAbsolutePath());
+            LOGGER.info("Group name " + _groupName);
+            LOGGER.info("Node name " + _nodeName);
+            LOGGER.info("Node host port " + _nodeHostPort);
+            LOGGER.info("Helper host port " + _helperHostPort);
+            LOGGER.info("Replication policy " + _replicationPolicy);
+        }
+
+        final ReplicationConfig replicationConfig = new ReplicationConfig(_groupName, _nodeName, _nodeHostPort);
+
+        replicationConfig.setHelperHosts(_helperHostPort);
+        setReplicationConfigProperties(replicationConfig);
+
+        final EnvironmentConfig envConfig = createEnvironmentConfig();
+        envConfig.setDurability(_replicationDurability);
+
+        ReplicatedEnvironment replicatedEnvironment = null;
+        try
+        {
+            replicatedEnvironment = new ReplicatedEnvironment(environmentPath, replicationConfig, envConfig);
+        }
+        catch (final InsufficientLogException ile)
+        {
+            LOGGER.info("InsufficientLogException thrown and so full network restore required", ile);
+            NetworkRestore restore = new NetworkRestore();
+            NetworkRestoreConfig config = new NetworkRestoreConfig();
+            config.setRetainLogFiles(false);
+            restore.execute(ile, config);
+            replicatedEnvironment = new ReplicatedEnvironment(environmentPath, replicationConfig, envConfig);
+        }
+
+        return replicatedEnvironment;
+    }
+
+    @Override
+    public void configureMessageStore(String name, MessageStoreRecoveryHandler messageRecoveryHandler,
+                                        TransactionLogRecoveryHandler tlogRecoveryHandler,
+                                        Configuration config) throws Exception
+    {
+        super.configureMessageStore(name, messageRecoveryHandler, tlogRecoveryHandler, config);
+
+        final ReplicatedEnvironment replicatedEnvironment = getReplicatedEnvironment();
+
+        replicatedEnvironment.setStateChangeListener(new BDBHAMessageStoreStateChangeListener());
+    }
+
+    @Override
+    public synchronized void passivate()
+    {
+        if (_stateManager.isNotInState(State.INITIALISED))
+        {
+            LOGGER.debug("Store becoming passive");
+            _stateManager.attainState(State.INITIALISED);
+        }
+    }
+
+    @Override
+    protected void closeInternal() throws Exception
+    {
+        try
+        {
+            if(_localMultiSyncCommits)
+            {
+                _commitThreadWrapper.stopCommitThread();
+            }
+            super.closeInternal();
+        }
+        finally
+        {
+            if (_managedObject != null)
+            {
+                _managedObject.unregister();
+            }
+        }
+    }
+
+    @Override
+    protected StoreFuture commit(Transaction tx, boolean syncCommit) throws DatabaseException
+    {
+        // Using commit() instead of commitNoSync() for the HA store to allow
+        // the HA durability configuration to influence resulting behaviour.
+        tx.commit();
+
+
+        if(_localMultiSyncCommits)
+        {
+            return _commitThreadWrapper.commit(tx, syncCommit);
+        }
+        else
+        {
+            return StoreFuture.IMMEDIATE_FUTURE;
+        }
+    }
+
+    public String getName()
+    {
+        return _name;
+    }
+
+    public String getGroupName()
+    {
+        return _groupName;
+    }
+
+    public String getNodeName()
+    {
+        return _nodeName;
+    }
+
+    public String getNodeHostPort()
+    {
+        return _nodeHostPort;
+    }
+
+    public String getHelperHostPort()
+    {
+        return _helperHostPort;
+    }
+
+    public String getReplicationPolicy()
+    {
+        return _replicationPolicy;
+    }
+
+    public String getNodeState()
+    {
+        ReplicatedEnvironment.State state = getReplicatedEnvironment().getState();
+        return state.toString();
+    }
+
+    public Boolean isDesignatedPrimary()
+    {
+        return getReplicatedEnvironment().getRepMutableConfig().getDesignatedPrimary();
+    }
+
+    public List<Map<String, String>> getGroupMembers()
+    {
+        List<Map<String, String>> members = new ArrayList<Map<String,String>>();
+
+        for (ReplicationNode node : getReplicatedEnvironment().getGroup().getNodes())
+        {
+            Map<String, String> nodeMap = new HashMap<String, String>();
+            nodeMap.put(BDBHAMessageStore.GRP_MEM_COL_NODE_NAME, node.getName());
+            nodeMap.put(BDBHAMessageStore.GRP_MEM_COL_NODE_HOST_PORT, node.getHostName() + ":" + node.getPort());
+            members.add(nodeMap);
+        }
+
+        return members;
+    }
+
+    public void removeNodeFromGroup(String nodeName)  throws AMQStoreException
+    {
+        try
+        {
+            createReplicationGroupAdmin().removeMember(nodeName);
+        }
+        catch (OperationFailureException ofe)
+        {
+            throw new AMQStoreException("Failed to remove '" + nodeName + "' from group. " + ofe.getMessage(), ofe);
+        }
+        catch (DatabaseException e)
+        {
+            throw new AMQStoreException("Failed to remove '" + nodeName + "' from group. " + e.getMessage(), e);
+        }
+    }
+
+    public void setDesignatedPrimary(boolean isPrimary) throws AMQStoreException
+    {
+        try
+        {
+            final ReplicatedEnvironment replicatedEnvironment = getReplicatedEnvironment();
+            synchronized(replicatedEnvironment)
+            {
+                final ReplicationMutableConfig oldConfig = replicatedEnvironment.getRepMutableConfig();
+                final ReplicationMutableConfig newConfig = oldConfig.setDesignatedPrimary(isPrimary);
+                replicatedEnvironment.setRepMutableConfig(newConfig);
+            }
+
+            LOGGER.info("Node " + _nodeName + " successfully set as designated primary for group");
+        }
+        catch (DatabaseException e)
+        {
+            throw new AMQStoreException("Failed to set '" + _nodeName + "' as designated primary for group. " + e.getMessage(), e);
+        }
+    }
+
+    ReplicatedEnvironment getReplicatedEnvironment()
+    {
+        return (ReplicatedEnvironment)getEnvironment();
+    }
+
+    public void updateAddress(String nodeName, String newHostName, int newPort) throws AMQStoreException
+    {
+        try
+        {
+            createReplicationGroupAdmin().updateAddress(nodeName, newHostName, newPort);
+        }
+        catch (OperationFailureException ofe)
+        {
+            throw new AMQStoreException("Failed to update address for '" + nodeName +
+                                        "' with new host " + newHostName + " and new port " + newPort + ". " + ofe.getMessage(), ofe);
+        }
+        catch (DatabaseException e)
+        {
+            throw new AMQStoreException("Failed to update address for '" + nodeName +
+                                        "' with new host " + newHostName + " and new port " + newPort + ". " + e.getMessage(),  e);
+        }
+    }
+
+    private ReplicationGroupAdmin createReplicationGroupAdmin()
+    {
+        final Set<InetSocketAddress> helpers = new HashSet<InetSocketAddress>();
+        helpers.addAll(getReplicatedEnvironment().getRepConfig().getHelperSockets());
+
+        final ReplicationConfig repConfig = getReplicatedEnvironment().getRepConfig();
+        helpers.add(InetSocketAddress.createUnresolved(repConfig.getNodeHostname(), repConfig.getNodePort()));
+
+        return new ReplicationGroupAdmin(_groupName, helpers);
+    }
+
+    private class BDBHAMessageStoreStateChangeListener implements StateChangeListener
+    {
+        private final Executor _executor = Executors.newSingleThreadExecutor();
+
+        @Override
+        public void stateChange(StateChangeEvent stateChangeEvent) throws RuntimeException
+        {
+            com.sleepycat.je.rep.ReplicatedEnvironment.State state = stateChangeEvent.getState();
+
+            LOGGER.info("Received BDB event indicating transition to state " + state);
+
+            switch (state)
+            {
+            case MASTER:
+                activateStoreAsync();
+                break;
+            case REPLICA:
+                passivateStore();
+                break;
+            case DETACHED:
+                LOGGER.error("BDB replicated node in detached state, therefore passivating.");
+                passivateStore();
+                break;
+            case UNKNOWN:
+                LOGGER.warn("BDB replicated node in unknown state (hopefully temporarily)");
+                break;
+            default:
+                LOGGER.error("Unexpected state change: " + state);
+                throw new IllegalStateException("Unexpected state change: " + state);
+            }
+        }
+
+        /** synchronously calls passivate. This is acceptable because {@link HAMessageStore#passivate()} is expected to be fast */
+        private void passivateStore()
+        {
+            try
+            {
+                passivate();
+            }
+            catch(Exception e)
+            {
+                LOGGER.error("Unable to passivate", e);
+                throw new RuntimeException("Unable to passivate", e);
+            }
+        }
+
+        /**
+         * Calls {@link MessageStore#activate()}.
+         *
+         * <p/>
+         *
+         * This is done a background thread, in line with
+         * {@link StateChangeListener#stateChange(StateChangeEvent)}'s JavaDoc, because
+         * activate may execute transactions, which can't complete until
+         * {@link StateChangeListener#stateChange(StateChangeEvent)} has returned.
+         */
+        private void activateStoreAsync()
+        {
+            final RootMessageLogger _rootLogger = CurrentActor.get().getRootMessageLogger();
+
+            _executor.execute(new Runnable()
+            {
+                private static final String _THREAD_NAME = "BDBHANodeActivationThread";
+
+                @Override
+                public void run()
+                {
+                    Thread.currentThread().setName(_THREAD_NAME);
+                    CurrentActor.set(new AbstractActor(_rootLogger)
+                    {
+                        @Override
+                        public String getLogMessage()
+                        {
+                            return _THREAD_NAME;
+                        }
+                    });
+
+                    try
+                    {
+                        activate();
+                    }
+                    catch (Exception e)
+                    {
+                        LOGGER.error("Failed to activate on hearing MASTER change event",e);
+                    }
+
+                }
+            });
+        }
+    }
+
+    @Override
+    public synchronized void activate() throws Exception
+    {
+        // Before proceeding, perform a log flush with an fsync
+        getEnvironment().flushLog(true);
+
+        super.activate();
+
+        //For replica groups with 2 electable nodes, set the new master to be the
+        //designated primary, such that it can continue working if the replica goes
+        //down and leaves it without a 'majority of 2'.
+        if(getReplicatedEnvironment().getGroup().getElectableNodes().size() <= 2 && _autoDesignatedPrimary)
+        {
+            setDesignatedPrimary(true);
+        }
+    }
+
+    private void setReplicationConfigProperties(ReplicationConfig replicationConfig)
+    {
+        for (Map.Entry<String, String> configItem : _repConfigMap.entrySet())
+        {
+            LOGGER.debug("Setting ReplicationConfig key " + configItem.getKey() + " to '" + configItem.getValue() + "'");
+            replicationConfig.setConfigParam(configItem.getKey(), configItem.getValue());
+        }
+    }
+}
diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBHAMessageStoreManagerMBean.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBHAMessageStoreManagerMBean.java
new file mode 100644
index 0000000000..cd4e990607
--- /dev/null
+++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBHAMessageStoreManagerMBean.java
@@ -0,0 +1,184 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.store.berkeleydb;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+import javax.management.JMException;
+import javax.management.NotCompliantMBeanException;
+import javax.management.openmbean.CompositeData;
+import javax.management.openmbean.CompositeDataSupport;
+import javax.management.openmbean.CompositeType;
+import javax.management.openmbean.OpenDataException;
+import javax.management.openmbean.OpenType;
+import javax.management.openmbean.SimpleType;
+import javax.management.openmbean.TabularData;
+import javax.management.openmbean.TabularDataSupport;
+import javax.management.openmbean.TabularType;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.AMQStoreException;
+import org.apache.qpid.server.management.AMQManagedObject;
+
+public class BDBHAMessageStoreManagerMBean extends AMQManagedObject implements ManagedBDBHAMessageStore
+{
+    private static final Logger LOGGER = Logger.getLogger(BDBHAMessageStoreManagerMBean.class);
+
+    private static  final TabularType GROUP_MEMBERS_TABLE;
+    private static final CompositeType GROUP_MEMBER_ROW;
+    private static final OpenType<?>[] GROUP_MEMBER_ATTRIBUTE_TYPES;
+
+    static
+    {
+        try
+        {
+            GROUP_MEMBER_ATTRIBUTE_TYPES = new OpenType<?>[] {SimpleType.STRING, SimpleType.STRING};
+            final String[] itemNames = new String[] {BDBHAMessageStore.GRP_MEM_COL_NODE_NAME, BDBHAMessageStore.GRP_MEM_COL_NODE_HOST_PORT};
+            final String[] itemDescriptions = new String[] {"Unique node name", "Node host / port "};
+            GROUP_MEMBER_ROW = new CompositeType("GroupMember", "Replication group member",
+                                                itemNames,
+                                                itemDescriptions,
+                                                GROUP_MEMBER_ATTRIBUTE_TYPES );
+            GROUP_MEMBERS_TABLE = new TabularType("GroupMembers", "Replication group memebers",
+                                                GROUP_MEMBER_ROW,
+                                                new String[] {BDBHAMessageStore.GRP_MEM_COL_NODE_NAME});
+        }
+        catch (final OpenDataException ode)
+        {
+            throw new ExceptionInInitializerError(ode);
+        }
+    }
+
+    private final BDBHAMessageStore _store;
+
+    protected BDBHAMessageStoreManagerMBean(BDBHAMessageStore store) throws NotCompliantMBeanException
+    {
+        super(ManagedBDBHAMessageStore.class, ManagedBDBHAMessageStore.TYPE);
+        _store = store;
+    }
+
+    @Override
+    public String getObjectInstanceName()
+    {
+        return _store.getName();
+    }
+
+    @Override
+    public String getGroupName() throws IOException
+    {
+        return _store.getGroupName();
+    }
+
+    @Override
+    public String getNodeName() throws IOException
+    {
+        return _store.getNodeName();
+    }
+
+    @Override
+    public String getNodeHostPort() throws IOException
+    {
+        return _store.getNodeHostPort();
+    }
+
+    @Override
+    public String getHelperHostPort() throws IOException
+    {
+        return _store.getHelperHostPort();
+    }
+
+    @Override
+    public String getReplicationPolicy() throws IOException
+    {
+        return _store.getReplicationPolicy();
+    }
+
+    @Override
+    public String getNodeState() throws IOException
+    {
+        return _store.getNodeState();
+    }
+
+    @Override
+    public boolean getDesignatedPrimary() throws IOException
+    {
+        return _store.isDesignatedPrimary();
+    }
+
+    @Override
+    public TabularData getAllNodesInGroup() throws IOException, JMException
+    {
+        final TabularDataSupport data = new TabularDataSupport(GROUP_MEMBERS_TABLE);
+        final List<Map<String, String>> members = _store.getGroupMembers();
+
+        for (Map<String, String> map : members)
+        {
+            CompositeData memberData = new CompositeDataSupport(GROUP_MEMBER_ROW, map);
+            data.put(memberData);
+        }
+        return data;
+    }
+
+    @Override
+    public void removeNodeFromGroup(String nodeName) throws JMException
+    {
+        try
+        {
+            _store.removeNodeFromGroup(nodeName);
+        }
+        catch (AMQStoreException e)
+        {
+            LOGGER.error("Failed to remove node " + nodeName + " from group", e);
+            throw new JMException(e.getMessage());
+        }
+    }
+
+    @Override
+    public void setDesignatedPrimary(boolean primary) throws JMException
+    {
+        try
+        {
+            _store.setDesignatedPrimary(primary);
+        }
+        catch (AMQStoreException e)
+        {
+            LOGGER.error("Failed to set node " + _store.getNodeName() + " as designated primary", e);
+            throw new JMException(e.getMessage());
+        }
+    }
+
+    @Override
+    public void updateAddress(String nodeName, String newHostName, int newPort) throws JMException
+    {
+        try
+        {
+            _store.updateAddress(nodeName, newHostName, newPort);
+        }
+        catch(AMQStoreException e)
+        {
+            LOGGER.error("Failed to update address for node " + nodeName + " to " + newHostName + ":" + newPort, e);
+            throw new JMException(e.getMessage());
+        }
+    }
+
+
+}
diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStore.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStore.java
index b414441b92..7c29e281d9 100644
--- a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStore.java
+++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStore.java
@@ -21,23 +21,12 @@
 package org.apache.qpid.server.store.berkeleydb;
 
 import java.io.File;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.Queue;
-import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.log4j.Logger;
 import org.apache.qpid.AMQStoreException;
-import org.apache.qpid.server.store.Event;
-import org.apache.qpid.server.store.EventListener;
 import org.apache.qpid.server.store.MessageStore;
 import org.apache.qpid.server.store.StoreFuture;
 
-import com.sleepycat.je.CheckpointConfig;
 import com.sleepycat.je.DatabaseException;
 import com.sleepycat.je.Environment;
 import com.sleepycat.je.EnvironmentConfig;
@@ -53,52 +42,22 @@ import com.sleepycat.je.EnvironmentConfig;
 public class BDBMessageStore extends AbstractBDBMessageStore
 {
     private static final Logger LOGGER = Logger.getLogger(BDBMessageStore.class);
-
-    private final CommitThread _commitThread = new CommitThread("Commit-Thread");
-
-    private final Map<Event, List<EventListener>> _eventListeners = new HashMap<Event, List<EventListener>>();
+    private CommitThreadWrapper _commitThreadWrapper;
 
     @Override
     protected void setupStore(File storePath, String name) throws DatabaseException, AMQStoreException
     {
         super.setupStore(storePath, name);
 
-        startCommitThread();
+        _commitThreadWrapper = new CommitThreadWrapper("Commit-Thread-" + name, getEnvironment());
+        _commitThreadWrapper.startCommitThread();
     }
 
     protected Environment createEnvironment(File environmentPath) throws DatabaseException
     {
         LOGGER.info("BDB message store using environment path " + environmentPath.getAbsolutePath());
-        EnvironmentConfig envConfig = new EnvironmentConfig();
-        // This is what allows the creation of the store if it does not already exist.
-        envConfig.setAllowCreate(true);
-        envConfig.setTransactional(true);
-
-        Properties props = System.getProperties();
-
-        for(String propName : props.stringPropertyNames())
-        {
-            if(propName.startsWith("qpid.bdb.envconfig.je."))
-            {
-                envConfig.setConfigParam(propName.substring(19), props.getProperty(propName));
-            }
-        }
-
-        envConfig.setConfigParam("je.lock.nLockTables", "7");
-
-        // Added to help diagnosis of Deadlock issue
-        // http://www.oracle.com/technology/products/berkeley-db/faq/je_faq.html#23
-        if (Boolean.getBoolean("qpid.bdb.lock.debug"))
-        {
-            envConfig.setConfigParam("je.txn.deadlockStackTrace", "true");
-            envConfig.setConfigParam("je.txn.dumpLocks", "true");
-        }
-
-        // Set transaction mode
-        _transactionConfig.setReadCommitted(true);
+        EnvironmentConfig envConfig = createEnvironmentConfig();
 
-        //This prevents background threads running which will potentially update the store.
-        envConfig.setReadOnly(false);
         try
         {
             return new Environment(environmentPath, envConfig);
@@ -118,12 +77,10 @@ public class BDBMessageStore extends AbstractBDBMessageStore
         }
     }
 
-
-
     @Override
     protected void closeInternal() throws Exception
     {
-        stopCommitThread();
+        _commitThreadWrapper.stopCommitThread();
 
         super.closeInternal();
     }
@@ -133,204 +90,6 @@ public class BDBMessageStore extends AbstractBDBMessageStore
     {
         tx.commitNoSync();
 
-        BDBCommitFuture commitFuture = new BDBCommitFuture(_commitThread, tx, syncCommit);
-        commitFuture.commit();
-
-        return commitFuture;
-    }
-
-    private void startCommitThread()
-    {
-        _commitThread.start();
-    }
-
-    private void stopCommitThread() throws InterruptedException
-    {
-        _commitThread.close();
-        _commitThread.join();
-    }
-
-    private static final class BDBCommitFuture implements StoreFuture
-    {
-        private final CommitThread _commitThread;
-        private final com.sleepycat.je.Transaction _tx;
-        private DatabaseException _databaseException;
-        private boolean _complete;
-        private boolean _syncCommit;
-
-        public BDBCommitFuture(CommitThread commitThread, com.sleepycat.je.Transaction tx, boolean syncCommit)
-        {
-            _commitThread = commitThread;
-            _tx = tx;
-            _syncCommit = syncCommit;
-        }
-
-        public synchronized void complete()
-        {
-            if (LOGGER.isDebugEnabled())
-            {
-                LOGGER.debug("public synchronized void complete(): called (Transaction = " + _tx + ")");
-            }
-            _complete = true;
-
-            notifyAll();
-        }
-
-        public synchronized void abort(DatabaseException databaseException)
-        {
-            _complete = true;
-            _databaseException = databaseException;
-
-            notifyAll();
-        }
-
-        public void commit() throws DatabaseException
-        {
-            _commitThread.addJob(this, _syncCommit);
-
-            if(!_syncCommit)
-            {
-                LOGGER.debug("CommitAsync was requested, returning immediately.");
-                return;
-            }
-
-            waitForCompletion();
-
-            if (_databaseException != null)
-            {
-                throw _databaseException;
-            }
-
-        }
-
-        public synchronized boolean isComplete()
-        {
-            return _complete;
-        }
-
-        public synchronized void waitForCompletion()
-        {
-            while (!isComplete())
-            {
-                _commitThread.explicitNotify();
-                try
-                {
-                    wait(250);
-                }
-                catch (InterruptedException e)
-                {
-                    //TODO Should we ignore, or throw a 'StoreException'?
-                    throw new RuntimeException(e);
-                }
-            }
-        }
-    }
-
-    /**
-     * Implements a thread which batches and commits a queue of {@link BDBCommitFuture} operations. The commit operations
-     * themselves are responsible for adding themselves to the queue and waiting for the commit to happen before
-     * continuing, but it is the responsibility of this thread to tell the commit operations when they have been
-     * completed by calling back on their {@link BDBCommitFuture#complete()} and {@link BDBCommitFuture#abort} methods.
-     *
-     * <p/><table id="crc"><caption>CRC Card</caption> <tr><th> Responsibilities <th> Collaborations </table>
-     */
-    private class CommitThread extends Thread
-    {
-        private final AtomicBoolean _stopped = new AtomicBoolean(false);
-        private final Queue<BDBCommitFuture> _jobQueue = new ConcurrentLinkedQueue<BDBCommitFuture>();
-        private final CheckpointConfig _config = new CheckpointConfig();
-        private final Object _lock = new Object();
-
-        public CommitThread(String name)
-        {
-            super(name);
-            _config.setForce(true);
-
-        }
-
-        public void explicitNotify()
-        {
-            synchronized (_lock)
-            {
-                _lock.notify();
-            }
-        }
-
-        public void run()
-        {
-            while (!_stopped.get())
-            {
-                synchronized (_lock)
-                {
-                    while (!_stopped.get() && !hasJobs())
-                    {
-                        try
-                        {
-                            // RHM-7 Periodically wake up and check, just in case we
-                            // missed a notification. Don't want to lock the broker hard.
-                            _lock.wait(1000);
-                        }
-                        catch (InterruptedException e)
-                        {
-                        }
-                    }
-                }
-                processJobs();
-            }
-        }
-
-        private void processJobs()
-        {
-            int size = _jobQueue.size();
-
-            try
-            {
-                getEnvironment().flushLog(true);
-
-                for(int i = 0; i < size; i++)
-                {
-                    BDBCommitFuture commit = _jobQueue.poll();
-                    commit.complete();
-                }
-
-            }
-            catch (DatabaseException e)
-            {
-                for(int i = 0; i < size; i++)
-                {
-                    BDBCommitFuture commit = _jobQueue.poll();
-                    commit.abort(e);
-                }
-            }
-
-        }
-
-        private boolean hasJobs()
-        {
-            return !_jobQueue.isEmpty();
-        }
-
-        public void addJob(BDBCommitFuture commit, final boolean sync)
-        {
-
-            _jobQueue.add(commit);
-            if(sync)
-            {
-                synchronized (_lock)
-                {
-                    _lock.notifyAll();
-                }
-            }
-        }
-
-        public void close()
-        {
-            synchronized (_lock)
-            {
-                _stopped.set(true);
-                _lock.notifyAll();
-            }
-        }
+        return _commitThreadWrapper.commit(tx, syncCommit);
     }
-
 }
diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/CommitThreadWrapper.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/CommitThreadWrapper.java
new file mode 100644
index 0000000000..5b0abbec93
--- /dev/null
+++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/CommitThreadWrapper.java
@@ -0,0 +1,227 @@
+package org.apache.qpid.server.store.berkeleydb;
+
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.server.store.StoreFuture;
+
+import com.sleepycat.je.CheckpointConfig;
+import com.sleepycat.je.DatabaseException;
+import com.sleepycat.je.Environment;
+import com.sleepycat.je.Transaction;
+
+public class CommitThreadWrapper
+{
+    private final CommitThread _commitThread;
+    
+    public CommitThreadWrapper(String name, Environment env)
+    {
+        _commitThread = new CommitThread(name, env);
+    }
+
+    public void startCommitThread()
+    {
+        _commitThread.start();
+    }
+
+    public void stopCommitThread() throws InterruptedException
+    {
+        _commitThread.close();
+        _commitThread.join();
+    }
+
+    public StoreFuture commit(Transaction tx, boolean syncCommit)
+    {
+        BDBCommitFuture commitFuture = new BDBCommitFuture(_commitThread, tx, syncCommit);
+        commitFuture.commit();
+        return commitFuture;
+    }
+
+    private static final class BDBCommitFuture implements StoreFuture
+    {
+        private static final Logger LOGGER = Logger.getLogger(BDBCommitFuture.class);
+
+        private final CommitThread _commitThread;
+        private final Transaction _tx;
+        private DatabaseException _databaseException;
+        private boolean _complete;
+        private boolean _syncCommit;
+
+        public BDBCommitFuture(CommitThread commitThread, Transaction tx, boolean syncCommit)
+        {
+            _commitThread = commitThread;
+            _tx = tx;
+            _syncCommit = syncCommit;
+        }
+
+        public synchronized void complete()
+        {
+            if (LOGGER.isDebugEnabled())
+            {
+                LOGGER.debug("public synchronized void complete(): called (Transaction = " + _tx + ")");
+            }
+            _complete = true;
+
+            notifyAll();
+        }
+
+        public synchronized void abort(DatabaseException databaseException)
+        {
+            _complete = true;
+            _databaseException = databaseException;
+
+            notifyAll();
+        }
+
+        public void commit() throws DatabaseException
+        {
+            _commitThread.addJob(this, _syncCommit);
+
+            if(!_syncCommit)
+            {
+                LOGGER.debug("CommitAsync was requested, returning immediately.");
+                return;
+            }
+
+            waitForCompletion();
+
+            if (_databaseException != null)
+            {
+                throw _databaseException;
+            }
+
+        }
+
+        public synchronized boolean isComplete()
+        {
+            return _complete;
+        }
+
+        public synchronized void waitForCompletion()
+        {
+            while (!isComplete())
+            {
+                _commitThread.explicitNotify();
+                try
+                {
+                    wait(250);
+                }
+                catch (InterruptedException e)
+                {
+                    //TODO Should we ignore, or throw a 'StoreException'?
+                    throw new RuntimeException(e);
+                }
+            }
+        }
+    }
+
+    /**
+     * Implements a thread which batches and commits a queue of {@link BDBCommitFuture} operations. The commit operations
+     * themselves are responsible for adding themselves to the queue and waiting for the commit to happen before
+     * continuing, but it is the responsibility of this thread to tell the commit operations when they have been
+     * completed by calling back on their {@link BDBCommitFuture#complete()} and {@link BDBCommitFuture#abort} methods.
+     *
+     * <p/><table id="crc"><caption>CRC Card</caption> <tr><th> Responsibilities <th> Collaborations </table>
+     */
+    private static class CommitThread extends Thread
+    {
+        private final AtomicBoolean _stopped = new AtomicBoolean(false);
+        private final Queue<BDBCommitFuture> _jobQueue = new ConcurrentLinkedQueue<BDBCommitFuture>();
+        private final CheckpointConfig _config = new CheckpointConfig();
+        private final Object _lock = new Object();
+        private Environment _environment;
+
+        public CommitThread(String name, Environment env)
+        {
+            super(name);
+            _config.setForce(true);
+            _environment = env;
+        }
+
+        public void explicitNotify()
+        {
+            synchronized (_lock)
+            {
+                _lock.notify();
+            }
+        }
+
+        public void run()
+        {
+            while (!_stopped.get())
+            {
+                synchronized (_lock)
+                {
+                    while (!_stopped.get() && !hasJobs())
+                    {
+                        try
+                        {
+                            // RHM-7 Periodically wake up and check, just in case we
+                            // missed a notification. Don't want to lock the broker hard.
+                            _lock.wait(1000);
+                        }
+                        catch (InterruptedException e)
+                        {
+                        }
+                    }
+                }
+                processJobs();
+            }
+        }
+
+        private void processJobs()
+        {
+            int size = _jobQueue.size();
+
+            try
+            {
+                _environment.flushLog(true);
+
+                for(int i = 0; i < size; i++)
+                {
+                    BDBCommitFuture commit = _jobQueue.poll();
+                    commit.complete();
+                }
+
+            }
+            catch (DatabaseException e)
+            {
+                for(int i = 0; i < size; i++)
+                {
+                    BDBCommitFuture commit = _jobQueue.poll();
+                    commit.abort(e);
+                }
+            }
+
+        }
+
+        private boolean hasJobs()
+        {
+            return !_jobQueue.isEmpty();
+        }
+
+        public void addJob(BDBCommitFuture commit, final boolean sync)
+        {
+
+            _jobQueue.add(commit);
+            if(sync)
+            {
+                synchronized (_lock)
+                {
+                    _lock.notifyAll();
+                }
+            }
+        }
+
+        public void close()
+        {
+            synchronized (_lock)
+            {
+                _stopped.set(true);
+                _lock.notifyAll();
+            }
+        }
+    }
+}
diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ManagedBDBHAMessageStore.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ManagedBDBHAMessageStore.java
new file mode 100644
index 0000000000..bfc7bbf128
--- /dev/null
+++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/ManagedBDBHAMessageStore.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.store.berkeleydb;
+
+import java.io.IOException;
+
+import javax.management.JMException;
+import javax.management.openmbean.TabularData;
+
+import org.apache.qpid.management.common.mbeans.annotations.MBeanAttribute;
+import org.apache.qpid.management.common.mbeans.annotations.MBeanOperation;
+
+public interface ManagedBDBHAMessageStore
+{
+    public static final String TYPE = "BDBHAMessageStore";
+
+    public static final String ATTR_GROUP_NAME = "GroupName";
+    public static final String ATTR_NODE_NAME = "NodeName";
+    public static final String ATTR_NODE_HOST_PORT = "NodeHostPort";
+    public static final String ATTR_HELPER_HOST_PORT = "HelperHostPort";
+    public static final String ATTR_REPLICATION_POLICY = "ReplicationPolicy";
+    public static final String ATTR_NODE_STATE = "NodeState";
+    public static final String ATTR_DESIGNATED_PRIMARY = "DesignatedPrimary";
+
+    @MBeanAttribute(name=ATTR_GROUP_NAME, description="Name identifying the group")
+    String getGroupName() throws IOException;
+
+    @MBeanAttribute(name=ATTR_NODE_NAME, description="Unique name identifying the node within the group")
+    String getNodeName() throws IOException;
+
+    @MBeanAttribute(name=ATTR_NODE_HOST_PORT, description="Host/port used to replicate data between this node and others in the group")
+    String getNodeHostPort() throws IOException;
+
+    @MBeanAttribute(name=ATTR_NODE_STATE, description="Current state of this node")
+    String getNodeState() throws IOException;
+
+    @MBeanAttribute(name=ATTR_HELPER_HOST_PORT, description="Host/port used to allow a new node to discover other group members")
+    String getHelperHostPort() throws IOException;
+
+    @MBeanAttribute(name=ATTR_REPLICATION_POLICY, description="Replication policy")
+    String getReplicationPolicy() throws IOException;
+
+    @MBeanAttribute(name=ATTR_DESIGNATED_PRIMARY, description="Designated primary flag. Applicable to the two node case.")
+    boolean getDesignatedPrimary() throws IOException;
+
+    @MBeanOperation(name="getAllNodesInGroup", description="Get all nodes within the group, regardless of whether currently attached or not")
+    TabularData getAllNodesInGroup() throws IOException, JMException;
+
+    @MBeanOperation(name="removeNodeFromGroup", description="Remove an existing node from the group")
+    void removeNodeFromGroup(String nodeName) throws JMException;
+
+    @MBeanOperation(name="setDesignatedPrimary", description="Set/unset this node as the designated primary for the group. Applicable to the two node case.")
+    void setDesignatedPrimary(boolean primary) throws JMException;
+
+    @MBeanOperation(name="updateAddress", description="Update the address of another node. The node must be in a STOPPED state.")
+    void updateAddress(String nodeName, String newHostName, int newPort) throws JMException;
+}
+
diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/Upgrader.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/Upgrader.java
index e71e39cbb8..f1ab012efc 100644
--- a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/Upgrader.java
+++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/upgrade/Upgrader.java
@@ -24,7 +24,7 @@ import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationTargetException;
 
 import org.apache.qpid.AMQStoreException;
-import org.apache.qpid.server.store.berkeleydb.BDBMessageStore;
+import org.apache.qpid.server.store.berkeleydb.AbstractBDBMessageStore;
 
 import com.sleepycat.bind.tuple.IntegerBinding;
 import com.sleepycat.bind.tuple.LongBinding;
@@ -63,7 +63,7 @@ public class Upgrader
 
             if(versionDb.count() == 0L)
             {
-                int sourceVersion = isEmpty ? BDBMessageStore.VERSION: identifyOldStoreVersion();
+                int sourceVersion = isEmpty ? AbstractBDBMessageStore.VERSION: identifyOldStoreVersion();
                 DatabaseEntry key = new DatabaseEntry();
                 IntegerBinding.intToEntry(sourceVersion, key);
                 DatabaseEntry value = new DatabaseEntry();
@@ -87,7 +87,7 @@ public class Upgrader
 
     int getSourceVersion(Database versionDb)
     {
-        int version = BDBMessageStore.VERSION + 1;
+        int version = AbstractBDBMessageStore.VERSION + 1;
         OperationStatus result;
 
         do
@@ -106,7 +106,7 @@ public class Upgrader
     void performUpgradeFromVersion(int sourceVersion, Database versionDb)
             throws AMQStoreException
     {
-        while(sourceVersion != BDBMessageStore.VERSION)
+        while(sourceVersion != AbstractBDBMessageStore.VERSION)
         {
             upgrade(sourceVersion, ++sourceVersion);
             DatabaseEntry key = new DatabaseEntry();
diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAMessageStoreManagerMBeanTest.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAMessageStoreManagerMBeanTest.java
new file mode 100644
index 0000000000..00f99b7097
--- /dev/null
+++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAMessageStoreManagerMBeanTest.java
@@ -0,0 +1,218 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.store.berkeleydb;
+
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import javax.management.JMException;
+import javax.management.openmbean.CompositeData;
+import javax.management.openmbean.CompositeType;
+import javax.management.openmbean.TabularData;
+
+import junit.framework.TestCase;
+
+import org.apache.qpid.AMQStoreException;
+import org.apache.qpid.server.logging.SystemOutMessageLogger;
+import org.apache.qpid.server.logging.actors.CurrentActor;
+import org.apache.qpid.server.logging.actors.TestLogActor;
+
+public class BDBHAMessageStoreManagerMBeanTest extends TestCase
+{
+    private static final String TEST_GROUP_NAME = "testGroupName";
+    private static final String TEST_NODE_NAME = "testNodeName";
+    private static final String TEST_NODE_HOST_PORT = "host:1234";
+    private static final String TEST_HELPER_HOST_PORT = "host:5678";
+    private static final String TEST_REPLICATION_POLICY = "sync,sync,all";
+    private static final String TEST_NODE_STATE = "MASTER";
+    private static final String TEST_STORE_NAME = "testStoreName";
+    private static final boolean TEST_DESIGNATED_PRIMARY_FLAG = false;
+
+    private BDBHAMessageStore _store;
+    private BDBHAMessageStoreManagerMBean _mBean;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+
+        CurrentActor.set(new TestLogActor(new SystemOutMessageLogger()));
+        _store = mock(BDBHAMessageStore.class);
+        _mBean = new BDBHAMessageStoreManagerMBean(_store);
+    }
+
+    @Override
+    protected void tearDown() throws Exception
+    {
+        super.tearDown();
+        CurrentActor.remove();
+    }
+
+    public void testObjectName() throws Exception
+    {
+        when(_store.getName()).thenReturn(TEST_STORE_NAME);
+
+        String expectedObjectName = "org.apache.qpid:type=BDBHAMessageStore,name=" + TEST_STORE_NAME;
+        assertEquals(expectedObjectName, _mBean.getObjectName().toString());
+    }
+
+    public void testGroupName() throws Exception
+    {
+        when(_store.getGroupName()).thenReturn(TEST_GROUP_NAME);
+
+        assertEquals(TEST_GROUP_NAME, _mBean.getAttribute(ManagedBDBHAMessageStore.ATTR_GROUP_NAME));
+    }
+
+    public void testNodeName() throws Exception
+    {
+        when(_store.getNodeName()).thenReturn(TEST_NODE_NAME);
+
+        assertEquals(TEST_NODE_NAME, _mBean.getAttribute(ManagedBDBHAMessageStore.ATTR_NODE_NAME));
+    }
+
+    public void testNodeHostPort() throws Exception
+    {
+        when(_store.getNodeHostPort()).thenReturn(TEST_NODE_HOST_PORT);
+
+        assertEquals(TEST_NODE_HOST_PORT, _mBean.getAttribute(ManagedBDBHAMessageStore.ATTR_NODE_HOST_PORT));
+    }
+
+    public void testHelperHostPort() throws Exception
+    {
+        when(_store.getHelperHostPort()).thenReturn(TEST_HELPER_HOST_PORT);
+
+        assertEquals(TEST_HELPER_HOST_PORT, _mBean.getAttribute(ManagedBDBHAMessageStore.ATTR_HELPER_HOST_PORT));
+    }
+
+    public void testReplicationPolicy() throws Exception
+    {
+        when(_store.getReplicationPolicy()).thenReturn(TEST_REPLICATION_POLICY);
+
+        assertEquals(TEST_REPLICATION_POLICY, _mBean.getAttribute(ManagedBDBHAMessageStore.ATTR_REPLICATION_POLICY));
+    }
+
+    public void testNodeState() throws Exception
+    {
+        when(_store.getNodeState()).thenReturn(TEST_NODE_STATE);
+
+        assertEquals(TEST_NODE_STATE, _mBean.getAttribute(ManagedBDBHAMessageStore.ATTR_NODE_STATE));
+    }
+
+    public void testDesignatedPrimaryFlag() throws Exception
+    {
+        when(_store.isDesignatedPrimary()).thenReturn(TEST_DESIGNATED_PRIMARY_FLAG);
+
+        assertEquals(TEST_DESIGNATED_PRIMARY_FLAG, _mBean.getAttribute(ManagedBDBHAMessageStore.ATTR_DESIGNATED_PRIMARY));
+    }
+
+    public void testGroupMembersForGroupWithOneNode() throws Exception
+    {
+        List<Map<String, String>> members = Collections.singletonList(createTestNodeResult());
+        when(_store.getGroupMembers()).thenReturn(members);
+
+        final TabularData resultsTable = _mBean.getAllNodesInGroup();
+
+        assertTableHasHeadingsNamed(resultsTable, BDBHAMessageStore.GRP_MEM_COL_NODE_NAME, BDBHAMessageStore.GRP_MEM_COL_NODE_HOST_PORT);
+
+        final int numberOfDataRows = resultsTable.size();
+        assertEquals("Unexpected number of data rows", 1 ,numberOfDataRows);
+        final CompositeData row = (CompositeData) resultsTable.values().iterator().next();
+        assertEquals(TEST_NODE_NAME, row.get(BDBHAMessageStore.GRP_MEM_COL_NODE_NAME));
+        assertEquals(TEST_NODE_HOST_PORT, row.get(BDBHAMessageStore.GRP_MEM_COL_NODE_HOST_PORT));
+    }
+
+    public void testRemoveNodeFromReplicationGroup() throws Exception
+    {
+        _mBean.removeNodeFromGroup(TEST_NODE_NAME);
+
+        verify(_store).removeNodeFromGroup(TEST_NODE_NAME);
+    }
+
+    public void testRemoveNodeFromReplicationGroupWithError() throws Exception
+    {
+        doThrow(new AMQStoreException("mocked exception")).when(_store).removeNodeFromGroup(TEST_NODE_NAME);
+
+        try
+        {
+            _mBean.removeNodeFromGroup(TEST_NODE_NAME);
+            fail("Exception not thrown");
+        }
+        catch (JMException je)
+        {
+            // PASS
+        }
+    }
+
+    public void testSetAsDesignatedPrimary() throws Exception
+    {
+        _mBean.setDesignatedPrimary(true);
+
+        verify(_store).setDesignatedPrimary(true);
+    }
+
+    public void testSetAsDesignatedPrimaryWithError() throws Exception
+    {
+        doThrow(new AMQStoreException("mocked exception")).when(_store).setDesignatedPrimary(true);
+
+        try
+        {
+            _mBean.setDesignatedPrimary(true);
+            fail("Exception not thrown");
+        }
+        catch (JMException je)
+        {
+            // PASS
+        }
+    }
+
+    public void testUpdateAddress() throws Exception
+    {
+        String newHostName = "newHostName";
+        int newPort = 1967;
+
+        _mBean.updateAddress(TEST_NODE_NAME, newHostName, newPort);
+
+        verify(_store).updateAddress(TEST_NODE_NAME, newHostName, newPort);
+    }
+
+    private void assertTableHasHeadingsNamed(final TabularData resultsTable, String... headingNames)
+    {
+        CompositeType headingsRow = resultsTable.getTabularType().getRowType();
+        for (final String headingName : headingNames)
+        {
+            assertTrue("Table should have column with heading " + headingName, headingsRow.containsKey(headingName));
+        }
+    }
+
+    private Map<String, String> createTestNodeResult()
+    {
+        Map<String, String> items = new HashMap<String, String>();
+        items.put(BDBHAMessageStore.GRP_MEM_COL_NODE_NAME, TEST_NODE_NAME);
+        items.put(BDBHAMessageStore.GRP_MEM_COL_NODE_HOST_PORT, TEST_NODE_HOST_PORT);
+        return items;
+    }
+}
diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAMessageStoreTest.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAMessageStoreTest.java
new file mode 100644
index 0000000000..6f851bd94e
--- /dev/null
+++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAMessageStoreTest.java
@@ -0,0 +1,144 @@
+package org.apache.qpid.server.store.berkeleydb;
+
+import java.io.File;
+import java.net.InetAddress;
+
+import org.apache.commons.configuration.XMLConfiguration;
+import org.apache.qpid.server.configuration.ServerConfiguration;
+import org.apache.qpid.server.logging.SystemOutMessageLogger;
+import org.apache.qpid.server.logging.actors.CurrentActor;
+import org.apache.qpid.server.logging.actors.TestLogActor;
+import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.server.util.TestApplicationRegistry;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.test.utils.QpidTestCase;
+import org.apache.qpid.util.FileUtils;
+
+import com.sleepycat.je.Environment;
+import com.sleepycat.je.EnvironmentConfig;
+import com.sleepycat.je.rep.ReplicatedEnvironment;
+import com.sleepycat.je.rep.ReplicationConfig;
+
+public class BDBHAMessageStoreTest extends QpidTestCase
+{
+    private static final String TEST_LOG_FILE_MAX = "1000000";
+    private static final String TEST_ELECTION_RETRIES = "1000";
+    private static final String TEST_NUMBER_OF_THREADS = "10";
+    private static final String TEST_ENV_CONSISTENCY_TIMEOUT = "9999999";
+    private String _groupName;
+    private String _workDir;
+    private int _masterPort;
+    private String _host;
+    private XMLConfiguration _configXml;
+
+    public void setUp() throws Exception
+    {
+        super.setUp();
+
+        _workDir = TMP_FOLDER + File.separator + getName();
+        _host = InetAddress.getByName("localhost").getHostAddress();
+        _groupName = "group" + getName();
+        _masterPort = -1;
+
+        FileUtils.delete(new File(_workDir), true);
+        _configXml = new XMLConfiguration();
+    }
+
+    public void tearDown() throws Exception
+    {
+        FileUtils.delete(new File(_workDir), true);
+        super.tearDown();
+    }
+
+    public void testSetSystemConfiguration() throws Exception
+    {
+        // create virtual host configuration, registry and host instance
+        addVirtualHostConfiguration();
+        TestApplicationRegistry registry = initialize();
+        try
+        {
+            VirtualHost virtualhost = registry.getVirtualHostRegistry().getVirtualHost("test" + _masterPort);
+            BDBHAMessageStore store = (BDBHAMessageStore) virtualhost.getMessageStore();
+
+            // test whether JVM system settings were applied
+            Environment env = store.getEnvironment();
+            assertEquals("Unexpected number of cleaner threads", TEST_NUMBER_OF_THREADS, env.getConfig().getConfigParam(EnvironmentConfig.CLEANER_THREADS));
+            assertEquals("Unexpected log file max", TEST_LOG_FILE_MAX, env.getConfig().getConfigParam(EnvironmentConfig.LOG_FILE_MAX));
+
+            ReplicatedEnvironment repEnv = store.getReplicatedEnvironment();
+            assertEquals("Unexpected number of elections primary retries", TEST_ELECTION_RETRIES,
+                    repEnv.getConfig().getConfigParam(ReplicationConfig.ELECTIONS_PRIMARY_RETRIES));
+            assertEquals("Unexpected number of elections primary retries", TEST_ENV_CONSISTENCY_TIMEOUT,
+                    repEnv.getConfig().getConfigParam(ReplicationConfig.ENV_CONSISTENCY_TIMEOUT));
+        }
+        finally
+        {
+            ApplicationRegistry.remove();
+        }
+    }
+
+    private void addVirtualHostConfiguration() throws Exception
+    {
+        int port = findFreePort();
+        if (_masterPort == -1)
+        {
+            _masterPort = port;
+        }
+        String nodeName = getNodeNameForNodeAt(port);
+
+        String vhostName = "test" + port;
+        String vhostPrefix = "virtualhosts.virtualhost." + vhostName;
+
+        _configXml.addProperty("virtualhosts.virtualhost.name", vhostName);
+        _configXml.addProperty(vhostPrefix + ".store.class", BDBHAMessageStore.class.getName());
+        _configXml.addProperty(vhostPrefix + ".store.environment-path", _workDir + File.separator
+                + port);
+        _configXml.addProperty(vhostPrefix + ".store.highAvailability.groupName", _groupName);
+        _configXml.addProperty(vhostPrefix + ".store.highAvailability.nodeName", nodeName);
+        _configXml.addProperty(vhostPrefix + ".store.highAvailability.nodeHostPort",
+                getNodeHostPortForNodeAt(port));
+        _configXml.addProperty(vhostPrefix + ".store.highAvailability.helperHostPort",
+                getHelperHostPort());
+
+        _configXml.addProperty(vhostPrefix + ".store.envConfig(-1).name", EnvironmentConfig.CLEANER_THREADS);
+        _configXml.addProperty(vhostPrefix + ".store.envConfig.value", TEST_NUMBER_OF_THREADS);
+
+        _configXml.addProperty(vhostPrefix + ".store.envConfig(-1).name", EnvironmentConfig.LOG_FILE_MAX);
+        _configXml.addProperty(vhostPrefix + ".store.envConfig.value", TEST_LOG_FILE_MAX);
+
+        _configXml.addProperty(vhostPrefix + ".store.repConfig(-1).name", ReplicationConfig.ELECTIONS_PRIMARY_RETRIES);
+        _configXml.addProperty(vhostPrefix + ".store.repConfig.value", TEST_ELECTION_RETRIES);
+
+        _configXml.addProperty(vhostPrefix + ".store.repConfig(-1).name", ReplicationConfig.ENV_CONSISTENCY_TIMEOUT);
+        _configXml.addProperty(vhostPrefix + ".store.repConfig.value", TEST_ENV_CONSISTENCY_TIMEOUT);
+    }
+
+    private String getNodeNameForNodeAt(final int bdbPort)
+    {
+        return "node" + getName() + bdbPort;
+    }
+
+    private String getNodeHostPortForNodeAt(final int bdbPort)
+    {
+        return _host + ":" + bdbPort;
+    }
+
+    private String getHelperHostPort()
+    {
+        if (_masterPort == -1)
+        {
+            throw new IllegalStateException("Helper port not yet assigned.");
+        }
+        return _host + ":" + _masterPort;
+    }
+
+    private TestApplicationRegistry initialize() throws Exception
+    {
+        CurrentActor.set(new TestLogActor(new SystemOutMessageLogger()));
+        ServerConfiguration configuration = new ServerConfiguration(_configXml);
+        TestApplicationRegistry registry = new TestApplicationRegistry(configuration);
+        ApplicationRegistry.initialise(registry);
+        registry.getVirtualHostRegistry().setDefaultVirtualHostName("test" + _masterPort);
+        return registry;
+    }
+}
diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStoreTest.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStoreTest.java
index a318187f13..591bc27d1e 100644
--- a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStoreTest.java
+++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStoreTest.java
@@ -70,7 +70,7 @@ public class BDBMessageStoreTest extends org.apache.qpid.server.store.MessageSto
     {
         MessageStore store = getVirtualHost().getMessageStore();
 
-        BDBMessageStore bdbStore = assertBDBStore(store);
+        AbstractBDBMessageStore bdbStore = assertBDBStore(store);
 
         // Create content ByteBuffers.
         // Split the content into 2 chunks for the 0-8 message, as per broker behaviour.
@@ -220,11 +220,11 @@ public class BDBMessageStoreTest extends org.apache.qpid.server.store.MessageSto
      * Use this method instead of reloading the virtual host like other tests in order
      * to avoid the recovery handler deleting the message for not being on a queue.
      */
-    private BDBMessageStore reloadStore(BDBMessageStore messageStore) throws Exception
+    private AbstractBDBMessageStore reloadStore(AbstractBDBMessageStore messageStore) throws Exception
     {
         messageStore.close();
 
-        BDBMessageStore newStore = new BDBMessageStore();
+        AbstractBDBMessageStore newStore = new BDBMessageStore();
         newStore.configure("", _config.subset("store"));
 
         newStore.startWithNoRecover();
@@ -282,7 +282,7 @@ public class BDBMessageStoreTest extends org.apache.qpid.server.store.MessageSto
     public void testGetContentWithOffset() throws Exception
     {
         MessageStore store = getVirtualHost().getMessageStore();
-        BDBMessageStore bdbStore = assertBDBStore(store);
+        AbstractBDBMessageStore bdbStore = assertBDBStore(store);
         StoredMessage<MessageMetaData> storedMessage_0_8 = createAndStoreSingleChunkMessage_0_8(store);
         long messageid_0_8 = storedMessage_0_8.getMessageNumber();
 
@@ -342,7 +342,7 @@ public class BDBMessageStoreTest extends org.apache.qpid.server.store.MessageSto
     public void testMessageCreationAndRemoval() throws Exception
     {
         MessageStore store = getVirtualHost().getMessageStore();
-        BDBMessageStore bdbStore = assertBDBStore(store);
+        AbstractBDBMessageStore bdbStore = assertBDBStore(store);
 
         StoredMessage<MessageMetaData> storedMessage_0_8 = createAndStoreSingleChunkMessage_0_8(store);
         long messageid_0_8 = storedMessage_0_8.getMessageNumber();
@@ -367,12 +367,12 @@ public class BDBMessageStoreTest extends org.apache.qpid.server.store.MessageSto
         assertEquals("Retrieved content when none was expected",
                         0, bdbStore.getContent(messageid_0_8, 0, dst));
     }
-    private BDBMessageStore assertBDBStore(MessageStore store)
+    private AbstractBDBMessageStore assertBDBStore(MessageStore store)
     {
 
         assertEquals("Test requires an instance of BDBMessageStore to proceed", BDBMessageStore.class, store.getClass());
 
-        return (BDBMessageStore) store;
+        return (AbstractBDBMessageStore) store;
     }
 
     private StoredMessage<MessageMetaData> createAndStoreSingleChunkMessage_0_8(MessageStore store)
@@ -405,7 +405,7 @@ public class BDBMessageStoreTest extends org.apache.qpid.server.store.MessageSto
     {
         MessageStore log = getVirtualHost().getMessageStore();
 
-        BDBMessageStore bdbStore = assertBDBStore(log);
+        AbstractBDBMessageStore bdbStore = assertBDBStore(log);
 
         final UUID mockQueueId = UUIDGenerator.generateUUID();
         TransactionLogResource mockQueue = new TransactionLogResource()
@@ -443,7 +443,7 @@ public class BDBMessageStoreTest extends org.apache.qpid.server.store.MessageSto
     {
         MessageStore log = getVirtualHost().getMessageStore();
 
-        BDBMessageStore bdbStore = assertBDBStore(log);
+        AbstractBDBMessageStore bdbStore = assertBDBStore(log);
 
         final UUID mockQueueId = UUIDGenerator.generateUUID();
         TransactionLogResource mockQueue = new TransactionLogResource()
@@ -484,7 +484,7 @@ public class BDBMessageStoreTest extends org.apache.qpid.server.store.MessageSto
     {
         MessageStore log = getVirtualHost().getMessageStore();
 
-        BDBMessageStore bdbStore = assertBDBStore(log);
+        AbstractBDBMessageStore bdbStore = assertBDBStore(log);
 
         final UUID mockQueueId = UUIDGenerator.generateUUID();
         TransactionLogResource mockQueue = new TransactionLogResource()
diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/HAClusterBlackboxTest.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/HAClusterBlackboxTest.java
new file mode 100644
index 0000000000..afe0435901
--- /dev/null
+++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/HAClusterBlackboxTest.java
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.store.berkeleydb;
+
+import java.io.File;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import javax.jms.Connection;
+import javax.jms.Session;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.client.AMQConnection;
+import org.apache.qpid.jms.ConnectionListener;
+import org.apache.qpid.jms.ConnectionURL;
+import org.apache.qpid.test.utils.QpidBrokerTestCase;
+
+/**
+ * The HA black box tests test the BDB cluster as a opaque unit.  Client connects to
+ * the cluster via a failover url
+ *
+ * @see HAClusterWhiteboxTest
+ */
+public class HAClusterBlackboxTest extends QpidBrokerTestCase
+{
+    protected static final Logger LOGGER = Logger.getLogger(HAClusterBlackboxTest.class);
+
+    private static final String VIRTUAL_HOST = "test";
+    private static final int NUMBER_OF_NODES = 3;
+
+    private final HATestClusterCreator _clusterCreator = new HATestClusterCreator(this, VIRTUAL_HOST, NUMBER_OF_NODES);
+
+    private FailoverAwaitingListener _failoverAwaitingListener;
+    private ConnectionURL _brokerFailoverUrl;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        _brokerType = BrokerType.SPAWNED;
+
+        assertTrue(isJavaBroker());
+        assertTrue(isBrokerStorePersistent());
+
+        setBrokerEnvironment("QPID_OPTS", "-Djava.util.logging.config.file=" + System.getProperty(QPID_HOME)
+                + File.separator + "etc" + File.separator + "log.properties");
+
+        _clusterCreator.configureClusterNodes();
+
+        _brokerFailoverUrl = _clusterCreator.getConnectionUrlForAllClusterNodes();
+
+        _clusterCreator.startCluster();
+        _failoverAwaitingListener = new FailoverAwaitingListener();
+
+        super.setUp();
+    }
+
+    @Override
+    public void startBroker() throws Exception
+    {
+        // Don't start default broker provided by QBTC.
+    }
+
+    public void testLossOfActiveNodeCausesClientToFailover() throws Exception
+    {
+        final Connection connection = getConnection(_brokerFailoverUrl);
+
+        ((AMQConnection)connection).setConnectionListener(_failoverAwaitingListener);
+
+        final int activeBrokerPort = _clusterCreator.getBrokerPortNumberFromConnection(connection);
+        LOGGER.info("Active connection port " + activeBrokerPort);
+
+        _clusterCreator.stopNode(activeBrokerPort);
+        LOGGER.info("Node is stopped");
+        _failoverAwaitingListener.assertFailoverOccurs(20000);
+        LOGGER.info("Listener has finished");
+        // any op to ensure connection remains
+        connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+    }
+
+    public void testLossOfInactiveNodeDoesNotCauseClientToFailover() throws Exception
+    {
+        LOGGER.info("Connecting to " + _brokerFailoverUrl);
+        final Connection connection = getConnection(_brokerFailoverUrl);
+
+        ((AMQConnection)connection).setConnectionListener(_failoverAwaitingListener);
+        final int activeBrokerPort = _clusterCreator.getBrokerPortNumberFromConnection(connection);
+        LOGGER.info("Active connection port " + activeBrokerPort);
+        final int inactiveBrokerPort = _clusterCreator.getPortNumberOfAnInactiveBroker(connection);
+
+        LOGGER.info("Stopping inactive broker on port " + inactiveBrokerPort);
+
+        _clusterCreator.stopNode(inactiveBrokerPort);
+
+        _failoverAwaitingListener.assertFailoverDoesNotOccur(2000);
+
+        // any op to ensure connection remains
+        connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+    }
+
+    private final class FailoverAwaitingListener implements ConnectionListener
+    {
+        private final CountDownLatch _failoverLatch = new CountDownLatch(1);
+
+        @Override
+        public boolean preResubscribe()
+        {
+            return true;
+        }
+
+        @Override
+        public boolean preFailover(boolean redirect)
+        {
+            return true;
+        }
+
+        public void assertFailoverOccurs(long delay) throws InterruptedException
+        {
+            _failoverLatch.await(delay, TimeUnit.MILLISECONDS);
+            assertEquals("Failover did not occur", 0, _failoverLatch.getCount());
+        }
+
+        public void assertFailoverDoesNotOccur(long delay) throws InterruptedException
+        {
+            _failoverLatch.await(delay, TimeUnit.MILLISECONDS);
+            assertEquals("Failover occurred unexpectedly", 1L, _failoverLatch.getCount());
+        }
+
+
+        @Override
+        public void failoverComplete()
+        {
+            _failoverLatch.countDown();
+        }
+
+        @Override
+        public void bytesSent(long count)
+        {
+        }
+
+        @Override
+        public void bytesReceived(long count)
+        {
+        }
+    }
+
+}
diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/HAClusterManagementTest.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/HAClusterManagementTest.java
new file mode 100644
index 0000000000..1afa45fd5a
--- /dev/null
+++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/HAClusterManagementTest.java
@@ -0,0 +1,233 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.store.berkeleydb;
+
+import static com.sleepycat.je.rep.ReplicatedEnvironment.State.DETACHED;
+import static com.sleepycat.je.rep.ReplicatedEnvironment.State.MASTER;
+import static com.sleepycat.je.rep.ReplicatedEnvironment.State.REPLICA;
+import static com.sleepycat.je.rep.ReplicatedEnvironment.State.UNKNOWN;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Set;
+
+import javax.jms.Connection;
+import javax.management.openmbean.CompositeData;
+import javax.management.openmbean.TabularData;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.jms.ConnectionURL;
+import org.apache.qpid.test.utils.JMXTestUtils;
+import org.apache.qpid.test.utils.QpidBrokerTestCase;
+
+import com.sleepycat.je.EnvironmentFailureException;
+
+/**
+ * System test verifying the ability to control a cluster via the Management API.
+ *
+ * @see HAClusterBlackboxTest
+ */
+public class HAClusterManagementTest extends QpidBrokerTestCase
+{
+    protected static final Logger LOGGER = Logger.getLogger(HAClusterManagementTest.class);
+
+    private static final Set<String> NON_MASTER_STATES = new HashSet<String>(Arrays.asList(REPLICA.toString(), DETACHED.toString(), UNKNOWN.toString()));;
+    private static final String VIRTUAL_HOST = "test";
+
+    private static final String MANAGED_OBJECT_QUERY = "org.apache.qpid:type=BDBHAMessageStore,name=" + VIRTUAL_HOST;
+    private static final int NUMBER_OF_NODES = 4;
+
+    private final HATestClusterCreator _clusterCreator = new HATestClusterCreator(this, VIRTUAL_HOST, NUMBER_OF_NODES);
+    private final JMXTestUtils _jmxUtils = new JMXTestUtils(this);
+
+    private ConnectionURL _brokerFailoverUrl;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        _brokerType = BrokerType.SPAWNED;
+
+        assertTrue(isJavaBroker());
+        assertTrue(isBrokerStorePersistent());
+        _jmxUtils.setUp();
+
+        _clusterCreator.configureClusterNodes();
+        _brokerFailoverUrl = _clusterCreator.getConnectionUrlForAllClusterNodes();
+        _clusterCreator.startCluster();
+
+        super.setUp();
+    }
+
+    @Override
+    protected void tearDown() throws Exception
+    {
+        try
+        {
+            _jmxUtils.close();
+        }
+        finally
+        {
+            super.tearDown();
+        }
+    }
+
+    @Override
+    public void startBroker() throws Exception
+    {
+        // Don't start default broker provided by QBTC.
+    }
+
+    public void testReadonlyMBeanAttributes() throws Exception
+    {
+        final int brokerPortNumber = getBrokerPortNumbers().iterator().next();
+        final int bdbPortNumber = _clusterCreator.getBdbPortForBrokerPort(brokerPortNumber);
+
+        ManagedBDBHAMessageStore storeBean = getStoreBeanForNodeAtBrokerPort(brokerPortNumber);
+        assertEquals("Unexpected store group name", _clusterCreator.getGroupName(), storeBean.getGroupName());
+        assertEquals("Unexpected store node name", _clusterCreator.getNodeNameForNodeAt(bdbPortNumber), storeBean.getNodeName());
+        assertEquals("Unexpected store node host port",_clusterCreator.getNodeHostPortForNodeAt(bdbPortNumber), storeBean.getNodeHostPort());
+        assertEquals("Unexpected store helper host port", _clusterCreator.getHelperHostPort(), storeBean.getHelperHostPort());
+        // As we have chosen an arbitrary broker from the cluster, we cannot predict its state
+        assertNotNull("Store state must not be null", storeBean.getNodeState());
+    }
+
+    public void testStateOfActiveBrokerIsMaster() throws Exception
+    {
+        final Connection activeConnection = getConnection(_brokerFailoverUrl);
+        final int activeBrokerPortNumber = _clusterCreator.getBrokerPortNumberFromConnection(activeConnection);
+
+        ManagedBDBHAMessageStore storeBean = getStoreBeanForNodeAtBrokerPort(activeBrokerPortNumber);
+        assertEquals("Unexpected store state", MASTER.toString(), storeBean.getNodeState());
+    }
+
+    public void testStateOfNonActiveBrokerIsNotMaster() throws Exception
+    {
+        final Connection activeConnection = getConnection(_brokerFailoverUrl);
+        final int inactiveBrokerPortNumber = _clusterCreator.getPortNumberOfAnInactiveBroker(activeConnection);
+        ManagedBDBHAMessageStore storeBean = getStoreBeanForNodeAtBrokerPort(inactiveBrokerPortNumber);
+        final String nodeState = storeBean.getNodeState();
+        assertTrue("Unexpected store state : " + nodeState, NON_MASTER_STATES.contains(nodeState));
+    }
+
+    public void testGroupMembers() throws Exception
+    {
+        final int brokerPortNumber = getBrokerPortNumbers().iterator().next();
+
+        ManagedBDBHAMessageStore storeBean = getStoreBeanForNodeAtBrokerPort(brokerPortNumber);
+        final TabularData groupMembers = storeBean.getAllNodesInGroup();
+        assertNotNull(groupMembers);
+
+        final int numberOfDataRows = groupMembers.size();
+        assertEquals("Unexpected number of data rows", NUMBER_OF_NODES ,numberOfDataRows);
+
+        for(int bdbPortNumber : _clusterCreator.getBdbPortNumbers())
+        {
+            final String nodeName = _clusterCreator.getNodeNameForNodeAt(bdbPortNumber);
+            final String nodeHostPort = _clusterCreator.getNodeHostPortForNodeAt(bdbPortNumber);
+
+            CompositeData row = groupMembers.get(new Object[] {nodeName});
+            assertNotNull("Table does not contain row for node name " + nodeName, row);
+            assertEquals(nodeHostPort, row.get(BDBHAMessageStore.GRP_MEM_COL_NODE_HOST_PORT));
+        }
+    }
+
+    public void testRemoveNodeFromGroup() throws Exception
+    {
+        final Iterator<Integer> brokerPortNumberIterator = getBrokerPortNumbers().iterator();
+        final int brokerPortNumberToMakeObservation = brokerPortNumberIterator.next();
+        final int brokerPortNumberToBeRemoved = brokerPortNumberIterator.next();
+        final ManagedBDBHAMessageStore storeBean = getStoreBeanForNodeAtBrokerPort(brokerPortNumberToMakeObservation);
+        final int numberOfDataRows = storeBean.getAllNodesInGroup().size();
+        assertEquals("Unexpected number of data rows before test", NUMBER_OF_NODES ,numberOfDataRows);
+
+        final String removedNodeName = _clusterCreator.getNodeNameForNodeAt(_clusterCreator.getBdbPortForBrokerPort(brokerPortNumberToBeRemoved));
+        _clusterCreator.stopNode(brokerPortNumberToBeRemoved);
+        storeBean.removeNodeFromGroup(removedNodeName);
+
+        final int numberOfDataRowsAfterRemoval = storeBean.getAllNodesInGroup().size();
+        assertEquals("Unexpected number of data rows before test", NUMBER_OF_NODES - 1,numberOfDataRowsAfterRemoval);
+    }
+
+    /**
+     * Updates the address of a node.
+     *
+     * If the broker (node) can subsequently start without error then the update was a success, hence no need for an explicit
+     * assert.
+     *
+     * @see #testRestartNodeWithNewPortNumberWithoutFirstCallingUpdateAddressThrowsAnException() for converse case
+     */
+    public void testUpdateAddress() throws Exception
+    {
+        final Iterator<Integer> brokerPortNumberIterator = getBrokerPortNumbers().iterator();
+        final int brokerPortNumberToPerformUpdate = brokerPortNumberIterator.next();
+        final int brokerPortNumberToBeMoved = brokerPortNumberIterator.next();
+        final ManagedBDBHAMessageStore storeBean = getStoreBeanForNodeAtBrokerPort(brokerPortNumberToPerformUpdate);
+
+        _clusterCreator.stopNode(brokerPortNumberToBeMoved);
+
+        final int oldBdbPort = _clusterCreator.getBdbPortForBrokerPort(brokerPortNumberToBeMoved);
+        final int newBdbPort = getNextAvailable(oldBdbPort + 1);
+
+        storeBean.updateAddress(_clusterCreator.getNodeNameForNodeAt(oldBdbPort), _clusterCreator.getIpAddressOfBrokerHost(), newBdbPort);
+
+        _clusterCreator.modifyClusterNodeBdbAddress(brokerPortNumberToBeMoved, newBdbPort);
+
+        _clusterCreator.startNode(brokerPortNumberToBeMoved);
+    }
+
+    /**
+     * @see #testUpdateAddress()
+     */
+    public void testRestartNodeWithNewPortNumberWithoutFirstCallingUpdateAddressThrowsAnException() throws Exception
+    {
+        final Iterator<Integer> brokerPortNumberIterator = getBrokerPortNumbers().iterator();
+        final int brokerPortNumberToBeMoved = brokerPortNumberIterator.next();
+
+        _clusterCreator.stopNode(brokerPortNumberToBeMoved);
+
+        final int oldBdbPort = _clusterCreator.getBdbPortForBrokerPort(brokerPortNumberToBeMoved);
+        final int newBdbPort = getNextAvailable(oldBdbPort + 1);
+
+        // now deliberately don't call updateAddress
+
+        _clusterCreator.modifyClusterNodeBdbAddress(brokerPortNumberToBeMoved, newBdbPort);
+
+        try
+        {
+            _clusterCreator.startNode(brokerPortNumberToBeMoved);
+            fail("Exception not thrown");
+        }
+        catch(RuntimeException rte)
+        {
+            //check cause was BDBs EnvironmentFailureException
+            assertTrue(rte.getMessage().contains(EnvironmentFailureException.class.getName()));
+            // PASS
+        }
+    }
+
+    private ManagedBDBHAMessageStore getStoreBeanForNodeAtBrokerPort(
+            final int activeBrokerPortNumber) throws Exception
+    {
+        _jmxUtils.open(activeBrokerPortNumber);
+
+        return _jmxUtils.getManagedObject(ManagedBDBHAMessageStore.class, MANAGED_OBJECT_QUERY);
+    }
+}
diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/HAClusterTwoNodeTest.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/HAClusterTwoNodeTest.java
new file mode 100644
index 0000000000..5f995ae25d
--- /dev/null
+++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/HAClusterTwoNodeTest.java
@@ -0,0 +1,234 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.store.berkeleydb;
+
+import java.io.File;
+
+import javax.jms.Connection;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.Session;
+
+import org.apache.qpid.jms.ConnectionURL;
+import org.apache.qpid.test.utils.JMXTestUtils;
+import org.apache.qpid.test.utils.QpidBrokerTestCase;
+
+import com.sleepycat.je.rep.ReplicationConfig;
+
+public class HAClusterTwoNodeTest extends QpidBrokerTestCase
+{
+    private static final long RECEIVE_TIMEOUT = 5000l;
+
+    private static final String VIRTUAL_HOST = "test";
+
+    private static final String MANAGED_OBJECT_QUERY = "org.apache.qpid:type=BDBHAMessageStore,name=" + VIRTUAL_HOST;
+    private static final int NUMBER_OF_NODES = 2;
+
+    private final HATestClusterCreator _clusterCreator = new HATestClusterCreator(this, VIRTUAL_HOST, NUMBER_OF_NODES);
+    private final JMXTestUtils _jmxUtils = new JMXTestUtils(this);
+
+    private ConnectionURL _brokerFailoverUrl;
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        _brokerType = BrokerType.SPAWNED;
+
+        assertTrue(isJavaBroker());
+        assertTrue(isBrokerStorePersistent());
+        _jmxUtils.setUp();
+
+        super.setUp();
+    }
+
+    @Override
+    protected void tearDown() throws Exception
+    {
+        try
+        {
+            _jmxUtils.close();
+        }
+        finally
+        {
+            super.tearDown();
+        }
+    }
+
+    @Override
+    public void startBroker() throws Exception
+    {
+        // Don't start default broker provided by QBTC.
+    }
+
+    private void startCluster(boolean autoDesignedPrimary) throws Exception
+    {
+        setSystemProperty("java.util.logging.config.file",
+                System.getProperty(QPID_HOME) + File.separator + "etc" + File.separator + "log.properties");
+
+        String vhostPrefix = "virtualhosts.virtualhost." + VIRTUAL_HOST;
+
+        setConfigurationProperty(vhostPrefix + ".store.repConfig(0).name", ReplicationConfig.INSUFFICIENT_REPLICAS_TIMEOUT);
+        setConfigurationProperty(vhostPrefix + ".store.repConfig(0).value", "2 s");
+
+        setConfigurationProperty(vhostPrefix + ".store.repConfig(1).name", ReplicationConfig.ELECTIONS_PRIMARY_RETRIES);
+        setConfigurationProperty(vhostPrefix + ".store.repConfig(1).value", "0");
+
+        _clusterCreator.configureClusterNodes();
+        _clusterCreator.setAutoDesignatedPrimary(autoDesignedPrimary);
+        _brokerFailoverUrl = _clusterCreator.getConnectionUrlForAllClusterNodes();
+        _clusterCreator.startCluster();
+    }
+
+    /**
+     * Tests that a two node cluster, in which the master CAN automatically designate itself primary
+     * (after becoming master) continues to operate after being shut down and restarted.
+     *
+     * The test does not concern itself with which broker becomes master at any given point
+     * (which is likely to swap during the test).
+     */
+    public void testClusterRestartWithAutoDesignatedPrimary() throws Exception
+    {
+        testClusterRestartImpl(true);
+    }
+
+    /**
+     * Tests that a two node cluster, in which the master can NOT automatically designate itself
+     * primary (after becoming master) continues to operate after being shut down and restarted.
+     *
+     * The test does not concern itself with which broker becomes master at any given point
+     * (which is likely to swap during the test).
+     */
+    public void testClusterRestartWithoutAutoDesignatedPrimary() throws Exception
+    {
+        testClusterRestartImpl(false);
+    }
+
+    private void testClusterRestartImpl(boolean autoDesignatedPrimary) throws Exception
+    {
+        startCluster(autoDesignatedPrimary);
+        final Connection initialConnection = getConnection(_brokerFailoverUrl);
+        assertProducingConsuming(initialConnection);
+        initialConnection.close();
+        _clusterCreator.stopCluster();
+        _clusterCreator.startClusterParallel();
+        final Connection secondConnection = getConnection(_brokerFailoverUrl);
+        assertProducingConsuming(secondConnection);
+        secondConnection.close();
+    }
+
+    /**
+     * This test make sure than JMS operations are still working after stopping replica
+     * when master is designated primary (which is by default).
+     * <p>
+     * When master is not designated primary this test should fail.
+     */
+    public void testAutoDesignatedPrimaryContinuesAfterSecondaryStopped() throws Exception
+    {
+        startCluster(true);
+        _clusterCreator.stopNode(_clusterCreator.getBrokerPortNumberOfSecondaryNode());
+        final Connection connection = getConnection(_brokerFailoverUrl);
+        assertNotNull("Expected to get a valid connection to primary", connection);
+        assertProducingConsuming(connection);
+    }
+
+    public void testPersistentOperationsFailOnNonAutoDesignatedPrimarysAfterSecondaryStopped() throws Exception
+    {
+        
+        startCluster(false);
+        _clusterCreator.stopNode(_clusterCreator.getBrokerPortNumberOfSecondaryNode());
+        final Connection connection = getConnection(_brokerFailoverUrl);
+        assertNotNull("Expected to get a valid connection to primary", connection);
+        try
+        {
+            assertProducingConsuming(connection);
+            fail("JMS peristent operations succeded on Master 'not designated primary' buy they should fail as replica is not available");
+        }
+        catch(JMSException e)
+        {
+            // JMSException should be thrown on transaction start/commit
+        }
+    }
+
+    public void testSecondaryDoesNotBecomePrimaryWhenAutoDesignatedPrimaryStopped() throws Exception
+    {
+        startCluster(true);
+        _clusterCreator.stopNode(_clusterCreator.getBrokerPortNumberOfPrimary());
+
+        try
+        {
+            getConnection(_brokerFailoverUrl);
+            fail("Connection not expected");
+        }
+        catch (JMSException e)
+        {
+            // PASS
+        }
+    }
+
+    public void testInitialDesignatedPrimaryStateOfNodes() throws Exception
+    {
+        startCluster(true);
+        final ManagedBDBHAMessageStore primaryStoreBean = getStoreBeanForNodeAtBrokerPort(_clusterCreator.getBrokerPortNumberOfPrimary());
+        assertTrue("Expected primary node to be set as designated primary", primaryStoreBean.getDesignatedPrimary());
+
+        final ManagedBDBHAMessageStore secondaryStoreBean = getStoreBeanForNodeAtBrokerPort(_clusterCreator.getBrokerPortNumberOfSecondaryNode());
+        assertFalse("Expected secondary node to NOT be set as designated primary", secondaryStoreBean.getDesignatedPrimary());
+    }
+
+    public void testSecondaryDesignatedAsPrimaryAfterOrginalPrimaryStopped() throws Exception
+    {
+        startCluster(true);
+        _clusterCreator.stopNode(_clusterCreator.getBrokerPortNumberOfPrimary());
+        final ManagedBDBHAMessageStore storeBean = getStoreBeanForNodeAtBrokerPort(_clusterCreator.getBrokerPortNumberOfSecondaryNode());
+
+        assertFalse("Expected node to NOT be set as designated primary", storeBean.getDesignatedPrimary());
+        storeBean.setDesignatedPrimary(true);
+        assertTrue("Expected node to now be set as designated primary", storeBean.getDesignatedPrimary());
+
+        final Connection connection = getConnection(_brokerFailoverUrl);
+        assertNotNull("Expected to get a valid connection to new primary", connection);
+        assertProducingConsuming(connection);
+    }
+
+    private ManagedBDBHAMessageStore getStoreBeanForNodeAtBrokerPort(
+            final int activeBrokerPortNumber) throws Exception
+    {
+        _jmxUtils.open(activeBrokerPortNumber);
+
+        ManagedBDBHAMessageStore storeBean = _jmxUtils.getManagedObject(ManagedBDBHAMessageStore.class, MANAGED_OBJECT_QUERY);
+        return storeBean;
+    }
+
+    private void assertProducingConsuming(final Connection connection) throws JMSException, Exception
+    {
+        Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
+        Destination destination = session.createQueue(getTestQueueName());
+        MessageConsumer consumer = session.createConsumer(destination);
+        sendMessage(session, destination, 1);
+        connection.start();
+        Message m1 = consumer.receive(RECEIVE_TIMEOUT);
+        assertNotNull("Message 1 is not received", m1);
+        assertEquals("Unexpected first message received", 0, m1.getIntProperty(INDEX));
+        session.commit();
+    }
+
+}
diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/HAClusterWhiteboxTest.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/HAClusterWhiteboxTest.java
new file mode 100644
index 0000000000..4b64466ff2
--- /dev/null
+++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/HAClusterWhiteboxTest.java
@@ -0,0 +1,288 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.store.berkeleydb;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import javax.jms.Connection;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.Queue;
+import javax.jms.Session;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.test.utils.QpidBrokerTestCase;
+import org.apache.qpid.url.URLSyntaxException;
+
+import com.sleepycat.je.rep.InsufficientLogException;
+
+/**
+ * The HA white box tests test the BDB cluster where the test retains the knowledge of the
+ * individual test nodes.  It uses this knowledge to examine the nodes  to ensure that they
+ * remain in the correct state throughout the test.
+ *
+ * @see HAClusterBlackboxTest
+ */
+public class HAClusterWhiteboxTest extends QpidBrokerTestCase
+{
+    protected static final Logger LOGGER = Logger.getLogger(HAClusterWhiteboxTest.class);
+
+    private static final String VIRTUAL_HOST = "test";
+
+    private final int NUMBER_OF_NODES = 3;
+    private final HATestClusterCreator _clusterCreator = new HATestClusterCreator(this, VIRTUAL_HOST, NUMBER_OF_NODES);
+
+    // TODO Test for node falling behind
+    // TODO Factory refactoring?? // MessageStore construction??
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        _brokerType = BrokerType.SPAWNED;
+
+        assertTrue(isJavaBroker());
+        assertTrue(isBrokerStorePersistent());
+
+        setBrokerEnvironment("QPID_OPTS", "-Djava.util.logging.config.file=" + System.getProperty(QPID_HOME)
+                + File.separator + "etc" + File.separator + "log.properties");
+
+        _clusterCreator.configureClusterNodes();
+        _clusterCreator.startCluster();
+
+        super.setUp();
+    }
+
+    @Override
+    public void startBroker() throws Exception
+    {
+        // Don't start default broker provided by QBTC.
+    }
+
+    public void testClusterPermitsConnectionToOnlyOneNode() throws Exception
+    {
+        int connectionSuccesses = 0;
+        int connectionFails = 0;
+
+        for (int brokerPortNumber : getBrokerPortNumbers())
+        {
+            try
+            {
+                getConnection(_clusterCreator.getConnectionUrlForSingleNode(brokerPortNumber));
+                connectionSuccesses++;
+            }
+            catch(JMSException e)
+            {
+                assertTrue(e.getMessage().contains("Virtual host '" + VIRTUAL_HOST + "' is not active"));
+                connectionFails++;
+            }
+        }
+
+        assertEquals("Unexpected number of failed connections", NUMBER_OF_NODES - 1, connectionFails);
+        assertEquals("Unexpected number of successful connections", 1, connectionSuccesses);
+    }
+
+    public void testClusterThatLosesNodeStillAllowsConnection() throws Exception
+    {
+        final Connection initialConnection = getConnectionToNodeInCluster();
+        assertNotNull(initialConnection);
+
+        killConnectionBrokerAndWaitForNewMasterElection(initialConnection);
+
+        final Connection subsequentConnection = getConnectionToNodeInCluster();
+        assertNotNull(subsequentConnection);
+
+        // verify that JMS persistence operations are working
+        assertProducingConsuming(subsequentConnection);
+
+        closeConnection(initialConnection);
+    }
+
+    public void testClusterThatLosesAllButOneNodeRefusesConnection() throws Exception
+    {
+        final Connection initialConnection = getConnectionToNodeInCluster();
+        assertNotNull(initialConnection);
+
+        killConnectionBrokerAndWaitForNewMasterElection(initialConnection);
+
+        final Connection subsequentConnection = getConnectionToNodeInCluster();
+        assertNotNull(subsequentConnection);
+        final int subsequentPortNumber = _clusterCreator.getBrokerPortNumberFromConnection(subsequentConnection);
+
+        killBroker(subsequentPortNumber);
+
+        final Connection finalConnection = getConnectionToNodeInCluster();
+        assertNull(finalConnection);
+
+        closeConnection(initialConnection);
+    }
+
+    public void testClusterWithRestartedNodeStillAllowsConnection() throws Exception
+    {
+        final Connection connection = getConnectionToNodeInCluster();
+        assertNotNull(connection);
+
+        final int brokerPortNumber = _clusterCreator.getBrokerPortNumberFromConnection(connection);
+
+        _clusterCreator.stopNode(brokerPortNumber);
+        _clusterCreator.startNode(brokerPortNumber);
+
+        final Connection subsequentConnection = getConnectionToNodeInCluster();
+        assertNotNull(subsequentConnection);
+    }
+
+    public void testClusterLosingNodeRetainsData() throws Exception
+    {
+        final Connection initialConnection = getConnectionToNodeInCluster();
+
+        final String queueNamePrefix = getTestQueueName();
+        final String inbuiltExchangeQueueUrl = "direct://amq.direct/" + queueNamePrefix + "1/" + queueNamePrefix + "1?durable='true'";
+        final String customExchangeQueueUrl = "direct://my.exchange/" + queueNamePrefix + "2/" + queueNamePrefix + "2?durable='true'";
+
+        populateBrokerWithData(initialConnection, inbuiltExchangeQueueUrl, customExchangeQueueUrl);
+
+        killConnectionBrokerAndWaitForNewMasterElection(initialConnection);
+
+        final Connection subsequentConnection = getConnectionToNodeInCluster();
+
+        assertNotNull("no valid connection obtained", subsequentConnection);
+
+        checkBrokerData(subsequentConnection, inbuiltExchangeQueueUrl, customExchangeQueueUrl);
+    }
+
+    public void testRecoveryOfOutOfDateNode() throws Exception
+    {
+        /*
+         * TODO: Implement
+         *
+         * Cant yet find a way to control cleaning in a deterministic way to allow provoking
+         * a node to become out of date. We do now know that even a new joiner to the group
+         * can throw the InsufficientLogException, so ensuring an existing cluster of nodes has
+         * done *any* cleaning and then adding a new node should be sufficient to cause this.
+         */
+    }
+
+    private void populateBrokerWithData(final Connection connection, final String... queueUrls) throws JMSException, Exception
+    {
+        populateBrokerWithData(connection, 1, queueUrls);
+    }
+
+    private void populateBrokerWithData(final Connection connection, int noOfMessages, final String... queueUrls) throws JMSException, Exception
+    {
+        final Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
+        for (final String queueUrl : queueUrls)
+        {
+            final Queue queue = session.createQueue(queueUrl);
+            session.createConsumer(queue).close();
+            sendMessage(session, queue, noOfMessages);
+        }
+    }
+
+    private void checkBrokerData(final Connection connection, final String... queueUrls) throws JMSException
+    {
+        connection.start();
+        final Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
+        for (final String queueUrl : queueUrls)
+        {
+            final Queue queue = session.createQueue(queueUrl);
+            final MessageConsumer consumer = session.createConsumer(queue);
+            final Message message = consumer.receive(1000);
+            session.commit();
+            assertNotNull("Queue " + queue + " should have message", message);
+            assertEquals("Queue " + queue + " message has unexpected content", 0, message.getIntProperty(INDEX));
+        }
+    }
+
+    private Connection getConnectionToNodeInCluster() throws URLSyntaxException
+    {
+        Connection connection = null;
+        Set<Integer> runningBrokerPorts = getBrokerPortNumbers();
+
+        for (int brokerPortNumber : runningBrokerPorts)
+        {
+            try
+            {
+                connection = getConnection(_clusterCreator.getConnectionUrlForSingleNode(brokerPortNumber));
+                break;
+            }
+            catch(JMSException je)
+            {
+                assertTrue(je.getMessage().contains("Virtual host '" + VIRTUAL_HOST + "' is not active"));
+            }
+        }
+        return connection;
+    }
+
+    private void killConnectionBrokerAndWaitForNewMasterElection(final Connection initialConnection) throws IOException,
+            InterruptedException
+    {
+        try
+        {
+            // NewMasterEvent is received twice: first for the existing master,
+            // second for a new master
+            CountDownLatch newMasterLatch = new CountDownLatch(2);
+            _clusterCreator.startMonitorNode();
+            _clusterCreator.statListeningForNewMasterEvent(newMasterLatch);
+
+            final int initialPortNumber = _clusterCreator.getBrokerPortNumberFromConnection(initialConnection);
+            killBroker(initialPortNumber);
+
+            assertTrue("New master was not elected", newMasterLatch.await(30, TimeUnit.SECONDS));
+        }
+        finally
+        {
+            _clusterCreator.shutdownMonitor();
+        }
+    }
+
+    private void assertProducingConsuming(final Connection connection) throws JMSException, Exception
+    {
+        Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
+        Destination destination = session.createQueue(getTestQueueName());
+        MessageConsumer consumer = session.createConsumer(destination);
+        sendMessage(session, destination, 2);
+        connection.start();
+        Message m1 = consumer.receive(RECEIVE_TIMEOUT);
+        assertNotNull("Message 1 is not received", m1);
+        assertEquals("Unexpected first message received", 0, m1.getIntProperty(INDEX));
+        Message m2 = consumer.receive(RECEIVE_TIMEOUT);
+        assertNotNull("Message 2 is not received", m2);
+        assertEquals("Unexpected second message received", 1, m2.getIntProperty(INDEX));
+        session.commit();
+    }
+
+    private void closeConnection(final Connection initialConnection)
+    {
+        try
+        {
+            initialConnection.close();
+        }
+        catch(Exception e)
+        {
+            // ignore.
+            // java.net.SocketException is seen sometimes on active connection
+        }
+    }
+}
diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStoreFactory.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/HAMessageStoreSmokeTest.java
index 7e5ef3f94c..eaa3c3eba4 100644
--- a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStoreFactory.java
+++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/HAMessageStoreSmokeTest.java
@@ -19,22 +19,25 @@
  */
 package org.apache.qpid.server.store.berkeleydb;
 
-import org.apache.qpid.server.store.MessageStore;
-import org.apache.qpid.server.store.MessageStoreFactory;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.commons.configuration.XMLConfiguration;
+import org.apache.qpid.test.utils.QpidTestCase;
 
-public class BDBMessageStoreFactory implements MessageStoreFactory
+public class HAMessageStoreSmokeTest extends QpidTestCase
 {
+    private final BDBHAMessageStore _store = new BDBHAMessageStore();
+    private final XMLConfiguration _config = new XMLConfiguration();
 
-    @Override
-    public MessageStore createMessageStore()
+    public void testMissingHAConfigThrowsException() throws Exception
     {
-        return new BDBMessageStore();
+        try
+        {
+            _store.configure("test", _config);
+            fail("Expected an exception to be thrown");
+        }
+        catch (ConfigurationException ce)
+        {
+            assertTrue(ce.getMessage().contains("BDB HA configuration key not found"));
+        }
     }
-
-    @Override
-    public String getStoreClassName()
-    {
-        return BDBMessageStore.class.getSimpleName();
-    }
-
-}
+}
\ No newline at end of file
diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/HATestClusterCreator.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/HATestClusterCreator.java
new file mode 100644
index 0000000000..43cfa5f4d5
--- /dev/null
+++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/HATestClusterCreator.java
@@ -0,0 +1,479 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.store.berkeleydb;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import javax.jms.Connection;
+
+import org.apache.commons.configuration.XMLConfiguration;
+import org.apache.commons.lang.StringUtils;
+import org.apache.log4j.Logger;
+import org.apache.qpid.client.AMQConnection;
+import org.apache.qpid.client.AMQConnectionURL;
+import org.apache.qpid.test.utils.QpidBrokerTestCase;
+import org.apache.qpid.url.URLSyntaxException;
+
+import com.sleepycat.je.rep.ReplicationNode;
+import com.sleepycat.je.rep.monitor.GroupChangeEvent;
+import com.sleepycat.je.rep.monitor.JoinGroupEvent;
+import com.sleepycat.je.rep.monitor.LeaveGroupEvent;
+import com.sleepycat.je.rep.monitor.Monitor;
+import com.sleepycat.je.rep.monitor.MonitorChangeListener;
+import com.sleepycat.je.rep.monitor.MonitorConfig;
+import com.sleepycat.je.rep.monitor.NewMasterEvent;
+
+public class HATestClusterCreator
+{
+    protected static final Logger LOGGER = Logger.getLogger(HATestClusterCreator.class);
+
+    private static final String MANY_BROKER_URL_FORMAT = "amqp://guest:guest@/%s?brokerlist='%s'&failover='roundrobin?cyclecount='%d''";
+    private static final String BROKER_PORTION_FORMAT = "tcp://localhost:%d?connectdelay='%d',retries='%d'";
+    private static final String SINGLE_BROKER_URL_FORMAT = "amqp://guest:guest@/%s?brokerlist='tcp://localhost:%d?connectdelay='%d',retries='%d''";
+
+    private static final int CYCLECOUNT = 2;
+    private static final int RETRIES = 2;
+    private static final int CONNECTDELAY = 1000;
+
+    private final QpidBrokerTestCase _testcase;
+    private final Map<Integer, Integer> _brokerPortToBdbPortMap = new HashMap<Integer, Integer>();
+    private final Map<Integer, BrokerConfigHolder> _brokerConfigurations = new TreeMap<Integer, BrokerConfigHolder>();
+    private final String _virtualHostName;
+    private final String _configKeyPrefix;
+
+    private final String _ipAddressOfBroker;
+    private final String _groupName ;
+    private final int _numberOfNodes;
+    private int _bdbHelperPort;
+    private int _primaryBrokerPort;
+    private Monitor _monitor;
+
+    public HATestClusterCreator(QpidBrokerTestCase testcase, String virtualHostName, int numberOfNodes)
+    {
+        _testcase = testcase;
+        _virtualHostName = virtualHostName;
+        _groupName = "group" + _testcase.getName();
+        _ipAddressOfBroker = getIpAddressOfBrokerHost();
+        _numberOfNodes = numberOfNodes;
+        _configKeyPrefix = "virtualhosts.virtualhost." + _virtualHostName + ".store.";
+        _bdbHelperPort = 0;
+    }
+
+    public void configureClusterNodes() throws Exception
+    {
+        int brokerPort = _testcase.findFreePort();
+
+        for (int i = 0; i < _numberOfNodes; i++)
+        {
+            int bdbPort = _testcase.getNextAvailable(brokerPort + 1);
+            _brokerPortToBdbPortMap.put(brokerPort, bdbPort);
+
+            LOGGER.debug("Cluster broker port " + brokerPort + ", bdb replication port " + bdbPort);
+            if (_bdbHelperPort == 0)
+            {
+                _bdbHelperPort = bdbPort;
+            }
+
+            configureClusterNode(brokerPort, bdbPort);
+            collectConfig(brokerPort, _testcase.getTestConfiguration(), _testcase.getTestVirtualhosts());
+
+            brokerPort = _testcase.getNextAvailable(bdbPort + 1);
+        }
+    }
+
+    public void setAutoDesignatedPrimary(boolean autoDesignatedPrimary) throws Exception
+    {
+        if (_numberOfNodes != 2)
+        {
+            throw new IllegalArgumentException("Only two nodes groups have the concept of primary");
+        }
+
+        final Entry<Integer, BrokerConfigHolder> brokerConfigEntry = _brokerConfigurations.entrySet().iterator().next();
+        final String configKey = getConfigKey("highAvailability.autoDesignatedPrimary");
+        brokerConfigEntry.getValue().getTestVirtualhosts().setProperty(configKey, Boolean.toString(autoDesignatedPrimary));
+        _primaryBrokerPort = brokerConfigEntry.getKey();
+    }
+
+    /**
+     * @param configKeySuffix "highAvailability.designatedPrimary", for example
+     * @return "virtualhost.test.store.highAvailability.designatedPrimary", for example
+     */
+    private String getConfigKey(String configKeySuffix)
+    {
+        final String configKey = StringUtils.substringAfter(_configKeyPrefix + configKeySuffix, "virtualhosts.");
+        return configKey;
+    }
+
+    public void startNode(final int brokerPortNumber) throws Exception
+    {
+        final BrokerConfigHolder brokerConfigHolder = _brokerConfigurations.get(brokerPortNumber);
+
+        _testcase.setTestConfiguration(brokerConfigHolder.getTestConfiguration());
+        _testcase.setTestVirtualhosts(brokerConfigHolder.getTestVirtualhosts());
+
+        _testcase.startBroker(brokerPortNumber);
+    }
+
+    public void startCluster() throws Exception
+    {
+        for (final Integer brokerPortNumber : _brokerConfigurations.keySet())
+        {
+            startNode(brokerPortNumber);
+        }
+    }
+
+    public void startClusterParallel() throws Exception
+    {
+        final ExecutorService executor = Executors.newFixedThreadPool(_brokerConfigurations.size());
+        try
+        {
+            List<Future<Object>> brokers = new CopyOnWriteArrayList<Future<Object>>();
+            for (final Integer brokerPortNumber : _brokerConfigurations.keySet())
+            {
+                final BrokerConfigHolder brokerConfigHolder = _brokerConfigurations.get(brokerPortNumber);
+                Future<Object> future = executor.submit(new Callable<Object>()
+                {
+                    public Object call()
+                    {
+                        try
+                        {
+                            _testcase.startBroker(brokerPortNumber, brokerConfigHolder.getTestConfiguration(),
+                                    brokerConfigHolder.getTestVirtualhosts());
+                            return "OK";
+                        }
+                        catch (Exception e)
+                        {
+                            return e;
+                        }
+                    }
+                });
+                brokers.add(future);
+            }
+            for (Future<Object> future : brokers)
+            {
+                Object result = future.get(30, TimeUnit.SECONDS);
+                LOGGER.debug("Node startup result:" + result);
+                if (result instanceof Exception)
+                {
+                    throw (Exception) result;
+                }
+                else if (!"OK".equals(result))
+                {
+                    throw new Exception("One of the cluster nodes is not started");
+                }
+            }
+        }
+        catch (Exception e)
+        {
+            stopCluster();
+            throw e;
+        }
+        finally
+        {
+            executor.shutdown();
+        }
+
+    }
+
+    public void stopNode(final int brokerPortNumber)
+    {
+        _testcase.stopBroker(brokerPortNumber);
+    }
+
+    public void stopCluster() throws Exception
+    {
+        shutdownMonitor();
+        for (final Integer brokerPortNumber : _brokerConfigurations.keySet())
+        {
+            try
+            {
+                stopNode(brokerPortNumber);
+            }
+            catch(Exception e)
+            {
+                LOGGER.warn("Failed to stop node on port:" + brokerPortNumber);
+            }
+        }
+    }
+
+    public int getBrokerPortNumberFromConnection(Connection connection)
+    {
+        final AMQConnection amqConnection = (AMQConnection)connection;
+        return amqConnection.getActiveBrokerDetails().getPort();
+    }
+
+    public int getPortNumberOfAnInactiveBroker(final Connection activeConnection)
+    {
+        final Set<Integer> allBrokerPorts = _testcase.getBrokerPortNumbers();
+        LOGGER.debug("Broker ports:" + allBrokerPorts);
+        final int activeBrokerPort = getBrokerPortNumberFromConnection(activeConnection);
+        allBrokerPorts.remove(activeBrokerPort);
+        LOGGER.debug("Broker ports:" + allBrokerPorts);
+        final int inactiveBrokerPort = allBrokerPorts.iterator().next();
+        return inactiveBrokerPort;
+    }
+
+    public int getBdbPortForBrokerPort(final int brokerPortNumber)
+    {
+        return _brokerPortToBdbPortMap.get(brokerPortNumber);
+    }
+
+    public Set<Integer> getBdbPortNumbers()
+    {
+        return new HashSet<Integer>(_brokerPortToBdbPortMap.values());
+    }
+
+    public AMQConnectionURL getConnectionUrlForAllClusterNodes() throws Exception
+    {
+        final StringBuilder brokerList = new StringBuilder();
+
+        for(Iterator<Integer> itr = _brokerPortToBdbPortMap.keySet().iterator(); itr.hasNext(); )
+        {
+            int brokerPortNumber = itr.next();
+
+            brokerList.append(String.format(BROKER_PORTION_FORMAT, brokerPortNumber, CONNECTDELAY, RETRIES));
+            if (itr.hasNext())
+            {
+                brokerList.append(";");
+            }
+        }
+
+        return new AMQConnectionURL(String.format(MANY_BROKER_URL_FORMAT, _virtualHostName, brokerList, CYCLECOUNT));
+    }
+
+    public AMQConnectionURL getConnectionUrlForSingleNode(final int brokerPortNumber) throws URLSyntaxException
+    {
+        String url = String.format(SINGLE_BROKER_URL_FORMAT, _virtualHostName, brokerPortNumber, CONNECTDELAY, RETRIES);
+        return new AMQConnectionURL(url);
+    }
+
+    public String getGroupName()
+    {
+        return _groupName;
+    }
+
+    public String getNodeNameForNodeAt(final int bdbPort)
+    {
+        return "node" + _testcase.getName() + bdbPort;
+    }
+
+    public String getNodeHostPortForNodeAt(final int bdbPort)
+    {
+        return _ipAddressOfBroker + ":" + bdbPort;
+    }
+
+    public String getHelperHostPort()
+    {
+        if (_bdbHelperPort == 0)
+        {
+            throw new IllegalStateException("Helper port not yet assigned.");
+        }
+
+        return _ipAddressOfBroker + ":" + _bdbHelperPort;
+    }
+
+    public void setHelperHostPort(int bdbHelperPort)
+    {
+        _bdbHelperPort = bdbHelperPort;
+    }
+
+    public int getBrokerPortNumberOfPrimary()
+    {
+        if (_numberOfNodes != 2)
+        {
+            throw new IllegalArgumentException("Only two nodes groups have the concept of primary");
+        }
+
+        return _primaryBrokerPort;
+    }
+
+    public int getBrokerPortNumberOfSecondaryNode()
+    {
+        final Set<Integer> portNumbers = getBrokerPortNumbersForNodes();
+        portNumbers.remove(getBrokerPortNumberOfPrimary());
+        return portNumbers.iterator().next();
+    }
+
+    public Set<Integer> getBrokerPortNumbersForNodes()
+    {
+        return new HashSet<Integer>(_brokerConfigurations.keySet());
+    }
+
+    private void configureClusterNode(final int brokerPort, final int bdbPort) throws Exception
+    {
+        final String nodeName = getNodeNameForNodeAt(bdbPort);
+
+        _testcase.setConfigurationProperty(_configKeyPrefix + "class", "org.apache.qpid.server.store.berkeleydb.BDBHAMessageStore");
+
+        _testcase.setConfigurationProperty(_configKeyPrefix + "highAvailability.groupName", _groupName);
+        _testcase.setConfigurationProperty(_configKeyPrefix + "highAvailability.nodeName", nodeName);
+        _testcase.setConfigurationProperty(_configKeyPrefix + "highAvailability.nodeHostPort", getNodeHostPortForNodeAt(bdbPort));
+        _testcase.setConfigurationProperty(_configKeyPrefix + "highAvailability.helperHostPort", getHelperHostPort());
+        // TODO replication policy
+    }
+
+    public String getIpAddressOfBrokerHost()
+    {
+        String brokerHost = _testcase.getBroker().getHost();
+        try
+        {
+            return InetAddress.getByName(brokerHost).getHostAddress();
+        }
+        catch (UnknownHostException e)
+        {
+            throw new RuntimeException("Could not determine IP address of host : " + brokerHost, e);
+        }
+    }
+
+    private void collectConfig(final int brokerPortNumber, XMLConfiguration testConfiguration, XMLConfiguration testVirtualhosts)
+    {
+        _brokerConfigurations.put(brokerPortNumber, new BrokerConfigHolder((XMLConfiguration) testConfiguration.clone(),
+                                                                    (XMLConfiguration) testVirtualhosts.clone()));
+    }
+
+    public class BrokerConfigHolder
+    {
+        private final XMLConfiguration _testConfiguration;
+        private final XMLConfiguration _testVirtualhosts;
+
+        public BrokerConfigHolder(XMLConfiguration testConfiguration, XMLConfiguration testVirtualhosts)
+        {
+            _testConfiguration = testConfiguration;
+            _testVirtualhosts = testVirtualhosts;
+        }
+
+        public XMLConfiguration getTestConfiguration()
+        {
+            return _testConfiguration;
+        }
+
+        public XMLConfiguration getTestVirtualhosts()
+        {
+            return _testVirtualhosts;
+        }
+    }
+
+    public void modifyClusterNodeBdbAddress(int brokerPortNumberToBeMoved, int newBdbPort)
+    {
+        final BrokerConfigHolder brokerConfigHolder = _brokerConfigurations.get(brokerPortNumberToBeMoved);
+        final XMLConfiguration virtualHostConfig = brokerConfigHolder.getTestVirtualhosts();
+
+        final String configKey = getConfigKey("highAvailability.nodeHostPort");
+        final String oldBdbHostPort = virtualHostConfig.getString(configKey);
+
+        final String[] oldHostAndPort = StringUtils.split(oldBdbHostPort, ":");
+        final String oldHost = oldHostAndPort[0];
+
+        final String newBdbHostPort = oldHost + ":" + newBdbPort;
+
+        virtualHostConfig.setProperty(configKey, newBdbHostPort);
+        collectConfig(brokerPortNumberToBeMoved, brokerConfigHolder.getTestConfiguration(), virtualHostConfig);
+    }
+
+    public void startMonitorNode()
+    {
+        shutdownMonitor();
+
+        MonitorConfig config = new MonitorConfig();
+        config.setGroupName(_groupName);
+        int monitorPort = _testcase.findFreePort();
+        config.setNodeName(getNodeNameForNodeAt(monitorPort));
+        config.setNodeHostPort("" + monitorPort);
+        config.setHelperHosts(getHelperHostPort());
+
+        _monitor = new Monitor(config);
+
+        ReplicationNode currentMaster = _monitor.register();
+        LOGGER.info("Current master " + currentMaster.getName());
+    }
+
+    public void startListening(MonitorChangeListener listener) throws IOException
+    {
+        _monitor.startListener(listener);
+    }
+
+    public void statListeningForNewMasterEvent(final CountDownLatch latch) throws IOException
+    {
+        startListening(new MonitorChangeListenerSupport(){
+            @Override
+            public void notify(NewMasterEvent newMasterEvent)
+            {
+                LOGGER.debug("New master is elected " + newMasterEvent.getMasterName());
+                latch.countDown();
+            }
+        });
+    }
+
+    public void shutdownMonitor()
+    {
+        if (_monitor != null)
+        {
+            try
+            {
+                _monitor.shutdown();
+            }
+            catch (Exception e)
+            {
+                LOGGER.warn("Monitor shutdown error:", e);
+            }
+        }
+    }
+
+    public static class MonitorChangeListenerSupport implements MonitorChangeListener
+    {
+
+        @Override
+        public void notify(NewMasterEvent newMasterEvent)
+        {
+        }
+
+        @Override
+        public void notify(GroupChangeEvent groupChangeEvent)
+        {
+        }
+
+        @Override
+        public void notify(JoinGroupEvent joinGroupEvent)
+        {
+        }
+
+        @Override
+        public void notify(LeaveGroupEvent leaveGroupEvent)
+        {
+        }
+    }
+}
diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgraderTest.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgraderTest.java
index ba5ca842bf..23fd9bc24f 100644
--- a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgraderTest.java
+++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgraderTest.java
@@ -24,7 +24,7 @@ import java.io.File;
 import java.util.ArrayList;
 import java.util.List;
 
-import org.apache.qpid.server.store.berkeleydb.BDBMessageStore;
+import org.apache.qpid.server.store.berkeleydb.AbstractBDBMessageStore;
 import org.apache.qpid.server.store.berkeleydb.tuple.ContentBinding;
 
 import com.sleepycat.bind.tuple.IntegerBinding;
@@ -94,7 +94,7 @@ public class UpgraderTest extends AbstractUpgradeTestCase
     {
         assertEquals("Unexpected store version", -1, getStoreVersion());
         _upgrader.upgradeIfNecessary();
-        assertEquals("Unexpected store version", BDBMessageStore.VERSION, getStoreVersion());
+        assertEquals("Unexpected store version", AbstractBDBMessageStore.VERSION, getStoreVersion());
         assertContent();
     }
 
@@ -112,7 +112,7 @@ public class UpgraderTest extends AbstractUpgradeTestCase
         List<String> expectedDatabases = new ArrayList<String>();
         expectedDatabases.add(Upgrader.VERSION_DB_NAME);
         assertEquals("Expectedonly VERSION table in initially empty store after upgrade: ", expectedDatabases, databaseNames);
-        assertEquals("Unexpected store version", BDBMessageStore.VERSION, getStoreVersion());
+        assertEquals("Unexpected store version", AbstractBDBMessageStore.VERSION, getStoreVersion());
 
         nonExistentStoreLocation.delete();
     }