QPID-5800: [Java Broker} Refactor MessageStore implementations extracting a MessageStoreProvider interface.

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

12 files changed, 1838 insertions, 1802 deletions

diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBConfigurationStore.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBConfigurationStore.java
new file mode 100644
index 0000000000..2766c2ac98
--- /dev/null
+++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBConfigurationStore.java
@@ -0,0 +1,1802 @@
+/*
+ *
+ * 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.lang.ref.SoftReference;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+import com.sleepycat.bind.tuple.ByteBinding;
+import com.sleepycat.bind.tuple.LongBinding;
+import com.sleepycat.je.CheckpointConfig;
+import com.sleepycat.je.Cursor;
+import com.sleepycat.je.Database;
+import com.sleepycat.je.DatabaseConfig;
+import com.sleepycat.je.DatabaseEntry;
+import com.sleepycat.je.DatabaseException;
+import com.sleepycat.je.EnvironmentConfig;
+import com.sleepycat.je.LockConflictException;
+import com.sleepycat.je.LockMode;
+import com.sleepycat.je.OperationStatus;
+import com.sleepycat.je.Transaction;
+import org.apache.log4j.Logger;
+
+import org.apache.qpid.server.message.EnqueueableMessage;
+import org.apache.qpid.server.model.ConfiguredObject;
+import org.apache.qpid.server.model.VirtualHostNode;
+import org.apache.qpid.server.store.ConfiguredObjectRecord;
+import org.apache.qpid.server.store.DurableConfigurationStore;
+import org.apache.qpid.server.store.Event;
+import org.apache.qpid.server.store.EventListener;
+import org.apache.qpid.server.store.EventManager;
+import org.apache.qpid.server.store.MessageStore;
+import org.apache.qpid.server.store.MessageStoreProvider;
+import org.apache.qpid.server.store.StorableMessageMetaData;
+import org.apache.qpid.server.store.StoreException;
+import org.apache.qpid.server.store.StoreFuture;
+import org.apache.qpid.server.store.StoredMemoryMessage;
+import org.apache.qpid.server.store.StoredMessage;
+import org.apache.qpid.server.store.TransactionLogResource;
+import org.apache.qpid.server.store.Xid;
+import org.apache.qpid.server.store.berkeleydb.EnvironmentFacadeFactory.EnvironmentFacadeTask;
+import org.apache.qpid.server.store.berkeleydb.entry.HierarchyKey;
+import org.apache.qpid.server.store.berkeleydb.entry.PreparedTransaction;
+import org.apache.qpid.server.store.berkeleydb.entry.QueueEntryKey;
+import org.apache.qpid.server.store.berkeleydb.tuple.ConfiguredObjectBinding;
+import org.apache.qpid.server.store.berkeleydb.tuple.ContentBinding;
+import org.apache.qpid.server.store.berkeleydb.tuple.HierarchyKeyBinding;
+import org.apache.qpid.server.store.berkeleydb.tuple.MessageMetaDataBinding;
+import org.apache.qpid.server.store.berkeleydb.tuple.PreparedTransactionBinding;
+import org.apache.qpid.server.store.berkeleydb.tuple.QueueEntryBinding;
+import org.apache.qpid.server.store.berkeleydb.tuple.UUIDTupleBinding;
+import org.apache.qpid.server.store.berkeleydb.tuple.XidBinding;
+import org.apache.qpid.server.store.berkeleydb.upgrade.Upgrader;
+import org.apache.qpid.server.store.handler.ConfiguredObjectRecordHandler;
+import org.apache.qpid.server.store.handler.DistributedTransactionHandler;
+import org.apache.qpid.server.store.handler.MessageHandler;
+import org.apache.qpid.server.store.handler.MessageInstanceHandler;
+import org.apache.qpid.server.util.MapValueConverter;
+import org.apache.qpid.util.FileUtils;
+
+/**
+ * BDBMessageStore implements a persistent {@link MessageStore} using the BDB high performance log.
+ *
+ * <p/><table id="crc"><caption>CRC Card</caption> <tr><th> Responsibilities <th> Collaborations <tr><td> Accept
+ * transaction boundary demarcations: Begin, Commit, Abort. <tr><td> Store and remove queues. <tr><td> Store and remove
+ * exchanges. <tr><td> Store and remove messages. <tr><td> Bind and unbind queues to exchanges. <tr><td> Enqueue and
+ * dequeue messages to queues. <tr><td> Generate message identifiers. </table>
+ */
+public class BDBConfigurationStore implements MessageStoreProvider, DurableConfigurationStore
+{
+    private static final Logger LOGGER = Logger.getLogger(BDBConfigurationStore.class);
+
+    public static final int VERSION = 8;
+    private static String CONFIGURED_OBJECTS_DB_NAME = "CONFIGURED_OBJECTS";
+    private static String CONFIGURED_OBJECT_HIERARCHY_DB_NAME = "CONFIGURED_OBJECT_HIERARCHY";
+
+    private static String MESSAGE_META_DATA_DB_NAME = "MESSAGE_METADATA";
+    private static String MESSAGE_CONTENT_DB_NAME = "MESSAGE_CONTENT";
+    private static String DELIVERY_DB_NAME = "QUEUE_ENTRIES";
+
+    //TODO: Add upgrader to remove BRIDGES and LINKS
+    private static String BRIDGEDB_NAME = "BRIDGES";
+    private static String LINKDB_NAME = "LINKS";
+    private static String XID_DB_NAME = "XIDS";
+    private static final String[] CONFIGURATION_STORE_DATABASE_NAMES = new String[] { CONFIGURED_OBJECTS_DB_NAME, CONFIGURED_OBJECT_HIERARCHY_DB_NAME};
+    private static final String[] MESSAGE_STORE_DATABASE_NAMES = new String[] { MESSAGE_META_DATA_DB_NAME, MESSAGE_CONTENT_DB_NAME, DELIVERY_DB_NAME, BRIDGEDB_NAME, LINKDB_NAME, XID_DB_NAME };
+    private EnvironmentFacade _environmentFacade;
+    private final AtomicLong _messageId = new AtomicLong(0);
+
+    private final AtomicBoolean _messageStoreOpen = new AtomicBoolean();
+    private final AtomicBoolean _configurationStoreOpen = new AtomicBoolean();
+
+    private long _totalStoreSize;
+
+    private final EnvironmentFacadeFactory _environmentFacadeFactory;
+
+    private volatile Committer _committer;
+
+    private boolean _isMessageStoreProvider;
+
+    private String _storeLocation;
+    private final BDBMessageStore _messageStoreFacade = new BDBMessageStore();
+
+    public BDBConfigurationStore()
+    {
+        this(new StandardEnvironmentFacadeFactory());
+    }
+
+    public BDBConfigurationStore(EnvironmentFacadeFactory environmentFacadeFactory)
+    {
+        _environmentFacadeFactory = environmentFacadeFactory;
+    }
+
+    @Override
+    public void openConfigurationStore(ConfiguredObject<?> parent, Map<String, Object> storeSettings)
+    {
+        if (_configurationStoreOpen.compareAndSet(false,  true))
+        {
+            if (_environmentFacade == null)
+            {
+                EnvironmentFacadeTask[] initialisationTasks = null;
+                _isMessageStoreProvider = MapValueConverter.getBooleanAttribute(VirtualHostNode.IS_MESSAGE_STORE_PROVIDER, storeSettings, false);
+                if (_isMessageStoreProvider)
+                {
+                    String[] databaseNames = new String[CONFIGURATION_STORE_DATABASE_NAMES.length + MESSAGE_STORE_DATABASE_NAMES.length];
+                    System.arraycopy(CONFIGURATION_STORE_DATABASE_NAMES, 0, databaseNames, 0, CONFIGURATION_STORE_DATABASE_NAMES.length);
+                    System.arraycopy(MESSAGE_STORE_DATABASE_NAMES, 0, databaseNames, CONFIGURATION_STORE_DATABASE_NAMES.length, MESSAGE_STORE_DATABASE_NAMES.length);
+                    initialisationTasks = new EnvironmentFacadeTask[]{new UpgradeTask(parent), new OpenDatabasesTask(databaseNames), new DiskSpaceTask(), new MaxMessageIdTask()};
+                }
+                else
+                {
+                    initialisationTasks = new EnvironmentFacadeTask[]{new UpgradeTask(parent), new OpenDatabasesTask(CONFIGURATION_STORE_DATABASE_NAMES)};
+                }
+                _environmentFacade = _environmentFacadeFactory.createEnvironmentFacade(storeSettings, initialisationTasks);
+                _storeLocation = _environmentFacade.getStoreLocation();
+            }
+            else
+            {
+                throw new IllegalStateException("The database have been already opened as message store");
+            }
+        }
+    }
+
+    @Override
+    public void visitConfiguredObjectRecords(ConfiguredObjectRecordHandler handler)
+    {
+        checkConfigurationStoreOpen();
+
+        try
+        {
+            handler.begin();
+            doVisitAllConfiguredObjectRecords(handler);
+            handler.end();
+        }
+        catch (DatabaseException e)
+        {
+            throw _environmentFacade.handleDatabaseException("Cannot visit configured object records", e);
+        }
+
+    }
+
+    private void doVisitAllConfiguredObjectRecords(ConfiguredObjectRecordHandler handler)
+    {
+        Map<UUID, BDBConfiguredObjectRecord> configuredObjects = new HashMap<UUID, BDBConfiguredObjectRecord>();
+        Cursor objectsCursor = null;
+        Cursor hierarchyCursor = null;
+        try
+        {
+            objectsCursor = getConfiguredObjectsDb().openCursor(null, null);
+            DatabaseEntry key = new DatabaseEntry();
+            DatabaseEntry value = new DatabaseEntry();
+
+
+            while (objectsCursor.getNext(key, value, LockMode.RMW) == OperationStatus.SUCCESS)
+            {
+                UUID id = UUIDTupleBinding.getInstance().entryToObject(key);
+
+                BDBConfiguredObjectRecord configuredObject =
+                        (BDBConfiguredObjectRecord) new ConfiguredObjectBinding(id).entryToObject(value);
+                configuredObjects.put(configuredObject.getId(), configuredObject);
+            }
+
+            // set parents
+            hierarchyCursor = getConfiguredObjectHierarchyDb().openCursor(null, null);
+            while (hierarchyCursor.getNext(key, value, LockMode.RMW) == OperationStatus.SUCCESS)
+            {
+                HierarchyKey hk = HierarchyKeyBinding.getInstance().entryToObject(key);
+                UUID parentId = UUIDTupleBinding.getInstance().entryToObject(value);
+                BDBConfiguredObjectRecord child = configuredObjects.get(hk.getChildId());
+                if(child != null)
+                {
+                    ConfiguredObjectRecord parent = configuredObjects.get(parentId);
+                    if(parent != null)
+                    {
+                        child.addParent(hk.getParentType(), parent);
+                    }
+                }
+            }
+        }
+        finally
+        {
+            closeCursorSafely(objectsCursor);
+            closeCursorSafely(hierarchyCursor);
+        }
+
+        for (ConfiguredObjectRecord record : configuredObjects.values())
+        {
+            boolean shoudlContinue = handler.handle(record);
+            if (!shoudlContinue)
+            {
+                break;
+            }
+        }
+
+    }
+
+    public String getStoreLocation()
+    {
+        return _storeLocation;
+    }
+
+    public EnvironmentFacade getEnvironmentFacade()
+    {
+        return _environmentFacade;
+    }
+
+    @Override
+    public void closeConfigurationStore() throws StoreException
+    {
+        if (_configurationStoreOpen.compareAndSet(true, false))
+        {
+            if (!_messageStoreOpen.get())
+            {
+                closeEnvironment();
+            }
+        }
+    }
+
+    private void closeEnvironment()
+    {
+        if (_environmentFacade != null)
+        {
+            try
+            {
+                _environmentFacade.close();
+                _environmentFacade = null;
+            }
+            catch(DatabaseException e)
+            {
+                throw new StoreException("Exception occurred on message store close", e);
+            }
+        }
+    }
+
+    private void closeCursorSafely(Cursor cursor) throws StoreException
+    {
+        if (cursor != null)
+        {
+            try
+            {
+                cursor.close();
+            }
+            catch(DatabaseException e)
+            {
+                throw _environmentFacade.handleDatabaseException("Cannot close cursor", e);
+            }
+        }
+    }
+
+    private void abortTransactionIgnoringException(String errorMessage, com.sleepycat.je.Transaction tx)
+    {
+        try
+        {
+            if (tx != null)
+            {
+                tx.abort();
+            }
+        }
+        catch (DatabaseException e1)
+        {
+            // We need the possible side effect of the handler restarting the environment but don't care about the exception
+            _environmentFacade.handleDatabaseException(null, e1);
+            LOGGER.warn(errorMessage, e1);
+        }
+    }
+
+    @Override
+    public void create(ConfiguredObjectRecord configuredObject) throws StoreException
+    {
+        checkConfigurationStoreOpen();
+
+        if (LOGGER.isDebugEnabled())
+        {
+            LOGGER.debug("Create " + configuredObject);
+        }
+
+        com.sleepycat.je.Transaction txn = null;
+        try
+        {
+            txn = _environmentFacade.getEnvironment().beginTransaction(null, null);
+            storeConfiguredObjectEntry(txn, configuredObject);
+            txn.commit();
+            txn = null;
+        }
+        catch (DatabaseException e)
+        {
+            throw _environmentFacade.handleDatabaseException("Error creating configured object " + configuredObject
+                    + " in database: " + e.getMessage(), e);
+        }
+        finally
+        {
+            if (txn != null)
+            {
+                abortTransactionIgnoringException("Error creating configured object", txn);
+            }
+        }
+    }
+
+    @Override
+    public UUID[] remove(final ConfiguredObjectRecord... objects) throws StoreException
+    {
+        checkConfigurationStoreOpen();
+
+        com.sleepycat.je.Transaction txn = null;
+        try
+        {
+            txn = _environmentFacade.getEnvironment().beginTransaction(null, null);
+
+            Collection<UUID> removed = new ArrayList<UUID>(objects.length);
+            for(ConfiguredObjectRecord record : objects)
+            {
+                if(removeConfiguredObject(txn, record) == OperationStatus.SUCCESS)
+                {
+                    removed.add(record.getId());
+                }
+            }
+
+            txn.commit();
+            txn = null;
+            return removed.toArray(new UUID[removed.size()]);
+        }
+        catch (DatabaseException e)
+        {
+            throw _environmentFacade.handleDatabaseException("Error deleting configured objects from database", e);
+        }
+        finally
+        {
+            if (txn != null)
+            {
+                abortTransactionIgnoringException("Error deleting configured objects", txn);
+            }
+        }
+    }
+
+    @Override
+    public void update(boolean createIfNecessary, ConfiguredObjectRecord... records) throws StoreException
+    {
+        checkConfigurationStoreOpen();
+
+        com.sleepycat.je.Transaction txn = null;
+        try
+        {
+            txn = _environmentFacade.getEnvironment().beginTransaction(null, null);
+            for(ConfiguredObjectRecord record : records)
+            {
+                update(createIfNecessary, record, txn);
+            }
+            txn.commit();
+            txn = null;
+        }
+        catch (DatabaseException e)
+        {
+            throw _environmentFacade.handleDatabaseException("Error updating configuration details within the store: " + e,e);
+        }
+        finally
+        {
+            if (txn != null)
+            {
+                abortTransactionIgnoringException("Error updating configuration details within the store", txn);
+            }
+        }
+    }
+
+    private void update(boolean createIfNecessary, ConfiguredObjectRecord record, com.sleepycat.je.Transaction txn) throws StoreException
+    {
+        if (LOGGER.isDebugEnabled())
+        {
+            LOGGER.debug("Updating, creating " + createIfNecessary + " : "  + record);
+        }
+
+        DatabaseEntry key = new DatabaseEntry();
+        UUIDTupleBinding keyBinding = UUIDTupleBinding.getInstance();
+        keyBinding.objectToEntry(record.getId(), key);
+
+        DatabaseEntry value = new DatabaseEntry();
+        DatabaseEntry newValue = new DatabaseEntry();
+        ConfiguredObjectBinding configuredObjectBinding = ConfiguredObjectBinding.getInstance();
+
+        OperationStatus status = getConfiguredObjectsDb().get(txn, key, value, LockMode.DEFAULT);
+        final boolean isNewRecord = status == OperationStatus.NOTFOUND;
+        if (status == OperationStatus.SUCCESS || (createIfNecessary && isNewRecord))
+        {
+            // write the updated entry to the store
+            configuredObjectBinding.objectToEntry(record, newValue);
+            status = getConfiguredObjectsDb().put(txn, key, newValue);
+            if (status != OperationStatus.SUCCESS)
+            {
+                throw new StoreException("Error updating configuration details within the store: " + status);
+            }
+            if(isNewRecord)
+            {
+                writeHierarchyRecords(txn, record);
+            }
+        }
+        else if (status != OperationStatus.NOTFOUND)
+        {
+            throw new StoreException("Error finding configuration details within the store: " + status);
+        }
+    }
+
+
+    private void storeConfiguredObjectEntry(final Transaction txn, ConfiguredObjectRecord configuredObject) throws StoreException
+    {
+        if (LOGGER.isDebugEnabled())
+        {
+            LOGGER.debug("Storing configured object record: " + configuredObject);
+        }
+        DatabaseEntry key = new DatabaseEntry();
+        UUIDTupleBinding uuidBinding = UUIDTupleBinding.getInstance();
+        uuidBinding.objectToEntry(configuredObject.getId(), key);
+        DatabaseEntry value = new DatabaseEntry();
+        ConfiguredObjectBinding queueBinding = ConfiguredObjectBinding.getInstance();
+
+        queueBinding.objectToEntry(configuredObject, value);
+        try
+        {
+            OperationStatus status = getConfiguredObjectsDb().put(txn, key, value);
+            if (status != OperationStatus.SUCCESS)
+            {
+                throw new StoreException("Error writing configured object " + configuredObject + " to database: "
+                        + status);
+            }
+            writeHierarchyRecords(txn, configuredObject);
+        }
+        catch (DatabaseException e)
+        {
+            throw _environmentFacade.handleDatabaseException("Error writing configured object " + configuredObject
+                    + " to database: " + e.getMessage(), e);
+        }
+    }
+
+    private void writeHierarchyRecords(final Transaction txn, final ConfiguredObjectRecord configuredObject)
+    {
+        OperationStatus status;
+        HierarchyKeyBinding hierarchyBinding = HierarchyKeyBinding.getInstance();
+        DatabaseEntry hierarchyKey = new DatabaseEntry();
+        DatabaseEntry hierarchyValue = new DatabaseEntry();
+
+        for(Map.Entry<String, ConfiguredObjectRecord> parent : configuredObject.getParents().entrySet())
+        {
+
+            hierarchyBinding.objectToEntry(new HierarchyKey(configuredObject.getId(), parent.getKey()), hierarchyKey);
+            UUIDTupleBinding.getInstance().objectToEntry(parent.getValue().getId(), hierarchyValue);
+            status = getConfiguredObjectHierarchyDb().put(txn, hierarchyKey, hierarchyValue);
+            if (status != OperationStatus.SUCCESS)
+            {
+                throw new StoreException("Error writing configured object " + configuredObject + " parent record to database: "
+                                         + status);
+            }
+        }
+    }
+
+    private OperationStatus removeConfiguredObject(Transaction tx, ConfiguredObjectRecord record) throws StoreException
+    {
+        UUID id = record.getId();
+        Map<String, ConfiguredObjectRecord> parents = record.getParents();
+
+        if (LOGGER.isDebugEnabled())
+        {
+            LOGGER.debug("Removing configured object: " + id);
+        }
+        DatabaseEntry key = new DatabaseEntry();
+
+        UUIDTupleBinding uuidBinding = UUIDTupleBinding.getInstance();
+        uuidBinding.objectToEntry(id, key);
+        OperationStatus status = getConfiguredObjectsDb().delete(tx, key);
+        if(status == OperationStatus.SUCCESS)
+        {
+            for(String parentType : parents.keySet())
+            {
+                DatabaseEntry hierarchyKey = new DatabaseEntry();
+                HierarchyKeyBinding keyBinding = HierarchyKeyBinding.getInstance();
+                keyBinding.objectToEntry(new HierarchyKey(record.getId(), parentType), hierarchyKey);
+                getConfiguredObjectHierarchyDb().delete(tx, hierarchyKey);
+            }
+        }
+        return status;
+    }
+
+    @Override
+    public MessageStore getMessageStore()
+    {
+        return _messageStoreFacade;
+    }
+
+    private void checkConfigurationStoreOpen()
+    {
+        if (!_configurationStoreOpen.get())
+        {
+            throw new IllegalStateException("Configuration store is not open");
+        }
+    }
+
+    public void onDelete()
+    {
+        if (!_configurationStoreOpen.get() && !_messageStoreOpen.get())
+        {
+            if (_storeLocation != null)
+            {
+                if (LOGGER.isDebugEnabled())
+                {
+                    LOGGER.debug("Deleting store " + _storeLocation);
+                }
+
+                File location = new File(_storeLocation);
+                if (location.exists())
+                {
+                    if (!FileUtils.delete(location, true))
+                    {
+                        LOGGER.error("Cannot delete " + _storeLocation);
+                    }
+                }
+            }
+        }
+    }
+
+    private Database getConfiguredObjectsDb()
+    {
+        return _environmentFacade.getOpenDatabase(CONFIGURED_OBJECTS_DB_NAME);
+    }
+
+    private Database getConfiguredObjectHierarchyDb()
+    {
+        return _environmentFacade.getOpenDatabase(CONFIGURED_OBJECT_HIERARCHY_DB_NAME);
+    }
+
+    private Database getMessageContentDb()
+    {
+        return _environmentFacade.getOpenDatabase(MESSAGE_CONTENT_DB_NAME);
+    }
+
+    private Database getMessageMetaDataDb()
+    {
+        return _environmentFacade.getOpenDatabase(MESSAGE_META_DATA_DB_NAME);
+    }
+
+    private Database getDeliveryDb()
+    {
+        return _environmentFacade.getOpenDatabase(DELIVERY_DB_NAME);
+    }
+
+    private Database getXidDb()
+    {
+        return _environmentFacade.getOpenDatabase(XID_DB_NAME);
+    }
+
+    class UpgradeTask implements EnvironmentFacadeTask
+    {
+        private final ConfiguredObject<?> _parent;
+
+        public UpgradeTask(ConfiguredObject<?> parent)
+        {
+            _parent = parent;
+        }
+
+        @Override
+        public void execute(EnvironmentFacade facade)
+        {
+            try
+            {
+                new Upgrader(facade.getEnvironment(), _parent).upgradeIfNecessary();
+            }
+            catch(DatabaseException e)
+            {
+                throw facade.handleDatabaseException("Cannot upgrade store", e);
+            }
+        }
+    }
+
+    class OpenDatabasesTask implements EnvironmentFacadeTask
+    {
+        private String[] _names;
+
+        public OpenDatabasesTask(String[] names)
+        {
+            _names = names;
+        }
+
+        @Override
+        public void execute(EnvironmentFacade facade)
+        {
+            try
+            {
+                DatabaseConfig dbConfig = new DatabaseConfig();
+                dbConfig.setTransactional(true);
+                dbConfig.setAllowCreate(true);
+                facade.openDatabases(dbConfig, _names);
+            }
+            catch(DatabaseException e)
+            {
+                throw facade.handleDatabaseException("Cannot open databases", e);
+            }
+        }
+
+    }
+
+    class DiskSpaceTask implements EnvironmentFacadeTask
+    {
+
+        @Override
+        public void execute(EnvironmentFacade facade)
+        {
+            try
+            {
+                _totalStoreSize = facade.getEnvironment().getStats(null).getTotalLogSize();
+            }
+            catch(DatabaseException e)
+            {
+                throw facade.handleDatabaseException("Cannot evaluate disk store size", e);
+            }
+        }
+
+    }
+
+    public class MaxMessageIdTask implements EnvironmentFacadeTask, MessageHandler
+    {
+        private long _maxId;
+
+        @Override
+        public void execute(EnvironmentFacade facade)
+        {
+            ((BDBMessageStore)getMessageStore()).visitMessagesInternal(this, facade);
+            _messageId.set(_maxId);
+        }
+
+        @Override
+        public boolean handle(StoredMessage<?> storedMessage)
+        {
+            long id = storedMessage.getMessageNumber();
+            if (_maxId<id)
+            {
+                _maxId = id;
+            }
+            return true;
+        }
+    }
+
+    class BDBMessageStore implements MessageStore
+    {
+        private static final int LOCK_RETRY_ATTEMPTS = 5;
+
+        private final EventManager _eventManager = new EventManager();
+
+        private boolean _limitBusted;
+        private long _persistentSizeLowThreshold;
+        private long _persistentSizeHighThreshold;
+
+        @Override
+        public void openMessageStore(final ConfiguredObject<?> parent, final Map<String, Object> messageStoreSettings)
+        {
+            if (_messageStoreOpen.compareAndSet(false, true))
+            {
+
+                Object overfullAttr = messageStoreSettings.get(OVERFULL_SIZE);
+                Object underfullAttr = messageStoreSettings.get(UNDERFULL_SIZE);
+
+                _persistentSizeHighThreshold = overfullAttr == null ? -1l :
+                                               overfullAttr instanceof Number ? ((Number) overfullAttr).longValue() : Long.parseLong(overfullAttr.toString());
+                _persistentSizeLowThreshold = underfullAttr == null ? _persistentSizeHighThreshold :
+                                               underfullAttr instanceof Number ? ((Number) underfullAttr).longValue() : Long.parseLong(underfullAttr.toString());
+
+                if(_persistentSizeLowThreshold > _persistentSizeHighThreshold || _persistentSizeLowThreshold < 0l)
+                {
+                    _persistentSizeLowThreshold = _persistentSizeHighThreshold;
+                }
+
+                if (_environmentFacade == null)
+                {
+                    _environmentFacade = _environmentFacadeFactory.createEnvironmentFacade(messageStoreSettings,
+                            new UpgradeTask(parent), new OpenDatabasesTask(MESSAGE_STORE_DATABASE_NAMES), new DiskSpaceTask(), new MaxMessageIdTask());
+                    _storeLocation = _environmentFacade.getStoreLocation();
+                }
+
+                _committer = _environmentFacade.createCommitter(parent.getName());
+                _committer.start();
+            }
+        }
+
+        @Override
+        public <T extends StorableMessageMetaData> StoredMessage<T> addMessage(T metaData)
+        {
+            if (metaData.isPersistent())
+            {
+                return (StoredMessage<T>) new StoredBDBMessage(getNewMessageId(), metaData);
+            }
+            else
+            {
+                return new StoredMemoryMessage<T>(getNewMessageId(), metaData);
+            }
+        }
+
+        /**
+         * Return a valid, currently unused message id.
+         *
+         * @return A fresh message id.
+         */
+        private long getNewMessageId()
+        {
+            return _messageId.incrementAndGet();
+        }
+
+        @Override
+        public boolean isPersistent()
+        {
+            return true;
+        }
+
+        @Override
+        public org.apache.qpid.server.store.Transaction newTransaction()
+        {
+            checkMessageStoreOpen();
+
+            return new BDBTransaction();
+        }
+
+        @Override
+        public void closeMessageStore()
+        {
+            if (_messageStoreOpen.compareAndSet(true, false))
+            {
+                try
+                {
+                    if (_committer != null)
+                    {
+                        _committer.stop();
+                    }
+                }
+                finally
+                {
+                    if (!_configurationStoreOpen.get())
+                    {
+                        closeEnvironment();
+                    }
+                }
+            }
+        }
+
+        @Override
+        public void addEventListener(final EventListener eventListener, final Event... events)
+        {
+            _eventManager.addEventListener(eventListener, events);
+        }
+
+        @Override
+        public void onDelete()
+        {
+            BDBConfigurationStore.this.onDelete();
+        }
+
+        @Override
+        public String getStoreLocation()
+        {
+            return BDBConfigurationStore.this.getStoreLocation();
+        }
+
+        @Override
+        public void visitMessages(final MessageHandler handler) throws StoreException
+        {
+            checkMessageStoreOpen();
+            visitMessagesInternal(handler, _environmentFacade);
+        }
+
+        @Override
+        public void visitMessageInstances(final MessageInstanceHandler handler) throws StoreException
+        {
+            checkMessageStoreOpen();
+
+            Cursor cursor = null;
+            List<QueueEntryKey> entries = new ArrayList<QueueEntryKey>();
+            try
+            {
+                cursor = getDeliveryDb().openCursor(null, null);
+                DatabaseEntry key = new DatabaseEntry();
+                QueueEntryBinding keyBinding = QueueEntryBinding.getInstance();
+
+                DatabaseEntry value = new DatabaseEntry();
+                while (cursor.getNext(key, value, LockMode.DEFAULT) == OperationStatus.SUCCESS)
+                {
+                    QueueEntryKey entry = keyBinding.entryToObject(key);
+                    entries.add(entry);
+                }
+            }
+            catch (DatabaseException e)
+            {
+                throw _environmentFacade.handleDatabaseException("Cannot visit message instances", e);
+            }
+            finally
+            {
+                closeCursorSafely(cursor);
+            }
+
+            for(QueueEntryKey entry : entries)
+            {
+                UUID queueId = entry.getQueueId();
+                long messageId = entry.getMessageId();
+                if (!handler.handle(queueId, messageId))
+                {
+                    break;
+                }
+            }
+
+        }
+
+        @Override
+        public void visitDistributedTransactions(final DistributedTransactionHandler handler) throws StoreException
+        {
+            checkMessageStoreOpen();
+
+            Cursor cursor = null;
+            try
+            {
+                cursor = getXidDb().openCursor(null, null);
+                DatabaseEntry key = new DatabaseEntry();
+                XidBinding keyBinding = XidBinding.getInstance();
+                PreparedTransactionBinding valueBinding = new PreparedTransactionBinding();
+                DatabaseEntry value = new DatabaseEntry();
+
+                while (cursor.getNext(key, value, LockMode.RMW) == OperationStatus.SUCCESS)
+                {
+                    Xid xid = keyBinding.entryToObject(key);
+                    PreparedTransaction preparedTransaction = valueBinding.entryToObject(value);
+                    if (!handler.handle(xid.getFormat(), xid.getGlobalId(), xid.getBranchId(),
+                                        preparedTransaction.getEnqueues(), preparedTransaction.getDequeues()))
+                    {
+                        break;
+                    }
+                }
+
+            }
+            catch (DatabaseException e)
+            {
+                throw _environmentFacade.handleDatabaseException("Cannot recover distributed transactions", e);
+            }
+            finally
+            {
+                closeCursorSafely(cursor);
+            }
+        }
+
+        /**
+         * Retrieves message meta-data.
+         *
+         * @param messageId The message to get the meta-data for.
+         *
+         * @return The message meta data.
+         *
+         * @throws StoreException If the operation fails for any reason, or if the specified message does not exist.
+         */
+        StorableMessageMetaData getMessageMetaData(long messageId) throws StoreException
+        {
+            if (LOGGER.isDebugEnabled())
+            {
+                LOGGER.debug("public MessageMetaData getMessageMetaData(Long messageId = "
+                             + messageId + "): called");
+            }
+
+            DatabaseEntry key = new DatabaseEntry();
+            LongBinding.longToEntry(messageId, key);
+            DatabaseEntry value = new DatabaseEntry();
+            MessageMetaDataBinding messageBinding = MessageMetaDataBinding.getInstance();
+
+            try
+            {
+                OperationStatus status = getMessageMetaDataDb().get(null, key, value, LockMode.READ_UNCOMMITTED);
+                if (status != OperationStatus.SUCCESS)
+                {
+                    throw new StoreException("Metadata not found for message with id " + messageId);
+                }
+
+                StorableMessageMetaData mdd = messageBinding.entryToObject(value);
+
+                return mdd;
+            }
+            catch (DatabaseException e)
+            {
+                throw _environmentFacade.handleDatabaseException("Error reading message metadata for message with id " + messageId + ": " + e.getMessage(), e);
+            }
+        }
+
+        void removeMessage(long messageId, boolean sync) throws StoreException
+        {
+            boolean complete = false;
+            com.sleepycat.je.Transaction tx = null;
+
+            Random rand = null;
+            int attempts = 0;
+            try
+            {
+                do
+                {
+                    tx = null;
+                    try
+                    {
+                        tx = _environmentFacade.getEnvironment().beginTransaction(null, null);
+
+                        //remove the message meta data from the store
+                        DatabaseEntry key = new DatabaseEntry();
+                        LongBinding.longToEntry(messageId, key);
+
+                        if (LOGGER.isDebugEnabled())
+                        {
+                            LOGGER.debug("Removing message id " + messageId);
+                        }
+
+
+                        OperationStatus status = getMessageMetaDataDb().delete(tx, key);
+                        if (status == OperationStatus.NOTFOUND)
+                        {
+                            LOGGER.info("Message not found (attempt to remove failed - probably application initiated rollback) " +
+                                        messageId);
+                        }
+
+                        if (LOGGER.isDebugEnabled())
+                        {
+                            LOGGER.debug("Deleted metadata for message " + messageId);
+                        }
+
+                        //now remove the content data from the store if there is any.
+                        DatabaseEntry contentKeyEntry = new DatabaseEntry();
+                        LongBinding.longToEntry(messageId, contentKeyEntry);
+                        getMessageContentDb().delete(tx, contentKeyEntry);
+
+                        if (LOGGER.isDebugEnabled())
+                        {
+                            LOGGER.debug("Deleted content for message " + messageId);
+                        }
+
+                        _environmentFacade.commit(tx);
+                        _committer.commit(tx, sync);
+
+                        complete = true;
+                        tx = null;
+                    }
+                    catch (LockConflictException e)
+                    {
+                        try
+                        {
+                            if(tx != null)
+                            {
+                                tx.abort();
+                            }
+                        }
+                        catch(DatabaseException e2)
+                        {
+                            LOGGER.warn("Unable to abort transaction after LockConflictExcption on removal of message with id " + messageId, e2);
+                            // rethrow the original log conflict exception, the secondary exception should already have
+                            // been logged.
+                            throw _environmentFacade.handleDatabaseException("Cannot remove message with id " + messageId, e);
+                        }
+
+
+                        LOGGER.warn("Lock timeout exception. Retrying (attempt "
+                                    + (attempts+1) + " of "+ LOCK_RETRY_ATTEMPTS +") " + e);
+
+                        if(++attempts < LOCK_RETRY_ATTEMPTS)
+                        {
+                            if(rand == null)
+                            {
+                                rand = new Random();
+                            }
+
+                            try
+                            {
+                                Thread.sleep(500l + (long)(500l * rand.nextDouble()));
+                            }
+                            catch (InterruptedException e1)
+                            {
+
+                            }
+                        }
+                        else
+                        {
+                            // rethrow the lock conflict exception since we could not solve by retrying
+                            throw _environmentFacade.handleDatabaseException("Cannot remove messages", e);
+                        }
+                    }
+                }
+                while(!complete);
+            }
+            catch (DatabaseException e)
+            {
+                LOGGER.error("Unexpected BDB exception", e);
+
+                try
+                {
+                    abortTransactionIgnoringException("Error aborting transaction on removal of message with id " + messageId, tx);
+                }
+                finally
+                {
+                    tx = null;
+                }
+
+                throw _environmentFacade.handleDatabaseException("Error removing message with id " + messageId + " from database: " + e.getMessage(), e);
+            }
+            finally
+            {
+                try
+                {
+                    abortTransactionIgnoringException("Error aborting transaction on removal of message with id " + messageId, tx);
+                }
+                finally
+                {
+                    tx = null;
+                }
+            }
+        }
+
+
+        /**
+         * Fills the provided ByteBuffer with as much content for the specified message as possible, starting
+         * from the specified offset in the message.
+         *
+         * @param messageId The message to get the data for.
+         * @param offset    The offset of the data within the message.
+         * @param dst       The destination of the content read back
+         *
+         * @return The number of bytes inserted into the destination
+         *
+         * @throws StoreException If the operation fails for any reason, or if the specified message does not exist.
+         */
+        int getContent(long messageId, int offset, ByteBuffer dst) throws StoreException
+        {
+            DatabaseEntry contentKeyEntry = new DatabaseEntry();
+            LongBinding.longToEntry(messageId, contentKeyEntry);
+            DatabaseEntry value = new DatabaseEntry();
+            ContentBinding contentTupleBinding = ContentBinding.getInstance();
+
+
+            if (LOGGER.isDebugEnabled())
+            {
+                LOGGER.debug("Message Id: " + messageId + " Getting content body from offset: " + offset);
+            }
+
+            try
+            {
+
+                int written = 0;
+                OperationStatus status = getMessageContentDb().get(null, contentKeyEntry, value, LockMode.READ_UNCOMMITTED);
+                if (status == OperationStatus.SUCCESS)
+                {
+                    byte[] dataAsBytes = contentTupleBinding.entryToObject(value);
+                    int size = dataAsBytes.length;
+                    if (offset > size)
+                    {
+                        throw new RuntimeException("Offset " + offset + " is greater than message size " + size
+                                                   + " for message id " + messageId + "!");
+
+                    }
+
+                    written = size - offset;
+                    if(written > dst.remaining())
+                    {
+                        written = dst.remaining();
+                    }
+
+                    dst.put(dataAsBytes, offset, written);
+                }
+                return written;
+            }
+            catch (DatabaseException e)
+            {
+                throw _environmentFacade.handleDatabaseException("Error getting AMQMessage with id " + messageId + " to database: " + e.getMessage(), e);
+            }
+        }
+
+        private void visitMessagesInternal(MessageHandler handler, EnvironmentFacade environmentFacade)
+        {
+            Cursor cursor = null;
+            try
+            {
+                cursor = environmentFacade.getOpenDatabase(MESSAGE_META_DATA_DB_NAME).openCursor(null, null);
+                DatabaseEntry key = new DatabaseEntry();
+                DatabaseEntry value = new DatabaseEntry();
+                MessageMetaDataBinding valueBinding = MessageMetaDataBinding.getInstance();
+
+                while (cursor.getNext(key, value, LockMode.RMW) == OperationStatus.SUCCESS)
+                {
+                    long messageId = LongBinding.entryToLong(key);
+                    StorableMessageMetaData metaData = valueBinding.entryToObject(value);
+                    StoredBDBMessage message = new StoredBDBMessage(messageId, metaData, true);
+
+                    if (!handler.handle(message))
+                    {
+                        break;
+                    }
+                }
+            }
+            catch (DatabaseException e)
+            {
+                throw environmentFacade.handleDatabaseException("Cannot visit messages", e);
+            }
+            finally
+            {
+                if (cursor != null)
+                {
+                    try
+                    {
+                        cursor.close();
+                    }
+                    catch(DatabaseException e)
+                    {
+                        throw environmentFacade.handleDatabaseException("Cannot close cursor", e);
+                    }
+                }
+            }
+        }
+
+        /**
+         * Stores a chunk of message data.
+         *
+         * @param tx         The transaction for the operation.
+         * @param messageId       The message to store the data for.
+         * @param offset          The offset of the data chunk in the message.
+         * @param contentBody     The content of the data chunk.
+         *
+         * @throws StoreException If the operation fails for any reason, or if the specified message does not exist.
+         */
+        private void addContent(final com.sleepycat.je.Transaction tx, long messageId, int offset,
+                                ByteBuffer contentBody) throws StoreException
+        {
+            DatabaseEntry key = new DatabaseEntry();
+            LongBinding.longToEntry(messageId, key);
+            DatabaseEntry value = new DatabaseEntry();
+            ContentBinding messageBinding = ContentBinding.getInstance();
+            messageBinding.objectToEntry(contentBody.array(), value);
+            try
+            {
+                OperationStatus status = getMessageContentDb().put(tx, key, value);
+                if (status != OperationStatus.SUCCESS)
+                {
+                    throw new StoreException("Error adding content for message id " + messageId + ": " + status);
+                }
+
+                if (LOGGER.isDebugEnabled())
+                {
+                    LOGGER.debug("Storing content for message " + messageId + " in transaction " + tx);
+
+                }
+            }
+            catch (DatabaseException e)
+            {
+                throw _environmentFacade.handleDatabaseException("Error writing AMQMessage with id " + messageId + " to database: " + e.getMessage(), e);
+            }
+        }
+
+        /**
+         * Stores message meta-data.
+         *
+         * @param tx         The transaction for the operation.
+         * @param messageId       The message to store the data for.
+         * @param messageMetaData The message meta data to store.
+         *
+         * @throws StoreException If the operation fails for any reason, or if the specified message does not exist.
+         */
+        private void storeMetaData(final com.sleepycat.je.Transaction tx, long messageId,
+                                   StorableMessageMetaData messageMetaData)
+                throws StoreException
+        {
+            if (LOGGER.isDebugEnabled())
+            {
+                LOGGER.debug("storeMetaData called for transaction " + tx
+                             + ", messageId " + messageId
+                             + ", messageMetaData " + messageMetaData);
+            }
+
+            DatabaseEntry key = new DatabaseEntry();
+            LongBinding.longToEntry(messageId, key);
+            DatabaseEntry value = new DatabaseEntry();
+
+            MessageMetaDataBinding messageBinding = MessageMetaDataBinding.getInstance();
+            messageBinding.objectToEntry(messageMetaData, value);
+            try
+            {
+                getMessageMetaDataDb().put(tx, key, value);
+                if (LOGGER.isDebugEnabled())
+                {
+                    LOGGER.debug("Storing message metadata for message id " + messageId + " in transaction " + tx);
+                }
+            }
+            catch (DatabaseException e)
+            {
+                throw _environmentFacade.handleDatabaseException("Error writing message metadata with id " + messageId + " to database: " + e.getMessage(), e);
+            }
+        }
+
+
+        /**
+         * Places a message onto a specified queue, in a given transaction.
+         *
+         * @param tx   The transaction for the operation.
+         * @param queue     The the queue to place the message on.
+         * @param messageId The message to enqueue.
+         *
+         * @throws StoreException If the operation fails for any reason.
+         */
+        private void enqueueMessage(final com.sleepycat.je.Transaction tx, final TransactionLogResource queue,
+                                    long messageId) throws StoreException
+        {
+
+            DatabaseEntry key = new DatabaseEntry();
+            QueueEntryBinding keyBinding = QueueEntryBinding.getInstance();
+            QueueEntryKey dd = new QueueEntryKey(queue.getId(), messageId);
+            keyBinding.objectToEntry(dd, key);
+            DatabaseEntry value = new DatabaseEntry();
+            ByteBinding.byteToEntry((byte) 0, value);
+
+            try
+            {
+                if (LOGGER.isDebugEnabled())
+                {
+                    LOGGER.debug("Enqueuing message " + messageId + " on queue "
+                                 + queue.getName() + " with id " + queue.getId() + " in transaction " + tx);
+                }
+                getDeliveryDb().put(tx, key, value);
+            }
+            catch (DatabaseException e)
+            {
+                LOGGER.error("Failed to enqueue: " + e.getMessage(), e);
+                throw _environmentFacade.handleDatabaseException("Error writing enqueued message with id " + messageId + " for queue "
+                                                                 + queue.getName() + " with id " + queue.getId() + " to database", e);
+            }
+        }
+
+        /**
+         * Extracts a message from a specified queue, in a given transaction.
+         *
+         * @param tx   The transaction for the operation.
+         * @param queue     The queue to take the message from.
+         * @param messageId The message to dequeue.
+         *
+         * @throws StoreException If the operation fails for any reason, or if the specified message does not exist.
+         */
+        private void dequeueMessage(final com.sleepycat.je.Transaction tx, final TransactionLogResource queue,
+                                    long messageId) throws StoreException
+        {
+
+            DatabaseEntry key = new DatabaseEntry();
+            QueueEntryBinding keyBinding = QueueEntryBinding.getInstance();
+            QueueEntryKey queueEntryKey = new QueueEntryKey(queue.getId(), messageId);
+            UUID id = queue.getId();
+            keyBinding.objectToEntry(queueEntryKey, key);
+            if (LOGGER.isDebugEnabled())
+            {
+                LOGGER.debug("Dequeue message id " + messageId + " from queue "
+                             + queue.getName() + " with id " + id);
+            }
+
+            try
+            {
+
+                OperationStatus status = getDeliveryDb().delete(tx, key);
+                if (status == OperationStatus.NOTFOUND)
+                {
+                    throw new StoreException("Unable to find message with id " + messageId + " on queue "
+                                             + queue.getName() + " with id "  + id);
+                }
+                else if (status != OperationStatus.SUCCESS)
+                {
+                    throw new StoreException("Unable to remove message with id " + messageId + " on queue"
+                                             + queue.getName() + " with id " + id);
+                }
+
+                if (LOGGER.isDebugEnabled())
+                {
+                    LOGGER.debug("Removed message " + messageId + " on queue "
+                                 + queue.getName() + " with id " + id);
+
+                }
+            }
+            catch (DatabaseException e)
+            {
+
+                LOGGER.error("Failed to dequeue message " + messageId + " in transaction " + tx , e);
+
+                throw _environmentFacade.handleDatabaseException("Error accessing database while dequeuing message: " + e.getMessage(), e);
+            }
+        }
+
+        private void recordXid(com.sleepycat.je.Transaction txn,
+                               long format,
+                               byte[] globalId,
+                               byte[] branchId,
+                               org.apache.qpid.server.store.Transaction.Record[] enqueues,
+                               org.apache.qpid.server.store.Transaction.Record[] dequeues) throws StoreException
+        {
+            DatabaseEntry key = new DatabaseEntry();
+            Xid xid = new Xid(format, globalId, branchId);
+            XidBinding keyBinding = XidBinding.getInstance();
+            keyBinding.objectToEntry(xid,key);
+
+            DatabaseEntry value = new DatabaseEntry();
+            PreparedTransaction preparedTransaction = new PreparedTransaction(enqueues, dequeues);
+            PreparedTransactionBinding valueBinding = new PreparedTransactionBinding();
+            valueBinding.objectToEntry(preparedTransaction, value);
+
+            try
+            {
+                getXidDb().put(txn, key, value);
+            }
+            catch (DatabaseException e)
+            {
+                LOGGER.error("Failed to write xid: " + e.getMessage(), e);
+                throw _environmentFacade.handleDatabaseException("Error writing xid to database", e);
+            }
+        }
+
+        private void removeXid(com.sleepycat.je.Transaction txn, long format, byte[] globalId, byte[] branchId)
+                throws StoreException
+        {
+            DatabaseEntry key = new DatabaseEntry();
+            Xid xid = new Xid(format, globalId, branchId);
+            XidBinding keyBinding = XidBinding.getInstance();
+
+            keyBinding.objectToEntry(xid, key);
+
+
+            try
+            {
+
+                OperationStatus status = getXidDb().delete(txn, key);
+                if (status == OperationStatus.NOTFOUND)
+                {
+                    throw new StoreException("Unable to find xid");
+                }
+                else if (status != OperationStatus.SUCCESS)
+                {
+                    throw new StoreException("Unable to remove xid");
+                }
+
+            }
+            catch (DatabaseException e)
+            {
+
+                LOGGER.error("Failed to remove xid in transaction " + txn, e);
+
+                throw _environmentFacade.handleDatabaseException("Error accessing database while removing xid: " + e.getMessage(), e);
+            }
+        }
+
+        /**
+         * Commits all operations performed within a given transaction.
+         *
+         * @param tx The transaction to commit all operations for.
+         *
+         * @throws StoreException If the operation fails for any reason.
+         */
+        private StoreFuture commitTranImpl(final com.sleepycat.je.Transaction tx, boolean syncCommit) throws StoreException
+        {
+            if (tx == null)
+            {
+                throw new StoreException("Fatal internal error: transactional is null at commitTran");
+            }
+
+            _environmentFacade.commit(tx);
+            StoreFuture result =  _committer.commit(tx, syncCommit);
+
+            if (LOGGER.isDebugEnabled())
+            {
+                String transactionType = syncCommit ? "synchronous" : "asynchronous";
+                LOGGER.debug("commitTranImpl completed " + transactionType + " transaction " + tx);
+            }
+
+            return result;
+        }
+
+        /**
+         * Abandons all operations performed within a given transaction.
+         *
+         * @param tx The transaction to abandon.
+         *
+         * @throws StoreException If the operation fails for any reason.
+         */
+        private void abortTran(final com.sleepycat.je.Transaction tx) throws StoreException
+        {
+            if (LOGGER.isDebugEnabled())
+            {
+                LOGGER.debug("abortTran called for transaction " + tx);
+            }
+
+            try
+            {
+                tx.abort();
+            }
+            catch (DatabaseException e)
+            {
+                throw _environmentFacade.handleDatabaseException("Error aborting transaction: " + e.getMessage(), e);
+            }
+        }
+
+        private void checkMessageStoreOpen()
+        {
+            if (!_messageStoreOpen.get())
+            {
+                throw new IllegalStateException("Message store is not open");
+            }
+        }
+
+        private void storedSizeChangeOccurred(final int delta) throws StoreException
+        {
+            try
+            {
+                storedSizeChange(delta);
+            }
+            catch(DatabaseException e)
+            {
+                throw _environmentFacade.handleDatabaseException("Stored size change exception", e);
+            }
+        }
+
+        private void storedSizeChange(final int delta)
+        {
+            if(getPersistentSizeHighThreshold() > 0)
+            {
+                synchronized (this)
+                {
+                    // the delta supplied is an approximation of a store size change. we don;t want to check the statistic every
+                    // time, so we do so only when there's been enough change that it is worth looking again. We do this by
+                    // assuming the total size will change by less than twice the amount of the message data change.
+                    long newSize = _totalStoreSize += 2*delta;
+
+                    if(!_limitBusted &&  newSize > getPersistentSizeHighThreshold())
+                    {
+                        _totalStoreSize = getSizeOnDisk();
+
+                        if(_totalStoreSize > getPersistentSizeHighThreshold())
+                        {
+                            _limitBusted = true;
+                            _eventManager.notifyEvent(Event.PERSISTENT_MESSAGE_SIZE_OVERFULL);
+                        }
+                    }
+                    else if(_limitBusted && newSize < getPersistentSizeLowThreshold())
+                    {
+                        long oldSize = _totalStoreSize;
+                        _totalStoreSize = getSizeOnDisk();
+
+                        if(oldSize <= _totalStoreSize)
+                        {
+
+                            reduceSizeOnDisk();
+
+                            _totalStoreSize = getSizeOnDisk();
+
+                        }
+
+                        if(_totalStoreSize < getPersistentSizeLowThreshold())
+                        {
+                            _limitBusted = false;
+                            _eventManager.notifyEvent(Event.PERSISTENT_MESSAGE_SIZE_UNDERFULL);
+                        }
+
+
+                    }
+                }
+            }
+        }
+
+        private long getPersistentSizeLowThreshold()
+        {
+            return _persistentSizeLowThreshold;
+        }
+
+        private long getPersistentSizeHighThreshold()
+        {
+            return _persistentSizeHighThreshold;
+        }
+
+        private void reduceSizeOnDisk()
+        {
+            _environmentFacade.getEnvironment().getConfig().setConfigParam(EnvironmentConfig.ENV_RUN_CLEANER, "false");
+            boolean cleaned = false;
+            while (_environmentFacade.getEnvironment().cleanLog() > 0)
+            {
+                cleaned = true;
+            }
+            if (cleaned)
+            {
+                CheckpointConfig force = new CheckpointConfig();
+                force.setForce(true);
+                _environmentFacade.getEnvironment().checkpoint(force);
+            }
+
+
+            _environmentFacade.getEnvironment().getConfig().setConfigParam(EnvironmentConfig.ENV_RUN_CLEANER, "true");
+        }
+
+        private long getSizeOnDisk()
+        {
+            return _environmentFacade.getEnvironment().getStats(null).getTotalLogSize();
+        }
+
+        private class StoredBDBMessage<T extends StorableMessageMetaData> implements StoredMessage<T>
+        {
+
+            private final long _messageId;
+            private final boolean _isRecovered;
+
+            private T _metaData;
+            private volatile SoftReference<T> _metaDataRef;
+
+            private byte[] _data;
+            private volatile SoftReference<byte[]> _dataRef;
+
+            StoredBDBMessage(long messageId, T metaData)
+            {
+                this(messageId, metaData, false);
+            }
+
+            StoredBDBMessage(long messageId, T metaData, boolean isRecovered)
+            {
+                _messageId = messageId;
+                _isRecovered = isRecovered;
+
+                if(!_isRecovered)
+                {
+                    _metaData = metaData;
+                }
+                _metaDataRef = new SoftReference<T>(metaData);
+            }
+
+            @Override
+            public T getMetaData()
+            {
+                T metaData = _metaDataRef.get();
+                if(metaData == null)
+                {
+                    checkMessageStoreOpen();
+
+                    metaData = (T) getMessageMetaData(_messageId);
+                    _metaDataRef = new SoftReference<T>(metaData);
+                }
+
+                return metaData;
+            }
+
+            @Override
+            public long getMessageNumber()
+            {
+                return _messageId;
+            }
+
+            @Override
+            public void addContent(int offsetInMessage, java.nio.ByteBuffer src)
+            {
+                src = src.slice();
+
+                if(_data == null)
+                {
+                    _data = new byte[src.remaining()];
+                    _dataRef = new SoftReference<byte[]>(_data);
+                    src.duplicate().get(_data);
+                }
+                else
+                {
+                    byte[] oldData = _data;
+                    _data = new byte[oldData.length + src.remaining()];
+                    _dataRef = new SoftReference<byte[]>(_data);
+
+                    System.arraycopy(oldData, 0, _data, 0, oldData.length);
+                    src.duplicate().get(_data, oldData.length, src.remaining());
+                }
+
+            }
+
+            @Override
+            public int getContent(int offsetInMessage, java.nio.ByteBuffer dst)
+            {
+                byte[] data = _dataRef == null ? null : _dataRef.get();
+                if(data != null)
+                {
+                    int length = Math.min(dst.remaining(), data.length - offsetInMessage);
+                    dst.put(data, offsetInMessage, length);
+                    return length;
+                }
+                else
+                {
+                    checkMessageStoreOpen();
+
+                    return BDBMessageStore.this.getContent(_messageId, offsetInMessage, dst);
+                }
+            }
+
+            @Override
+            public ByteBuffer getContent(int offsetInMessage, int size)
+            {
+                byte[] data = _dataRef == null ? null : _dataRef.get();
+                if(data != null)
+                {
+                    return ByteBuffer.wrap(data,offsetInMessage,size);
+                }
+                else
+                {
+                    ByteBuffer buf = ByteBuffer.allocate(size);
+                    int length = getContent(offsetInMessage, buf);
+                    buf.limit(length);
+                    buf.position(0);
+                    return  buf;
+                }
+            }
+
+            synchronized void store(com.sleepycat.je.Transaction txn)
+            {
+                if (!stored())
+                {
+                    try
+                    {
+                        _dataRef = new SoftReference<byte[]>(_data);
+                        BDBMessageStore.this.storeMetaData(txn, _messageId, _metaData);
+                        BDBMessageStore.this.addContent(txn, _messageId, 0,
+                                                        _data == null ? ByteBuffer.allocate(0) : ByteBuffer.wrap(_data));
+                    }
+                    finally
+                    {
+                        _metaData = null;
+                        _data = null;
+                    }
+                }
+            }
+
+            @Override
+            public synchronized StoreFuture flushToStore()
+            {
+                if(!stored())
+                {
+                    checkMessageStoreOpen();
+
+                    com.sleepycat.je.Transaction txn;
+                    try
+                    {
+                        txn = _environmentFacade.getEnvironment().beginTransaction(
+                                null, null);
+                    }
+                    catch (DatabaseException e)
+                    {
+                        throw _environmentFacade.handleDatabaseException("failed to begin transaction", e);
+                    }
+                    store(txn);
+                    _environmentFacade.commit(txn);
+                    _committer.commit(txn, true);
+
+                    storedSizeChangeOccurred(getMetaData().getContentSize());
+                }
+                return StoreFuture.IMMEDIATE_FUTURE;
+            }
+
+            @Override
+            public void remove()
+            {
+                checkMessageStoreOpen();
+
+                int delta = getMetaData().getContentSize();
+                removeMessage(_messageId, false);
+                storedSizeChangeOccurred(-delta);
+            }
+
+            private boolean stored()
+            {
+                return _metaData == null || _isRecovered;
+            }
+
+            @Override
+            public String toString()
+            {
+                return this.getClass() + "[messageId=" + _messageId + "]";
+            }
+        }
+
+
+        private class BDBTransaction implements org.apache.qpid.server.store.Transaction
+        {
+            private com.sleepycat.je.Transaction _txn;
+            private int _storeSizeIncrease;
+
+            private BDBTransaction() throws StoreException
+            {
+                try
+                {
+                    _txn = _environmentFacade.getEnvironment().beginTransaction(null, null);
+                }
+                catch(DatabaseException e)
+                {
+                    throw _environmentFacade.handleDatabaseException("Cannot create store transaction", e);
+                }
+            }
+
+            @Override
+            public void enqueueMessage(TransactionLogResource queue, EnqueueableMessage message) throws StoreException
+            {
+                checkMessageStoreOpen();
+
+                if(message.getStoredMessage() instanceof StoredBDBMessage)
+                {
+                    final StoredBDBMessage storedMessage = (StoredBDBMessage) message.getStoredMessage();
+                    storedMessage.store(_txn);
+                    _storeSizeIncrease += storedMessage.getMetaData().getContentSize();
+                }
+
+                BDBMessageStore.this.enqueueMessage(_txn, queue, message.getMessageNumber());
+            }
+
+            @Override
+            public void dequeueMessage(TransactionLogResource queue, EnqueueableMessage message) throws StoreException
+            {
+                checkMessageStoreOpen();
+
+                BDBMessageStore.this.dequeueMessage(_txn, queue, message.getMessageNumber());
+            }
+
+            @Override
+            public void commitTran() throws StoreException
+            {
+                checkMessageStoreOpen();
+
+                BDBMessageStore.this.commitTranImpl(_txn, true);
+                BDBMessageStore.this.storedSizeChangeOccurred(_storeSizeIncrease);
+            }
+
+            @Override
+            public StoreFuture commitTranAsync() throws StoreException
+            {
+                checkMessageStoreOpen();
+
+                BDBMessageStore.this.storedSizeChangeOccurred(_storeSizeIncrease);
+                return BDBMessageStore.this.commitTranImpl(_txn, false);
+            }
+
+            @Override
+            public void abortTran() throws StoreException
+            {
+                checkMessageStoreOpen();
+
+                BDBMessageStore.this.abortTran(_txn);
+            }
+
+            @Override
+            public void removeXid(long format, byte[] globalId, byte[] branchId) throws StoreException
+            {
+                checkMessageStoreOpen();
+
+                BDBMessageStore.this.removeXid(_txn, format, globalId, branchId);
+            }
+
+            @Override
+            public void recordXid(long format, byte[] globalId, byte[] branchId, Record[] enqueues,
+                                  Record[] dequeues) throws StoreException
+            {
+                checkMessageStoreOpen();
+
+                BDBMessageStore.this.recordXid(_txn, format, globalId, branchId, enqueues, dequeues);
+            }
+        }
+
+    }
+
+}
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
deleted file mode 100644
index 9fc90f9d7c..0000000000
--- a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStore.java
+++ /dev/null
@@ -1,1766 +0,0 @@
-/*
- *
- * 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.lang.ref.SoftReference;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Random;
-import java.util.UUID;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicLong;
-
-import org.apache.log4j.Logger;
-import org.apache.qpid.server.message.EnqueueableMessage;
-import org.apache.qpid.server.model.ConfiguredObject;
-import org.apache.qpid.server.model.VirtualHostNode;
-import org.apache.qpid.server.store.ConfiguredObjectRecord;
-import org.apache.qpid.server.store.DurableConfigurationStore;
-import org.apache.qpid.server.store.Event;
-import org.apache.qpid.server.store.EventListener;
-import org.apache.qpid.server.store.EventManager;
-import org.apache.qpid.server.store.MessageStore;
-import org.apache.qpid.server.store.StorableMessageMetaData;
-import org.apache.qpid.server.store.StoreException;
-import org.apache.qpid.server.store.StoreFuture;
-import org.apache.qpid.server.store.StoredMemoryMessage;
-import org.apache.qpid.server.store.StoredMessage;
-import org.apache.qpid.server.store.TransactionLogResource;
-import org.apache.qpid.server.store.Xid;
-import org.apache.qpid.server.store.berkeleydb.EnvironmentFacadeFactory.EnvironmentFacadeTask;
-import org.apache.qpid.server.store.berkeleydb.entry.HierarchyKey;
-import org.apache.qpid.server.store.berkeleydb.entry.PreparedTransaction;
-import org.apache.qpid.server.store.berkeleydb.entry.QueueEntryKey;
-import org.apache.qpid.server.store.berkeleydb.tuple.ConfiguredObjectBinding;
-import org.apache.qpid.server.store.berkeleydb.tuple.ContentBinding;
-import org.apache.qpid.server.store.berkeleydb.tuple.HierarchyKeyBinding;
-import org.apache.qpid.server.store.berkeleydb.tuple.MessageMetaDataBinding;
-import org.apache.qpid.server.store.berkeleydb.tuple.PreparedTransactionBinding;
-import org.apache.qpid.server.store.berkeleydb.tuple.QueueEntryBinding;
-import org.apache.qpid.server.store.berkeleydb.tuple.UUIDTupleBinding;
-import org.apache.qpid.server.store.berkeleydb.tuple.XidBinding;
-import org.apache.qpid.server.store.berkeleydb.upgrade.Upgrader;
-import org.apache.qpid.server.store.handler.ConfiguredObjectRecordHandler;
-import org.apache.qpid.server.store.handler.DistributedTransactionHandler;
-import org.apache.qpid.server.store.handler.MessageHandler;
-import org.apache.qpid.server.store.handler.MessageInstanceHandler;
-import org.apache.qpid.server.util.MapValueConverter;
-import org.apache.qpid.util.FileUtils;
-
-import com.sleepycat.bind.tuple.ByteBinding;
-import com.sleepycat.bind.tuple.LongBinding;
-import com.sleepycat.je.CheckpointConfig;
-import com.sleepycat.je.Cursor;
-import com.sleepycat.je.Database;
-import com.sleepycat.je.DatabaseConfig;
-import com.sleepycat.je.DatabaseEntry;
-import com.sleepycat.je.DatabaseException;
-import com.sleepycat.je.EnvironmentConfig;
-import com.sleepycat.je.LockConflictException;
-import com.sleepycat.je.LockMode;
-import com.sleepycat.je.OperationStatus;
-import com.sleepycat.je.Transaction;
-
-/**
- * BDBMessageStore implements a persistent {@link MessageStore} using the BDB high performance log.
- *
- * <p/><table id="crc"><caption>CRC Card</caption> <tr><th> Responsibilities <th> Collaborations <tr><td> Accept
- * transaction boundary demarcations: Begin, Commit, Abort. <tr><td> Store and remove queues. <tr><td> Store and remove
- * exchanges. <tr><td> Store and remove messages. <tr><td> Bind and unbind queues to exchanges. <tr><td> Enqueue and
- * dequeue messages to queues. <tr><td> Generate message identifiers. </table>
- */
-public class BDBMessageStore implements MessageStore, DurableConfigurationStore
-{
-    private static final Logger LOGGER = Logger.getLogger(BDBMessageStore.class);
-
-    public static final int VERSION = 8;
-    private static final int LOCK_RETRY_ATTEMPTS = 5;
-    private static String CONFIGURED_OBJECTS_DB_NAME = "CONFIGURED_OBJECTS";
-    private static String CONFIGURED_OBJECT_HIERARCHY_DB_NAME = "CONFIGURED_OBJECT_HIERARCHY";
-
-    private static String MESSAGE_META_DATA_DB_NAME = "MESSAGE_METADATA";
-    private static String MESSAGE_CONTENT_DB_NAME = "MESSAGE_CONTENT";
-    private static String DELIVERY_DB_NAME = "QUEUE_ENTRIES";
-
-    //TODO: Add upgrader to remove BRIDGES and LINKS
-    private static String BRIDGEDB_NAME = "BRIDGES";
-    private static String LINKDB_NAME = "LINKS";
-    private static String XID_DB_NAME = "XIDS";
-    private static final String[] CONFIGURATION_STORE_DATABASE_NAMES = new String[] { CONFIGURED_OBJECTS_DB_NAME, CONFIGURED_OBJECT_HIERARCHY_DB_NAME};
-    private static final String[] MESSAGE_STORE_DATABASE_NAMES = new String[] { MESSAGE_META_DATA_DB_NAME, MESSAGE_CONTENT_DB_NAME, DELIVERY_DB_NAME, BRIDGEDB_NAME, LINKDB_NAME, XID_DB_NAME };
-
-    private EnvironmentFacade _environmentFacade;
-    private final AtomicLong _messageId = new AtomicLong(0);
-
-    private final AtomicBoolean _messageStoreOpen = new AtomicBoolean();
-    private final AtomicBoolean _configurationStoreOpen = new AtomicBoolean();
-
-    private long _totalStoreSize;
-    private boolean _limitBusted;
-    private long _persistentSizeLowThreshold;
-    private long _persistentSizeHighThreshold;
-
-    private final EventManager _eventManager = new EventManager();
-
-    private final EnvironmentFacadeFactory _environmentFacadeFactory;
-
-    private volatile Committer _committer;
-
-    private boolean _isMessageStoreProvider;
-
-    private String _storeLocation;
-
-    public BDBMessageStore()
-    {
-        this(new StandardEnvironmentFacadeFactory());
-    }
-
-    public BDBMessageStore(EnvironmentFacadeFactory environmentFacadeFactory)
-    {
-        _environmentFacadeFactory = environmentFacadeFactory;
-    }
-
-    @Override
-    public void addEventListener(EventListener eventListener, Event... events)
-    {
-        _eventManager.addEventListener(eventListener, events);
-    }
-
-    @Override
-    public void openConfigurationStore(ConfiguredObject<?> parent, Map<String, Object> storeSettings)
-    {
-        if (_configurationStoreOpen.compareAndSet(false,  true))
-        {
-            if (_environmentFacade == null)
-            {
-                EnvironmentFacadeTask[] initialisationTasks = null;
-                _isMessageStoreProvider = MapValueConverter.getBooleanAttribute(VirtualHostNode.IS_MESSAGE_STORE_PROVIDER, storeSettings, false);
-                if (_isMessageStoreProvider)
-                {
-                    String[] databaseNames = new String[CONFIGURATION_STORE_DATABASE_NAMES.length + MESSAGE_STORE_DATABASE_NAMES.length];
-                    System.arraycopy(CONFIGURATION_STORE_DATABASE_NAMES, 0, databaseNames, 0, CONFIGURATION_STORE_DATABASE_NAMES.length);
-                    System.arraycopy(MESSAGE_STORE_DATABASE_NAMES, 0, databaseNames, CONFIGURATION_STORE_DATABASE_NAMES.length, MESSAGE_STORE_DATABASE_NAMES.length);
-                    initialisationTasks = new EnvironmentFacadeTask[]{new UpgradeTask(parent), new OpenDatabasesTask(databaseNames), new DiskSpaceTask(), new MaxMessageIdTask() };
-                }
-                else
-                {
-                    initialisationTasks = new EnvironmentFacadeTask[]{new UpgradeTask(parent), new OpenDatabasesTask(CONFIGURATION_STORE_DATABASE_NAMES)};
-                }
-                _environmentFacade = _environmentFacadeFactory.createEnvironmentFacade(storeSettings, initialisationTasks);
-                _storeLocation = _environmentFacade.getStoreLocation();
-            }
-            else
-            {
-                throw new IllegalStateException("The database have been already opened as message store");
-            }
-        }
-    }
-
-    @Override
-    public void visitConfiguredObjectRecords(ConfiguredObjectRecordHandler handler)
-    {
-        checkConfigurationStoreOpen();
-
-        try
-        {
-            handler.begin();
-            doVisitAllConfiguredObjectRecords(handler);
-            handler.end();
-        }
-        catch (DatabaseException e)
-        {
-            throw _environmentFacade.handleDatabaseException("Cannot visit configured object records", e);
-        }
-
-    }
-
-    private void doVisitAllConfiguredObjectRecords(ConfiguredObjectRecordHandler handler)
-    {
-        Map<UUID, BDBConfiguredObjectRecord> configuredObjects = new HashMap<UUID, BDBConfiguredObjectRecord>();
-        Cursor objectsCursor = null;
-        Cursor hierarchyCursor = null;
-        try
-        {
-            objectsCursor = getConfiguredObjectsDb().openCursor(null, null);
-            DatabaseEntry key = new DatabaseEntry();
-            DatabaseEntry value = new DatabaseEntry();
-
-
-            while (objectsCursor.getNext(key, value, LockMode.RMW) == OperationStatus.SUCCESS)
-            {
-                UUID id = UUIDTupleBinding.getInstance().entryToObject(key);
-
-                BDBConfiguredObjectRecord configuredObject =
-                        (BDBConfiguredObjectRecord) new ConfiguredObjectBinding(id).entryToObject(value);
-                configuredObjects.put(configuredObject.getId(), configuredObject);
-            }
-
-            // set parents
-            hierarchyCursor = getConfiguredObjectHierarchyDb().openCursor(null, null);
-            while (hierarchyCursor.getNext(key, value, LockMode.RMW) == OperationStatus.SUCCESS)
-            {
-                HierarchyKey hk = HierarchyKeyBinding.getInstance().entryToObject(key);
-                UUID parentId = UUIDTupleBinding.getInstance().entryToObject(value);
-                BDBConfiguredObjectRecord child = configuredObjects.get(hk.getChildId());
-                if(child != null)
-                {
-                    ConfiguredObjectRecord parent = configuredObjects.get(parentId);
-                    if(parent != null)
-                    {
-                        child.addParent(hk.getParentType(), parent);
-                    }
-                }
-            }
-        }
-        finally
-        {
-            closeCursorSafely(objectsCursor);
-            closeCursorSafely(hierarchyCursor);
-        }
-
-        for (ConfiguredObjectRecord record : configuredObjects.values())
-        {
-            boolean shoudlContinue = handler.handle(record);
-            if (!shoudlContinue)
-            {
-                break;
-            }
-        }
-
-    }
-
-    @Override
-    public void openMessageStore(ConfiguredObject<?> parent, Map<String, Object> messageStoreSettings) throws StoreException
-    {
-        if (_messageStoreOpen.compareAndSet(false, true))
-        {
-
-            Object overfullAttr = messageStoreSettings.get(MessageStore.OVERFULL_SIZE);
-            Object underfullAttr = messageStoreSettings.get(MessageStore.UNDERFULL_SIZE);
-
-            _persistentSizeHighThreshold = overfullAttr == null ? -1l :
-                                           overfullAttr instanceof Number ? ((Number) overfullAttr).longValue() : Long.parseLong(overfullAttr.toString());
-            _persistentSizeLowThreshold = underfullAttr == null ? _persistentSizeHighThreshold :
-                                           underfullAttr instanceof Number ? ((Number) underfullAttr).longValue() : Long.parseLong(underfullAttr.toString());
-
-            if(_persistentSizeLowThreshold > _persistentSizeHighThreshold || _persistentSizeLowThreshold < 0l)
-            {
-                _persistentSizeLowThreshold = _persistentSizeHighThreshold;
-            }
-
-            if (_environmentFacade == null)
-            {
-                _environmentFacade = _environmentFacadeFactory.createEnvironmentFacade(messageStoreSettings,
-                        new UpgradeTask(parent), new OpenDatabasesTask(MESSAGE_STORE_DATABASE_NAMES), new DiskSpaceTask(), new MaxMessageIdTask());
-                _storeLocation = _environmentFacade.getStoreLocation();
-            }
-
-            _committer = _environmentFacade.createCommitter(parent.getName());
-            _committer.start();
-        }
-    }
-
-    @Override
-    public org.apache.qpid.server.store.Transaction newTransaction() throws StoreException
-    {
-        checkMessageStoreOpen();
-
-        return new BDBTransaction();
-    }
-
-    @Override
-    public String getStoreLocation()
-    {
-        return _storeLocation;
-    }
-
-    public EnvironmentFacade getEnvironmentFacade()
-    {
-        return _environmentFacade;
-    }
-
-    @Override
-    public void closeMessageStore() throws StoreException
-    {
-        if (_messageStoreOpen.compareAndSet(true, false))
-        {
-            try
-            {
-                if (_committer != null)
-                {
-                    _committer.close();
-                }
-            }
-            finally
-            {
-                if (!_configurationStoreOpen.get())
-                {
-                    closeEnvironment();
-                }
-            }
-        }
-    }
-
-    @Override
-    public void closeConfigurationStore() throws StoreException
-    {
-        if (_configurationStoreOpen.compareAndSet(true, false))
-        {
-            if (!_messageStoreOpen.get())
-            {
-                closeEnvironment();
-            }
-        }
-    }
-
-    private void closeEnvironment()
-    {
-        if (_environmentFacade != null)
-        {
-            try
-            {
-                _environmentFacade.close();
-                _environmentFacade = null;
-            }
-            catch(DatabaseException e)
-            {
-                throw new StoreException("Exception occured on message store close", e);
-            }
-        }
-    }
-
-    private void closeCursorSafely(Cursor cursor) throws StoreException
-    {
-        if (cursor != null)
-        {
-            try
-            {
-                cursor.close();
-            }
-            catch(DatabaseException e)
-            {
-                throw _environmentFacade.handleDatabaseException("Cannot close cursor", e);
-            }
-        }
-    }
-
-
-    void removeMessage(long messageId, boolean sync) throws StoreException
-    {
-        boolean complete = false;
-        com.sleepycat.je.Transaction tx = null;
-
-        Random rand = null;
-        int attempts = 0;
-        try
-        {
-            do
-            {
-                tx = null;
-                try
-                {
-                    tx = _environmentFacade.beginTransaction();
-
-                    //remove the message meta data from the store
-                    DatabaseEntry key = new DatabaseEntry();
-                    LongBinding.longToEntry(messageId, key);
-
-                    if (LOGGER.isDebugEnabled())
-                    {
-                        LOGGER.debug("Removing message id " + messageId);
-                    }
-
-
-                    OperationStatus status = getMessageMetaDataDb().delete(tx, key);
-                    if (status == OperationStatus.NOTFOUND)
-                    {
-                        LOGGER.info("Message not found (attempt to remove failed - probably application initiated rollback) " +
-                        messageId);
-                    }
-
-                    if (LOGGER.isDebugEnabled())
-                    {
-                        LOGGER.debug("Deleted metadata for message " + messageId);
-                    }
-
-                    //now remove the content data from the store if there is any.
-                    DatabaseEntry contentKeyEntry = new DatabaseEntry();
-                    LongBinding.longToEntry(messageId, contentKeyEntry);
-                    getMessageContentDb().delete(tx, contentKeyEntry);
-
-                    if (LOGGER.isDebugEnabled())
-                    {
-                        LOGGER.debug("Deleted content for message " + messageId);
-                    }
-
-                    _environmentFacade.commit(tx);
-                    _committer.commit(tx, sync);
-
-                    complete = true;
-                    tx = null;
-                }
-                catch (LockConflictException e)
-                {
-                    try
-                    {
-                        if(tx != null)
-                        {
-                            tx.abort();
-                        }
-                    }
-                    catch(DatabaseException e2)
-                    {
-                        LOGGER.warn("Unable to abort transaction after LockConflictExcption on removal of message with id " + messageId, e2);
-                        // rethrow the original log conflict exception, the secondary exception should already have
-                        // been logged.
-                        throw _environmentFacade.handleDatabaseException("Cannot remove message with id " + messageId, e);
-                    }
-
-
-                    LOGGER.warn("Lock timeout exception. Retrying (attempt "
-                              + (attempts+1) + " of "+ LOCK_RETRY_ATTEMPTS +") " + e);
-
-                    if(++attempts < LOCK_RETRY_ATTEMPTS)
-                    {
-                        if(rand == null)
-                        {
-                            rand = new Random();
-                        }
-
-                        try
-                        {
-                            Thread.sleep(500l + (long)(500l * rand.nextDouble()));
-                        }
-                        catch (InterruptedException e1)
-                        {
-
-                        }
-                    }
-                    else
-                    {
-                        // rethrow the lock conflict exception since we could not solve by retrying
-                        throw _environmentFacade.handleDatabaseException("Cannot remove messages", e);
-                    }
-                }
-            }
-            while(!complete);
-        }
-        catch (DatabaseException e)
-        {
-            LOGGER.error("Unexpected BDB exception", e);
-
-            try
-            {
-                abortTransactionIgnoringException("Error aborting transaction on removal of message with id " + messageId, tx);
-            }
-            finally
-            {
-                tx = null;
-            }
-
-            throw _environmentFacade.handleDatabaseException("Error removing message with id " + messageId + " from database: " + e.getMessage(), e);
-        }
-        finally
-        {
-            try
-            {
-                abortTransactionIgnoringException("Error aborting transaction on removal of message with id " + messageId, tx);
-            }
-            finally
-            {
-                tx = null;
-            }
-        }
-    }
-
-    private void abortTransactionIgnoringException(String errorMessage, com.sleepycat.je.Transaction tx)
-    {
-        try
-        {
-            if (tx != null)
-            {
-                tx.abort();
-            }
-        }
-        catch (DatabaseException e1)
-        {
-            // We need the possible side effect of the handler restarting the environment but don't care about the exception
-            _environmentFacade.handleDatabaseException(null, e1);
-            LOGGER.warn(errorMessage, e1);
-        }
-    }
-
-    @Override
-    public void create(ConfiguredObjectRecord configuredObject) throws StoreException
-    {
-        checkConfigurationStoreOpen();
-
-        if (LOGGER.isDebugEnabled())
-        {
-            LOGGER.debug("Create " + configuredObject);
-        }
-
-        com.sleepycat.je.Transaction txn = null;
-        try
-        {
-            txn = _environmentFacade.getEnvironment().beginTransaction(null, null);
-            storeConfiguredObjectEntry(txn, configuredObject);
-            txn.commit();
-            txn = null;
-        }
-        catch (DatabaseException e)
-        {
-            throw _environmentFacade.handleDatabaseException("Error creating configured object " + configuredObject
-                    + " in database: " + e.getMessage(), e);
-        }
-        finally
-        {
-            if (txn != null)
-            {
-                abortTransactionIgnoringException("Error creating configured object", txn);
-            }
-        }
-    }
-
-    @Override
-    public UUID[] remove(final ConfiguredObjectRecord... objects) throws StoreException
-    {
-        checkConfigurationStoreOpen();
-
-        com.sleepycat.je.Transaction txn = null;
-        try
-        {
-            txn = _environmentFacade.getEnvironment().beginTransaction(null, null);
-
-            Collection<UUID> removed = new ArrayList<UUID>(objects.length);
-            for(ConfiguredObjectRecord record : objects)
-            {
-                if(removeConfiguredObject(txn, record) == OperationStatus.SUCCESS)
-                {
-                    removed.add(record.getId());
-                }
-            }
-
-            txn.commit();
-            txn = null;
-            return removed.toArray(new UUID[removed.size()]);
-        }
-        catch (DatabaseException e)
-        {
-            throw _environmentFacade.handleDatabaseException("Error deleting configured objects from database", e);
-        }
-        finally
-        {
-            if (txn != null)
-            {
-                abortTransactionIgnoringException("Error deleting configured objects", txn);
-            }
-        }
-
-    }
-
-    @Override
-    public void update(boolean createIfNecessary, ConfiguredObjectRecord... records) throws StoreException
-    {
-        checkConfigurationStoreOpen();
-
-        com.sleepycat.je.Transaction txn = null;
-        try
-        {
-            txn = _environmentFacade.getEnvironment().beginTransaction(null, null);
-            for(ConfiguredObjectRecord record : records)
-            {
-                update(createIfNecessary, record, txn);
-            }
-            txn.commit();
-            txn = null;
-        }
-        catch (DatabaseException e)
-        {
-            throw _environmentFacade.handleDatabaseException("Error updating configuration details within the store: " + e,e);
-        }
-        finally
-        {
-            if (txn != null)
-            {
-                abortTransactionIgnoringException("Error updating configuration details within the store", txn);
-            }
-        }
-
-    }
-
-    private void update(boolean createIfNecessary, ConfiguredObjectRecord record, com.sleepycat.je.Transaction txn) throws StoreException
-    {
-        if (LOGGER.isDebugEnabled())
-        {
-            LOGGER.debug("Updating, creating " + createIfNecessary + " : "  + record);
-        }
-
-        DatabaseEntry key = new DatabaseEntry();
-        UUIDTupleBinding keyBinding = UUIDTupleBinding.getInstance();
-        keyBinding.objectToEntry(record.getId(), key);
-
-        DatabaseEntry value = new DatabaseEntry();
-        DatabaseEntry newValue = new DatabaseEntry();
-        ConfiguredObjectBinding configuredObjectBinding = ConfiguredObjectBinding.getInstance();
-
-        OperationStatus status = getConfiguredObjectsDb().get(txn, key, value, LockMode.DEFAULT);
-        final boolean isNewRecord = status == OperationStatus.NOTFOUND;
-        if (status == OperationStatus.SUCCESS || (createIfNecessary && isNewRecord))
-        {
-            // write the updated entry to the store
-            configuredObjectBinding.objectToEntry(record, newValue);
-            status = getConfiguredObjectsDb().put(txn, key, newValue);
-            if (status != OperationStatus.SUCCESS)
-            {
-                throw new StoreException("Error updating configuration details within the store: " + status);
-            }
-            if(isNewRecord)
-            {
-                writeHierarchyRecords(txn, record);
-            }
-        }
-        else if (status != OperationStatus.NOTFOUND)
-        {
-            throw new StoreException("Error finding configuration details within the store: " + status);
-        }
-    }
-
-    /**
-     * Places a message onto a specified queue, in a given transaction.
-     *
-     * @param tx   The transaction for the operation.
-     * @param queue     The the queue to place the message on.
-     * @param messageId The message to enqueue.
-     *
-     * @throws StoreException If the operation fails for any reason.
-     */
-    private void enqueueMessage(final com.sleepycat.je.Transaction tx, final TransactionLogResource queue,
-                               long messageId) throws StoreException
-    {
-
-        DatabaseEntry key = new DatabaseEntry();
-        QueueEntryBinding keyBinding = QueueEntryBinding.getInstance();
-        QueueEntryKey dd = new QueueEntryKey(queue.getId(), messageId);
-        keyBinding.objectToEntry(dd, key);
-        DatabaseEntry value = new DatabaseEntry();
-        ByteBinding.byteToEntry((byte) 0, value);
-
-        try
-        {
-            if (LOGGER.isDebugEnabled())
-            {
-                LOGGER.debug("Enqueuing message " + messageId + " on queue "
-                        + queue.getName() + " with id " + queue.getId() + " in transaction " + tx);
-            }
-            getDeliveryDb().put(tx, key, value);
-        }
-        catch (DatabaseException e)
-        {
-            LOGGER.error("Failed to enqueue: " + e.getMessage(), e);
-            throw _environmentFacade.handleDatabaseException("Error writing enqueued message with id " + messageId + " for queue "
-                    + queue.getName() + " with id " + queue.getId() + " to database", e);
-        }
-    }
-
-    /**
-     * Extracts a message from a specified queue, in a given transaction.
-     *
-     * @param tx   The transaction for the operation.
-     * @param queue     The queue to take the message from.
-     * @param messageId The message to dequeue.
-     *
-     * @throws StoreException If the operation fails for any reason, or if the specified message does not exist.
-     */
-    private void dequeueMessage(final com.sleepycat.je.Transaction tx, final TransactionLogResource queue,
-                               long messageId) throws StoreException
-    {
-
-        DatabaseEntry key = new DatabaseEntry();
-        QueueEntryBinding keyBinding = QueueEntryBinding.getInstance();
-        QueueEntryKey queueEntryKey = new QueueEntryKey(queue.getId(), messageId);
-        UUID id = queue.getId();
-        keyBinding.objectToEntry(queueEntryKey, key);
-        if (LOGGER.isDebugEnabled())
-        {
-            LOGGER.debug("Dequeue message id " + messageId + " from queue "
-                    + queue.getName() + " with id " + id);
-        }
-
-        try
-        {
-
-            OperationStatus status = getDeliveryDb().delete(tx, key);
-            if (status == OperationStatus.NOTFOUND)
-            {
-                throw new StoreException("Unable to find message with id " + messageId + " on queue "
-                        + queue.getName() + " with id "  + id);
-            }
-            else if (status != OperationStatus.SUCCESS)
-            {
-                throw new StoreException("Unable to remove message with id " + messageId + " on queue"
-                        + queue.getName() + " with id " + id);
-            }
-
-            if (LOGGER.isDebugEnabled())
-            {
-                LOGGER.debug("Removed message " + messageId + " on queue "
-                        + queue.getName() + " with id " + id);
-
-            }
-        }
-        catch (DatabaseException e)
-        {
-
-            LOGGER.error("Failed to dequeue message " + messageId + " in transaction " + tx , e);
-
-            throw _environmentFacade.handleDatabaseException("Error accessing database while dequeuing message: " + e.getMessage(), e);
-        }
-    }
-
-
-    private void recordXid(com.sleepycat.je.Transaction txn,
-                           long format,
-                           byte[] globalId,
-                           byte[] branchId,
-                           org.apache.qpid.server.store.Transaction.Record[] enqueues,
-                           org.apache.qpid.server.store.Transaction.Record[] dequeues) throws StoreException
-    {
-        DatabaseEntry key = new DatabaseEntry();
-        Xid xid = new Xid(format, globalId, branchId);
-        XidBinding keyBinding = XidBinding.getInstance();
-        keyBinding.objectToEntry(xid,key);
-
-        DatabaseEntry value = new DatabaseEntry();
-        PreparedTransaction preparedTransaction = new PreparedTransaction(enqueues, dequeues);
-        PreparedTransactionBinding valueBinding = new PreparedTransactionBinding();
-        valueBinding.objectToEntry(preparedTransaction, value);
-
-        try
-        {
-            getXidDb().put(txn, key, value);
-        }
-        catch (DatabaseException e)
-        {
-            LOGGER.error("Failed to write xid: " + e.getMessage(), e);
-            throw _environmentFacade.handleDatabaseException("Error writing xid to database", e);
-        }
-    }
-
-    private void removeXid(com.sleepycat.je.Transaction txn, long format, byte[] globalId, byte[] branchId)
-            throws StoreException
-    {
-        DatabaseEntry key = new DatabaseEntry();
-        Xid xid = new Xid(format, globalId, branchId);
-        XidBinding keyBinding = XidBinding.getInstance();
-
-        keyBinding.objectToEntry(xid, key);
-
-
-        try
-        {
-
-            OperationStatus status = getXidDb().delete(txn, key);
-            if (status == OperationStatus.NOTFOUND)
-            {
-                throw new StoreException("Unable to find xid");
-            }
-            else if (status != OperationStatus.SUCCESS)
-            {
-                throw new StoreException("Unable to remove xid");
-            }
-
-        }
-        catch (DatabaseException e)
-        {
-
-            LOGGER.error("Failed to remove xid in transaction " + txn, e);
-
-            throw _environmentFacade.handleDatabaseException("Error accessing database while removing xid: " + e.getMessage(), e);
-        }
-    }
-
-    /**
-     * Commits all operations performed within a given transaction.
-     *
-     * @param tx The transaction to commit all operations for.
-     *
-     * @throws StoreException If the operation fails for any reason.
-     */
-    private StoreFuture commitTranImpl(final com.sleepycat.je.Transaction tx, boolean syncCommit) throws StoreException
-    {
-        if (tx == null)
-        {
-            throw new StoreException("Fatal internal error: transactional is null at commitTran");
-        }
-
-        _environmentFacade.commit(tx);
-        StoreFuture result =  _committer.commit(tx, syncCommit);
-
-        if (LOGGER.isDebugEnabled())
-        {
-            String transactionType = syncCommit ? "synchronous" : "asynchronous";
-            LOGGER.debug("commitTranImpl completed " + transactionType + " transaction " + tx);
-        }
-
-        return result;
-    }
-
-    /**
-     * Abandons all operations performed within a given transaction.
-     *
-     * @param tx The transaction to abandon.
-     *
-     * @throws StoreException If the operation fails for any reason.
-     */
-    private void abortTran(final com.sleepycat.je.Transaction tx) throws StoreException
-    {
-        if (LOGGER.isDebugEnabled())
-        {
-            LOGGER.debug("abortTran called for transaction " + tx);
-        }
-
-        try
-        {
-            tx.abort();
-        }
-        catch (DatabaseException e)
-        {
-            throw _environmentFacade.handleDatabaseException("Error aborting transaction: " + e.getMessage(), e);
-        }
-    }
-
-    /**
-     * Return a valid, currently unused message id.
-     *
-     * @return A fresh message id.
-     */
-    private long getNewMessageId()
-    {
-        return _messageId.incrementAndGet();
-    }
-
-    /**
-     * Stores a chunk of message data.
-     *
-     * @param tx         The transaction for the operation.
-     * @param messageId       The message to store the data for.
-     * @param offset          The offset of the data chunk in the message.
-     * @param contentBody     The content of the data chunk.
-     *
-     * @throws StoreException If the operation fails for any reason, or if the specified message does not exist.
-     */
-    private void addContent(final com.sleepycat.je.Transaction tx, long messageId, int offset,
-                                      ByteBuffer contentBody) throws StoreException
-    {
-        DatabaseEntry key = new DatabaseEntry();
-        LongBinding.longToEntry(messageId, key);
-        DatabaseEntry value = new DatabaseEntry();
-        ContentBinding messageBinding = ContentBinding.getInstance();
-        messageBinding.objectToEntry(contentBody.array(), value);
-        try
-        {
-            OperationStatus status = getMessageContentDb().put(tx, key, value);
-            if (status != OperationStatus.SUCCESS)
-            {
-                throw new StoreException("Error adding content for message id " + messageId + ": " + status);
-            }
-
-            if (LOGGER.isDebugEnabled())
-            {
-                LOGGER.debug("Storing content for message " + messageId + " in transaction " + tx);
-
-            }
-        }
-        catch (DatabaseException e)
-        {
-            throw _environmentFacade.handleDatabaseException("Error writing AMQMessage with id " + messageId + " to database: " + e.getMessage(), e);
-        }
-    }
-
-    /**
-     * Stores message meta-data.
-     *
-     * @param tx         The transaction for the operation.
-     * @param messageId       The message to store the data for.
-     * @param messageMetaData The message meta data to store.
-     *
-     * @throws StoreException If the operation fails for any reason, or if the specified message does not exist.
-     */
-    private void storeMetaData(final com.sleepycat.je.Transaction tx, long messageId,
-                               StorableMessageMetaData messageMetaData)
-            throws StoreException
-    {
-        if (LOGGER.isDebugEnabled())
-        {
-            LOGGER.debug("storeMetaData called for transaction " + tx
-                    + ", messageId " + messageId
-                    + ", messageMetaData " + messageMetaData);
-        }
-
-        DatabaseEntry key = new DatabaseEntry();
-        LongBinding.longToEntry(messageId, key);
-        DatabaseEntry value = new DatabaseEntry();
-
-        MessageMetaDataBinding messageBinding = MessageMetaDataBinding.getInstance();
-        messageBinding.objectToEntry(messageMetaData, value);
-        try
-        {
-            getMessageMetaDataDb().put(tx, key, value);
-            if (LOGGER.isDebugEnabled())
-            {
-                LOGGER.debug("Storing message metadata for message id " + messageId + " in transaction " + tx);
-            }
-        }
-        catch (DatabaseException e)
-        {
-            throw _environmentFacade.handleDatabaseException("Error writing message metadata with id " + messageId + " to database: " + e.getMessage(), e);
-        }
-    }
-
-    /**
-     * Retrieves message meta-data.
-     *
-     * @param messageId The message to get the meta-data for.
-     *
-     * @return The message meta data.
-     *
-     * @throws StoreException If the operation fails for any reason, or if the specified message does not exist.
-     */
-    StorableMessageMetaData getMessageMetaData(long messageId) throws StoreException
-    {
-        if (LOGGER.isDebugEnabled())
-        {
-            LOGGER.debug("public MessageMetaData getMessageMetaData(Long messageId = "
-                       + messageId + "): called");
-        }
-
-        DatabaseEntry key = new DatabaseEntry();
-        LongBinding.longToEntry(messageId, key);
-        DatabaseEntry value = new DatabaseEntry();
-        MessageMetaDataBinding messageBinding = MessageMetaDataBinding.getInstance();
-
-        try
-        {
-            OperationStatus status = getMessageMetaDataDb().get(null, key, value, LockMode.READ_UNCOMMITTED);
-            if (status != OperationStatus.SUCCESS)
-            {
-                throw new StoreException("Metadata not found for message with id " + messageId);
-            }
-
-            StorableMessageMetaData mdd = messageBinding.entryToObject(value);
-
-            return mdd;
-        }
-        catch (DatabaseException e)
-        {
-            throw _environmentFacade.handleDatabaseException("Error reading message metadata for message with id " + messageId + ": " + e.getMessage(), e);
-        }
-    }
-
-    /**
-     * Fills the provided ByteBuffer with as much content for the specified message as possible, starting
-     * from the specified offset in the message.
-     *
-     * @param messageId The message to get the data for.
-     * @param offset    The offset of the data within the message.
-     * @param dst       The destination of the content read back
-     *
-     * @return The number of bytes inserted into the destination
-     *
-     * @throws StoreException If the operation fails for any reason, or if the specified message does not exist.
-     */
-    int getContent(long messageId, int offset, ByteBuffer dst) throws StoreException
-    {
-        DatabaseEntry contentKeyEntry = new DatabaseEntry();
-        LongBinding.longToEntry(messageId, contentKeyEntry);
-        DatabaseEntry value = new DatabaseEntry();
-        ContentBinding contentTupleBinding = ContentBinding.getInstance();
-
-
-        if (LOGGER.isDebugEnabled())
-        {
-            LOGGER.debug("Message Id: " + messageId + " Getting content body from offset: " + offset);
-        }
-
-        try
-        {
-
-            int written = 0;
-            OperationStatus status = getMessageContentDb().get(null, contentKeyEntry, value, LockMode.READ_UNCOMMITTED);
-            if (status == OperationStatus.SUCCESS)
-            {
-                byte[] dataAsBytes = contentTupleBinding.entryToObject(value);
-                int size = dataAsBytes.length;
-                if (offset > size)
-                {
-                    throw new RuntimeException("Offset " + offset + " is greater than message size " + size
-                            + " for message id " + messageId + "!");
-
-                }
-
-                written = size - offset;
-                if(written > dst.remaining())
-                {
-                    written = dst.remaining();
-                }
-
-                dst.put(dataAsBytes, offset, written);
-            }
-            return written;
-        }
-        catch (DatabaseException e)
-        {
-            throw _environmentFacade.handleDatabaseException("Error getting AMQMessage with id " + messageId + " to database: " + e.getMessage(), e);
-        }
-    }
-
-    @Override
-    public boolean isPersistent()
-    {
-        return true;
-    }
-
-    @Override
-    @SuppressWarnings("unchecked")
-    public <T extends StorableMessageMetaData> StoredMessage<T> addMessage(T metaData)
-    {
-        if(metaData.isPersistent())
-        {
-            return (StoredMessage<T>) new StoredBDBMessage(getNewMessageId(), metaData);
-        }
-        else
-        {
-            return new StoredMemoryMessage(getNewMessageId(), metaData);
-        }
-    }
-
-    private void storeConfiguredObjectEntry(final Transaction txn, ConfiguredObjectRecord configuredObject) throws StoreException
-    {
-        if (LOGGER.isDebugEnabled())
-        {
-            LOGGER.debug("Storing configured object record: " + configuredObject);
-        }
-        DatabaseEntry key = new DatabaseEntry();
-        UUIDTupleBinding uuidBinding = UUIDTupleBinding.getInstance();
-        uuidBinding.objectToEntry(configuredObject.getId(), key);
-
-        DatabaseEntry value = new DatabaseEntry();
-        ConfiguredObjectBinding queueBinding = ConfiguredObjectBinding.getInstance();
-
-        queueBinding.objectToEntry(configuredObject, value);
-        try
-        {
-            OperationStatus status = getConfiguredObjectsDb().put(txn, key, value);
-            if (status != OperationStatus.SUCCESS)
-            {
-                throw new StoreException("Error writing configured object " + configuredObject + " to database: "
-                        + status);
-            }
-            writeHierarchyRecords(txn, configuredObject);
-        }
-        catch (DatabaseException e)
-        {
-            throw _environmentFacade.handleDatabaseException("Error writing configured object " + configuredObject
-                    + " to database: " + e.getMessage(), e);
-        }
-    }
-
-    private void writeHierarchyRecords(final Transaction txn, final ConfiguredObjectRecord configuredObject)
-    {
-        OperationStatus status;
-        HierarchyKeyBinding hierarchyBinding = HierarchyKeyBinding.getInstance();
-        DatabaseEntry hierarchyKey = new DatabaseEntry();
-        DatabaseEntry hierarchyValue = new DatabaseEntry();
-
-        for(Map.Entry<String, ConfiguredObjectRecord> parent : configuredObject.getParents().entrySet())
-        {
-
-            hierarchyBinding.objectToEntry(new HierarchyKey(configuredObject.getId(), parent.getKey()), hierarchyKey);
-            UUIDTupleBinding.getInstance().objectToEntry(parent.getValue().getId(), hierarchyValue);
-            status = getConfiguredObjectHierarchyDb().put(txn, hierarchyKey, hierarchyValue);
-            if (status != OperationStatus.SUCCESS)
-            {
-                throw new StoreException("Error writing configured object " + configuredObject + " parent record to database: "
-                                         + status);
-            }
-        }
-    }
-
-    private OperationStatus removeConfiguredObject(Transaction tx, ConfiguredObjectRecord record) throws StoreException
-    {
-        UUID id = record.getId();
-        Map<String, ConfiguredObjectRecord> parents = record.getParents();
-
-        if (LOGGER.isDebugEnabled())
-        {
-            LOGGER.debug("Removing configured object: " + id);
-        }
-        DatabaseEntry key = new DatabaseEntry();
-        UUIDTupleBinding uuidBinding = UUIDTupleBinding.getInstance();
-        uuidBinding.objectToEntry(id, key);
-        OperationStatus status = getConfiguredObjectsDb().delete(tx, key);
-        if(status == OperationStatus.SUCCESS)
-        {
-            for(String parentType : parents.keySet())
-            {
-                DatabaseEntry hierarchyKey = new DatabaseEntry();
-                HierarchyKeyBinding keyBinding = HierarchyKeyBinding.getInstance();
-                keyBinding.objectToEntry(new HierarchyKey(record.getId(), parentType), hierarchyKey);
-                getConfiguredObjectHierarchyDb().delete(tx, hierarchyKey);
-            }
-        }
-        return status;
-    }
-
-    private class StoredBDBMessage implements StoredMessage<StorableMessageMetaData>
-    {
-
-        private final long _messageId;
-        private final boolean _isRecovered;
-
-        private StorableMessageMetaData _metaData;
-        private volatile SoftReference<StorableMessageMetaData> _metaDataRef;
-
-        private byte[] _data;
-        private volatile SoftReference<byte[]> _dataRef;
-
-        StoredBDBMessage(long messageId, StorableMessageMetaData metaData)
-        {
-            this(messageId, metaData, false);
-        }
-
-        StoredBDBMessage(long messageId, StorableMessageMetaData metaData, boolean isRecovered)
-        {
-            _messageId = messageId;
-            _isRecovered = isRecovered;
-
-            if(!_isRecovered)
-            {
-                _metaData = metaData;
-            }
-            _metaDataRef = new SoftReference<StorableMessageMetaData>(metaData);
-        }
-
-        @Override
-        public StorableMessageMetaData getMetaData()
-        {
-            StorableMessageMetaData metaData = _metaDataRef.get();
-            if(metaData == null)
-            {
-                checkMessageStoreOpen();
-
-                metaData = BDBMessageStore.this.getMessageMetaData(_messageId);
-                _metaDataRef = new SoftReference<StorableMessageMetaData>(metaData);
-            }
-
-            return metaData;
-        }
-
-        @Override
-        public long getMessageNumber()
-        {
-            return _messageId;
-        }
-
-        @Override
-        public void addContent(int offsetInMessage, java.nio.ByteBuffer src)
-        {
-            src = src.slice();
-
-            if(_data == null)
-            {
-                _data = new byte[src.remaining()];
-                _dataRef = new SoftReference<byte[]>(_data);
-                src.duplicate().get(_data);
-            }
-            else
-            {
-                byte[] oldData = _data;
-                _data = new byte[oldData.length + src.remaining()];
-                _dataRef = new SoftReference<byte[]>(_data);
-
-                System.arraycopy(oldData,0,_data,0,oldData.length);
-                src.duplicate().get(_data, oldData.length, src.remaining());
-            }
-
-        }
-
-        @Override
-        public int getContent(int offsetInMessage, java.nio.ByteBuffer dst)
-        {
-            byte[] data = _dataRef == null ? null : _dataRef.get();
-            if(data != null)
-            {
-                int length = Math.min(dst.remaining(), data.length - offsetInMessage);
-                dst.put(data, offsetInMessage, length);
-                return length;
-            }
-            else
-            {
-                checkMessageStoreOpen();
-
-                return BDBMessageStore.this.getContent(_messageId, offsetInMessage, dst);
-            }
-        }
-
-        @Override
-        public ByteBuffer getContent(int offsetInMessage, int size)
-        {
-            byte[] data = _dataRef == null ? null : _dataRef.get();
-            if(data != null)
-            {
-                return ByteBuffer.wrap(data,offsetInMessage,size);
-            }
-            else
-            {
-                ByteBuffer buf = ByteBuffer.allocate(size);
-                int length = getContent(offsetInMessage, buf);
-                buf.limit(length);
-                buf.position(0);
-                return  buf;
-            }
-        }
-
-        synchronized void store(com.sleepycat.je.Transaction txn)
-        {
-            if (!stored())
-            {
-                try
-                {
-                    _dataRef = new SoftReference<byte[]>(_data);
-                    BDBMessageStore.this.storeMetaData(txn, _messageId, _metaData);
-                    BDBMessageStore.this.addContent(txn, _messageId, 0,
-                                                    _data == null ? ByteBuffer.allocate(0) : ByteBuffer.wrap(_data));
-                }
-                finally
-                {
-                    _metaData = null;
-                    _data = null;
-                }
-            }
-        }
-
-        @Override
-        public synchronized StoreFuture flushToStore()
-        {
-            if(!stored())
-            {
-                checkMessageStoreOpen();
-
-                com.sleepycat.je.Transaction txn;
-                try
-                {
-                    txn = _environmentFacade.beginTransaction();
-                }
-                catch (DatabaseException e)
-                {
-                    throw _environmentFacade.handleDatabaseException("failed to begin transaction", e);
-                }
-                store(txn);
-                _environmentFacade.commit(txn);
-                _committer.commit(txn, true);
-
-                storedSizeChangeOccured(getMetaData().getContentSize());
-            }
-            return StoreFuture.IMMEDIATE_FUTURE;
-        }
-
-        @Override
-        public void remove()
-        {
-            checkMessageStoreOpen();
-
-            int delta = getMetaData().getContentSize();
-            BDBMessageStore.this.removeMessage(_messageId, false);
-            storedSizeChangeOccured(-delta);
-        }
-
-        private boolean stored()
-        {
-            return _metaData == null || _isRecovered;
-        }
-    }
-
-    private class BDBTransaction implements org.apache.qpid.server.store.Transaction
-    {
-        private com.sleepycat.je.Transaction _txn;
-        private int _storeSizeIncrease;
-
-        private BDBTransaction() throws StoreException
-        {
-            _txn = _environmentFacade.beginTransaction();
-        }
-
-        @Override
-        public void enqueueMessage(TransactionLogResource queue, EnqueueableMessage message) throws StoreException
-        {
-            checkMessageStoreOpen();
-
-            if(message.getStoredMessage() instanceof StoredBDBMessage)
-            {
-                final StoredBDBMessage storedMessage = (StoredBDBMessage) message.getStoredMessage();
-                storedMessage.store(_txn);
-                _storeSizeIncrease += storedMessage.getMetaData().getContentSize();
-            }
-
-            BDBMessageStore.this.enqueueMessage(_txn, queue, message.getMessageNumber());
-        }
-
-        @Override
-        public void dequeueMessage(TransactionLogResource queue, EnqueueableMessage message) throws StoreException
-        {
-            checkMessageStoreOpen();
-
-            BDBMessageStore.this.dequeueMessage(_txn, queue, message.getMessageNumber());
-        }
-
-        @Override
-        public void commitTran() throws StoreException
-        {
-            checkMessageStoreOpen();
-
-            BDBMessageStore.this.commitTranImpl(_txn, true);
-            BDBMessageStore.this.storedSizeChangeOccured(_storeSizeIncrease);
-        }
-
-        @Override
-        public StoreFuture commitTranAsync() throws StoreException
-        {
-            checkMessageStoreOpen();
-
-            BDBMessageStore.this.storedSizeChangeOccured(_storeSizeIncrease);
-            return BDBMessageStore.this.commitTranImpl(_txn, false);
-        }
-
-        @Override
-        public void abortTran() throws StoreException
-        {
-            checkMessageStoreOpen();
-
-            BDBMessageStore.this.abortTran(_txn);
-        }
-
-        @Override
-        public void removeXid(long format, byte[] globalId, byte[] branchId) throws StoreException
-        {
-            checkMessageStoreOpen();
-
-            BDBMessageStore.this.removeXid(_txn, format, globalId, branchId);
-        }
-
-        @Override
-        public void recordXid(long format, byte[] globalId, byte[] branchId, Record[] enqueues,
-                              Record[] dequeues) throws StoreException
-        {
-            checkMessageStoreOpen();
-
-            BDBMessageStore.this.recordXid(_txn, format, globalId, branchId, enqueues, dequeues);
-        }
-    }
-
-    private void storedSizeChangeOccured(final int delta) throws StoreException
-    {
-        try
-        {
-            storedSizeChange(delta);
-        }
-        catch(DatabaseException e)
-        {
-            throw _environmentFacade.handleDatabaseException("Stored size change exception", e);
-        }
-    }
-
-    private void storedSizeChange(final int delta)
-    {
-        if(getPersistentSizeHighThreshold() > 0)
-        {
-            synchronized (this)
-            {
-                // the delta supplied is an approximation of a store size change. we don;t want to check the statistic every
-                // time, so we do so only when there's been enough change that it is worth looking again. We do this by
-                // assuming the total size will change by less than twice the amount of the message data change.
-                long newSize = _totalStoreSize += 2*delta;
-
-                if(!_limitBusted &&  newSize > getPersistentSizeHighThreshold())
-                {
-                    _totalStoreSize = getSizeOnDisk();
-
-                    if(_totalStoreSize > getPersistentSizeHighThreshold())
-                    {
-                        _limitBusted = true;
-                        _eventManager.notifyEvent(Event.PERSISTENT_MESSAGE_SIZE_OVERFULL);
-                    }
-                }
-                else if(_limitBusted && newSize < getPersistentSizeLowThreshold())
-                {
-                    long oldSize = _totalStoreSize;
-                    _totalStoreSize = getSizeOnDisk();
-
-                    if(oldSize <= _totalStoreSize)
-                    {
-
-                        reduceSizeOnDisk();
-
-                        _totalStoreSize = getSizeOnDisk();
-
-                    }
-
-                    if(_totalStoreSize < getPersistentSizeLowThreshold())
-                    {
-                        _limitBusted = false;
-                        _eventManager.notifyEvent(Event.PERSISTENT_MESSAGE_SIZE_UNDERFULL);
-                    }
-
-
-                }
-            }
-        }
-    }
-
-    private void checkConfigurationStoreOpen()
-    {
-        if (!_configurationStoreOpen.get())
-        {
-            throw new IllegalStateException("Configuration store is not open");
-        }
-    }
-
-    private void checkMessageStoreOpen()
-    {
-        if (!_messageStoreOpen.get())
-        {
-            throw new IllegalStateException("Message store is not open");
-        }
-    }
-
-    private void reduceSizeOnDisk()
-    {
-        _environmentFacade.getEnvironment().getConfig().setConfigParam(EnvironmentConfig.ENV_RUN_CLEANER, "false");
-        boolean cleaned = false;
-        while (_environmentFacade.getEnvironment().cleanLog() > 0)
-        {
-            cleaned = true;
-        }
-        if (cleaned)
-        {
-            CheckpointConfig force = new CheckpointConfig();
-            force.setForce(true);
-            _environmentFacade.getEnvironment().checkpoint(force);
-        }
-
-
-        _environmentFacade.getEnvironment().getConfig().setConfigParam(EnvironmentConfig.ENV_RUN_CLEANER, "true");
-    }
-
-    private long getSizeOnDisk()
-    {
-        return _environmentFacade.getEnvironment().getStats(null).getTotalLogSize();
-    }
-
-    private long getPersistentSizeLowThreshold()
-    {
-        return _persistentSizeLowThreshold;
-    }
-
-    private long getPersistentSizeHighThreshold()
-    {
-        return _persistentSizeHighThreshold;
-    }
-
-
-    @Override
-    public void onDelete()
-    {
-        if (!_configurationStoreOpen.get() && !_messageStoreOpen.get())
-        {
-            if (_storeLocation != null)
-            {
-                if (LOGGER.isDebugEnabled())
-                {
-                    LOGGER.debug("Deleting store " + _storeLocation);
-                }
-
-                File location = new File(_storeLocation);
-                if (location.exists())
-                {
-                    if (!FileUtils.delete(location, true))
-                    {
-                        LOGGER.error("Cannot delete " + _storeLocation);
-                    }
-                }
-            }
-        }
-    }
-
-    private Database getConfiguredObjectsDb()
-    {
-        return _environmentFacade.getOpenDatabase(CONFIGURED_OBJECTS_DB_NAME);
-    }
-
-    private Database getConfiguredObjectHierarchyDb()
-    {
-        return _environmentFacade.getOpenDatabase(CONFIGURED_OBJECT_HIERARCHY_DB_NAME);
-    }
-
-    private Database getMessageContentDb()
-    {
-        return _environmentFacade.getOpenDatabase(MESSAGE_CONTENT_DB_NAME);
-    }
-
-    private Database getMessageMetaDataDb()
-    {
-        return _environmentFacade.getOpenDatabase(MESSAGE_META_DATA_DB_NAME);
-    }
-
-    private Database getDeliveryDb()
-    {
-        return _environmentFacade.getOpenDatabase(DELIVERY_DB_NAME);
-    }
-
-    private Database getXidDb()
-    {
-        return _environmentFacade.getOpenDatabase(XID_DB_NAME);
-    }
-
-    class UpgradeTask implements EnvironmentFacadeTask
-    {
-        private final ConfiguredObject<?> _parent;
-
-        public UpgradeTask(ConfiguredObject<?> parent)
-        {
-            _parent = parent;
-        }
-
-        @Override
-        public void execute(EnvironmentFacade facade)
-        {
-            try
-            {
-                new Upgrader(facade.getEnvironment(), _parent).upgradeIfNecessary();
-            }
-            catch(DatabaseException e)
-            {
-                throw facade.handleDatabaseException("Cannot upgrade store", e);
-            }
-        }
-    }
-
-    class OpenDatabasesTask implements EnvironmentFacadeTask
-    {
-        private String[] _names;
-
-        public OpenDatabasesTask(String[] names)
-        {
-            _names = names;
-        }
-
-        @Override
-        public void execute(EnvironmentFacade facade)
-        {
-            try
-            {
-                DatabaseConfig dbConfig = new DatabaseConfig();
-                dbConfig.setTransactional(true);
-                dbConfig.setAllowCreate(true);
-                facade.openDatabases(dbConfig, _names);
-            }
-            catch(DatabaseException e)
-            {
-                throw facade.handleDatabaseException("Cannot open databases", e);
-            }
-        }
-
-    }
-
-    class DiskSpaceTask implements EnvironmentFacadeTask
-    {
-
-        @Override
-        public void execute(EnvironmentFacade facade)
-        {
-            try
-            {
-                _totalStoreSize = facade.getEnvironment().getStats(null).getTotalLogSize();
-            }
-            catch(DatabaseException e)
-            {
-                throw facade.handleDatabaseException("Cannot evaluate disk store size", e);
-            }
-        }
-
-    }
-
-    public class MaxMessageIdTask implements EnvironmentFacadeTask, MessageHandler
-    {
-        private long _maxId;
-
-        @Override
-        public void execute(EnvironmentFacade facade)
-        {
-            visitMessagesInternal(this, facade);
-            _messageId.set(_maxId);
-        }
-
-        @Override
-        public boolean handle(StoredMessage<?> storedMessage)
-        {
-            long id = storedMessage.getMessageNumber();
-            if (_maxId<id)
-            {
-                _maxId = id;
-            }
-            return true;
-        }
-
-    }
-
-    @Override
-    public void visitMessages(MessageHandler handler) throws StoreException
-    {
-        checkMessageStoreOpen();
-        visitMessagesInternal(handler, _environmentFacade);
-    }
-
-    private void visitMessagesInternal(MessageHandler handler, EnvironmentFacade environmentFacade)
-    {
-        Cursor cursor = null;
-        try
-        {
-            cursor = environmentFacade.getOpenDatabase(MESSAGE_META_DATA_DB_NAME).openCursor(null, null);
-            DatabaseEntry key = new DatabaseEntry();
-            DatabaseEntry value = new DatabaseEntry();
-            MessageMetaDataBinding valueBinding = MessageMetaDataBinding.getInstance();
-
-            while (cursor.getNext(key, value, LockMode.RMW) == OperationStatus.SUCCESS)
-            {
-                long messageId = LongBinding.entryToLong(key);
-                StorableMessageMetaData metaData = valueBinding.entryToObject(value);
-                StoredBDBMessage message = new StoredBDBMessage(messageId, metaData, true);
-
-                if (!handler.handle(message))
-                {
-                    break;
-                }
-            }
-        }
-        catch (DatabaseException e)
-        {
-            throw environmentFacade.handleDatabaseException("Cannot recover messages", e);
-        }
-        finally
-        {
-            if (cursor != null)
-            {
-                try
-                {
-                    cursor.close();
-                }
-                catch(DatabaseException e)
-                {
-                    throw environmentFacade.handleDatabaseException("Cannot close cursor", e);
-                }
-            }
-        }
-    }
-
-    @Override
-    public void visitMessageInstances(MessageInstanceHandler handler) throws StoreException
-    {
-        checkMessageStoreOpen();
-
-        Cursor cursor = null;
-        List<QueueEntryKey> entries = new ArrayList<QueueEntryKey>();
-        try
-        {
-            cursor = getDeliveryDb().openCursor(null, null);
-            DatabaseEntry key = new DatabaseEntry();
-            QueueEntryBinding keyBinding = QueueEntryBinding.getInstance();
-
-            DatabaseEntry value = new DatabaseEntry();
-            while (cursor.getNext(key, value, LockMode.DEFAULT) == OperationStatus.SUCCESS)
-            {
-                QueueEntryKey entry = keyBinding.entryToObject(key);
-                entries.add(entry);
-            }
-        }
-        catch (DatabaseException e)
-        {
-            throw _environmentFacade.handleDatabaseException("Cannot visit message instances", e);
-        }
-        finally
-        {
-            closeCursorSafely(cursor);
-        }
-
-        for(QueueEntryKey entry : entries)
-        {
-            UUID queueId = entry.getQueueId();
-            long messageId = entry.getMessageId();
-            if (!handler.handle(queueId, messageId))
-            {
-                break;
-            }
-        }
-
-    }
-
-    @Override
-    public void visitDistributedTransactions(DistributedTransactionHandler handler) throws StoreException
-    {
-        checkMessageStoreOpen();
-
-        Cursor cursor = null;
-        try
-        {
-            cursor = getXidDb().openCursor(null, null);
-            DatabaseEntry key = new DatabaseEntry();
-            XidBinding keyBinding = XidBinding.getInstance();
-            PreparedTransactionBinding valueBinding = new PreparedTransactionBinding();
-            DatabaseEntry value = new DatabaseEntry();
-
-            while (cursor.getNext(key, value, LockMode.RMW) == OperationStatus.SUCCESS)
-            {
-                Xid xid = keyBinding.entryToObject(key);
-                PreparedTransaction preparedTransaction = valueBinding.entryToObject(value);
-                if (!handler.handle(xid.getFormat(),xid.getGlobalId(),xid.getBranchId(),
-                                preparedTransaction.getEnqueues(),preparedTransaction.getDequeues()))
-                {
-                    break;
-                }
-            }
-
-        }
-        catch (DatabaseException e)
-        {
-            throw _environmentFacade.handleDatabaseException("Cannot recover distributed transactions", e);
-        }
-        finally
-        {
-            closeCursorSafely(cursor);
-        }
-    }
-}
diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStoreFactory.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStoreFactory.java
index 2d783fa181..067372a228 100644
--- a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStoreFactory.java
+++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStoreFactory.java
@@ -42,13 +42,13 @@ public class BDBMessageStoreFactory implements MessageStoreFactory, DurableConfi
     @Override
     public DurableConfigurationStore createDurableConfigurationStore()
     {
-        return new BDBMessageStore();
+        return new BDBConfigurationStore();
     }
 
     @Override
     public MessageStore createMessageStore()
     {
-        return new BDBMessageStore();
+        return (new BDBConfigurationStore()).getMessageStore();
     }
 
     @Override
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 e80d60609f..1be7e97a47 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
@@ -27,9 +27,8 @@ import java.lang.reflect.InvocationTargetException;
 
 import org.apache.log4j.Logger;
 import org.apache.qpid.server.model.ConfiguredObject;
-import org.apache.qpid.server.model.VirtualHost;
 import org.apache.qpid.server.store.StoreException;
-import org.apache.qpid.server.store.berkeleydb.BDBMessageStore;
+import org.apache.qpid.server.store.berkeleydb.BDBConfigurationStore;
 
 import com.sleepycat.bind.tuple.IntegerBinding;
 import com.sleepycat.bind.tuple.LongBinding;
@@ -70,7 +69,7 @@ public class Upgrader
             if(versionDb.count() == 0L)
             {
 
-                int sourceVersion = isEmpty ? BDBMessageStore.VERSION: identifyOldStoreVersion();
+                int sourceVersion = isEmpty ? BDBConfigurationStore.VERSION: identifyOldStoreVersion();
                 DatabaseEntry key = new DatabaseEntry();
                 IntegerBinding.intToEntry(sourceVersion, key);
                 DatabaseEntry value = new DatabaseEntry();
@@ -86,11 +85,11 @@ public class Upgrader
                 LOGGER.debug("Source message store version is " + version);
             }
 
-            if(version > BDBMessageStore.VERSION)
+            if(version > BDBConfigurationStore.VERSION)
             {
                 throw new StoreException("Database version " + version
                                             + " is higher than the most recent known version: "
-                                            + BDBMessageStore.VERSION);
+                                            + BDBConfigurationStore.VERSION);
             }
             performUpgradeFromVersion(version, versionDb);
         }
@@ -139,7 +138,7 @@ public class Upgrader
     void performUpgradeFromVersion(int sourceVersion, Database versionDb)
             throws StoreException
     {
-        while(sourceVersion != BDBMessageStore.VERSION)
+        while(sourceVersion != BDBConfigurationStore.VERSION)
         {
             upgrade(sourceVersion, ++sourceVersion);
             DatabaseEntry key = new DatabaseEntry();
@@ -191,7 +190,7 @@ public class Upgrader
 
     private int identifyOldStoreVersion() throws DatabaseException
     {
-        int version = BDBMessageStore.VERSION;
+        int version = BDBConfigurationStore.VERSION;
         for (String databaseName : _environment.getDatabaseNames())
         {
             if (databaseName.contains("_v"))
diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAVirtualHostImpl.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAVirtualHostImpl.java
index 974ff51a3b..af82d9fd2d 100644
--- a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAVirtualHostImpl.java
+++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAVirtualHostImpl.java
@@ -31,7 +31,7 @@ import org.apache.qpid.server.model.ManagedObjectFactoryConstructor;
 import org.apache.qpid.server.model.VirtualHostNode;
 import org.apache.qpid.server.store.DurableConfigurationStore;
 import org.apache.qpid.server.store.MessageStore;
-import org.apache.qpid.server.store.berkeleydb.BDBMessageStore;
+import org.apache.qpid.server.store.berkeleydb.BDBConfigurationStore;
 import org.apache.qpid.server.store.berkeleydb.replication.ReplicatedEnvironmentFacade;
 import org.apache.qpid.server.virtualhost.AbstractVirtualHost;
 
@@ -42,7 +42,7 @@ public class BDBHAVirtualHostImpl extends AbstractVirtualHost<BDBHAVirtualHostIm
 {
     public static final String TYPE = "BDB_HA";
 
-    private final BDBMessageStore _messageStore;
+    private final BDBConfigurationStore _configurationStore;
     private MessageStoreLogSubject _messageStoreLogSubject;
 
     @ManagedAttributeField(afterSet="setLocalTransactionSynchronizationPolicyOnEnvironment")
@@ -56,8 +56,8 @@ public class BDBHAVirtualHostImpl extends AbstractVirtualHost<BDBHAVirtualHostIm
     {
         super(attributes, virtualHostNode);
 
-        _messageStore = (BDBMessageStore) virtualHostNode.getConfigurationStore();
-        _messageStoreLogSubject = new MessageStoreLogSubject(getName(), _messageStore.getClass().getSimpleName());
+        _configurationStore = (BDBConfigurationStore) virtualHostNode.getConfigurationStore();
+        _messageStoreLogSubject = new MessageStoreLogSubject(getName(), _configurationStore.getMessageStore().getClass().getSimpleName());
     }
 
     @Override
@@ -68,13 +68,13 @@ public class BDBHAVirtualHostImpl extends AbstractVirtualHost<BDBHAVirtualHostIm
     @Override
     public DurableConfigurationStore getDurableConfigurationStore()
     {
-        return _messageStore;
+        return _configurationStore;
     }
 
     @Override
     public MessageStore getMessageStore()
     {
-        return _messageStore;
+        return _configurationStore.getMessageStore();
     }
 
     @Override
@@ -176,7 +176,7 @@ public class BDBHAVirtualHostImpl extends AbstractVirtualHost<BDBHAVirtualHostIm
 
     private ReplicatedEnvironmentFacade getReplicatedEnvironmentFacade()
     {
-        return (ReplicatedEnvironmentFacade)_messageStore.getEnvironmentFacade();
+        return (ReplicatedEnvironmentFacade) _configurationStore.getEnvironmentFacade();
     }
 
 }
diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/virtualhostnode/berkeleydb/BDBHAVirtualHostNodeImpl.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/virtualhostnode/berkeleydb/BDBHAVirtualHostNodeImpl.java
index 91ac63ff46..e36def581b 100644
--- a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/virtualhostnode/berkeleydb/BDBHAVirtualHostNodeImpl.java
+++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/virtualhostnode/berkeleydb/BDBHAVirtualHostNodeImpl.java
@@ -42,7 +42,6 @@ import com.sleepycat.je.rep.StateChangeEvent;
 import com.sleepycat.je.rep.StateChangeListener;
 import com.sleepycat.je.rep.util.ReplicationGroupAdmin;
 import com.sleepycat.je.rep.utilint.HostPortPair;
-
 import org.apache.log4j.Logger;
 
 import org.apache.qpid.server.logging.messages.ConfigStoreMessages;
@@ -59,7 +58,7 @@ import org.apache.qpid.server.model.VirtualHost;
 import org.apache.qpid.server.security.SecurityManager;
 import org.apache.qpid.server.store.DurableConfigurationStore;
 import org.apache.qpid.server.store.VirtualHostStoreUpgraderAndRecoverer;
-import org.apache.qpid.server.store.berkeleydb.BDBMessageStore;
+import org.apache.qpid.server.store.berkeleydb.BDBConfigurationStore;
 import org.apache.qpid.server.store.berkeleydb.replication.ReplicatedEnvironmentFacade;
 import org.apache.qpid.server.store.berkeleydb.replication.ReplicatedEnvironmentFacadeFactory;
 import org.apache.qpid.server.store.berkeleydb.replication.ReplicationGroupListener;
@@ -265,9 +264,9 @@ public class BDBHAVirtualHostNodeImpl extends AbstractVirtualHostNode<BDBHAVirtu
     }
 
     @Override
-    public BDBMessageStore getConfigurationStore()
+    public BDBConfigurationStore getConfigurationStore()
     {
-        return (BDBMessageStore) super.getConfigurationStore();
+        return (BDBConfigurationStore) super.getConfigurationStore();
     }
 
     protected ReplicatedEnvironmentFacade getReplicatedEnvironmentFacade()
@@ -278,7 +277,7 @@ public class BDBHAVirtualHostNodeImpl extends AbstractVirtualHostNode<BDBHAVirtu
     @Override
     protected DurableConfigurationStore createConfigurationStore()
     {
-        return new BDBMessageStore(new ReplicatedEnvironmentFacadeFactory());
+        return new BDBConfigurationStore(new ReplicatedEnvironmentFacadeFactory());
     }
 
     @Override
diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHostNodeTest.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHostNodeTest.java
index 962e19f81c..0254cbc909 100644
--- a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHostNodeTest.java
+++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHostNodeTest.java
@@ -165,8 +165,8 @@ public class BDBHAVirtualHostNodeTest extends QpidTestCase
         DurableConfigurationStore store = node.getConfigurationStore();
         assertNotNull(store);
 
-        BDBMessageStore bdbMessageStore = (BDBMessageStore) store;
-        ReplicatedEnvironment environment = (ReplicatedEnvironment) bdbMessageStore.getEnvironmentFacade().getEnvironment();
+        BDBConfigurationStore bdbConfigurationStore = (BDBConfigurationStore) store;
+        ReplicatedEnvironment environment = (ReplicatedEnvironment) bdbConfigurationStore.getEnvironmentFacade().getEnvironment();
         ReplicationConfig replicationConfig = environment.getRepConfig();
 
         assertEquals(nodeName, environment.getNodeName());
@@ -181,7 +181,7 @@ public class BDBHAVirtualHostNodeTest extends QpidTestCase
         VirtualHost<?, ?, ?> virtualHost = node.getVirtualHost();
         assertNotNull("Virtual host child was not added", virtualHost);
         assertEquals("Unexpected virtual host name", groupName, virtualHost.getName());
-        assertEquals("Unexpected virtual host store", store, virtualHost.getMessageStore());
+        assertEquals("Unexpected virtual host store", bdbConfigurationStore.getMessageStore(), virtualHost.getMessageStore());
         assertEquals("Unexpected virtual host state", State.ACTIVE, virtualHost.getState());
 
         node.stop();
@@ -212,8 +212,8 @@ public class BDBHAVirtualHostNodeTest extends QpidTestCase
 
         BDBHAVirtualHostNode<?> node = createAndStartHaVHN(attributes);
 
-        BDBMessageStore bdbMessageStore = (BDBMessageStore) node.getConfigurationStore();
-        ReplicatedEnvironment environment = (ReplicatedEnvironment) bdbMessageStore.getEnvironmentFacade().getEnvironment();
+        BDBConfigurationStore bdbConfigurationStore = (BDBConfigurationStore) node.getConfigurationStore();
+        ReplicatedEnvironment environment = (ReplicatedEnvironment) bdbConfigurationStore.getEnvironmentFacade().getEnvironment();
 
         assertEquals("Unexpected node priority value before mutation", 1, environment.getRepMutableConfig().getNodePriority());
         assertFalse("Unexpected designated primary value before mutation", environment.getRepMutableConfig().getDesignatedPrimary());
diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStoreConfigurationTest.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStoreConfigurationTest.java
index e1678e6f65..8657a3a0b1 100644
--- a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStoreConfigurationTest.java
+++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStoreConfigurationTest.java
@@ -28,6 +28,6 @@ public class BDBMessageStoreConfigurationTest extends AbstractDurableConfigurati
     @Override
     protected DurableConfigurationStore createConfigStore() throws Exception
     {
-        return new BDBMessageStore();
+        return new BDBConfigurationStore();
     }
 }
diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStoreQuotaEventsTest.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStoreQuotaEventsTest.java
index f2de01445d..2afdaa4dd5 100644
--- a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStoreQuotaEventsTest.java
+++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBMessageStoreQuotaEventsTest.java
@@ -74,7 +74,7 @@ public class BDBMessageStoreQuotaEventsTest extends MessageStoreQuotaEventsTestB
     @Override
     protected MessageStore createStore() throws Exception
     {
-        MessageStore store = new BDBMessageStore();
+        MessageStore store = (new BDBConfigurationStore()).getMessageStore();
         return store;
     }
 }
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 6fba1b215e..d623da846c 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
@@ -50,6 +50,7 @@ import org.apache.qpid.transport.MessageDeliveryPriority;
 import org.apache.qpid.transport.MessageProperties;
 import org.apache.qpid.transport.MessageTransfer;
 import org.apache.qpid.util.FileUtils;
+import org.apache.qpid.server.store.berkeleydb.BDBConfigurationStore.BDBMessageStore;
 
 /**
  * Subclass of MessageStoreTestCase which runs the standard tests from the superclass against
@@ -81,7 +82,7 @@ public class BDBMessageStoreTest extends MessageStoreTestCase
      */
     public void testBDBMessagePersistence() throws Exception
     {
-        BDBMessageStore bdbStore = (BDBMessageStore)getStore();
+        MessageStore bdbStore = getStore();
 
         // Create content ByteBuffers.
         // Split the content into 2 chunks for the 0-8 message, as per broker behaviour.
@@ -342,7 +343,7 @@ public class BDBMessageStoreTest extends MessageStoreTestCase
      */
     public void testMessageCreationAndRemoval() throws Exception
     {
-        BDBMessageStore bdbStore = (BDBMessageStore)getStore();
+        BDBMessageStore bdbStore = (BDBMessageStore) getStore();
 
         StoredMessage<MessageMetaData> storedMessage_0_8 = createAndStoreSingleChunkMessage_0_8(bdbStore);
         long messageid_0_8 = storedMessage_0_8.getMessageNumber();
@@ -430,7 +431,7 @@ public class BDBMessageStoreTest extends MessageStoreTestCase
     @Override
     protected MessageStore createMessageStore()
     {
-        return new BDBMessageStore();
+        return new BDBConfigurationStore().getMessageStore();
     }
 
 }
diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgraderFailOnNewerVersionTest.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgraderFailOnNewerVersionTest.java
index c407be50c6..5b869b67dc 100644
--- a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgraderFailOnNewerVersionTest.java
+++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/upgrade/UpgraderFailOnNewerVersionTest.java
@@ -26,7 +26,8 @@ import com.sleepycat.je.Database;
 import com.sleepycat.je.DatabaseConfig;
 import com.sleepycat.je.DatabaseEntry;
 import com.sleepycat.je.OperationStatus;
-import org.apache.qpid.server.store.berkeleydb.BDBMessageStore;
+
+import org.apache.qpid.server.store.berkeleydb.BDBConfigurationStore;
 import org.apache.qpid.server.util.ServerScopedRuntimeException;
 
 public class UpgraderFailOnNewerVersionTest extends AbstractUpgradeTestCase
@@ -94,7 +95,7 @@ public class UpgraderFailOnNewerVersionTest extends AbstractUpgradeTestCase
         catch(ServerScopedRuntimeException ex)
         {
             assertEquals("Incorrect exception thrown", "Database version 999 is higher than the most recent known version: "
-                                                        + BDBMessageStore.VERSION, ex.getMessage());
+                                                        + BDBConfigurationStore.VERSION, ex.getMessage());
         }
     }
 
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 4b9a8d19a8..54dd08fd98 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.BDBConfigurationStore;
 import org.apache.qpid.server.store.berkeleydb.tuple.ContentBinding;
 
 import com.sleepycat.bind.tuple.IntegerBinding;
@@ -95,7 +95,7 @@ public class UpgraderTest extends AbstractUpgradeTestCase
     {
         assertEquals("Unexpected store version", -1, getStoreVersion(_environment));
         _upgrader.upgradeIfNecessary();
-        assertEquals("Unexpected store version", BDBMessageStore.VERSION, getStoreVersion(_environment));
+        assertEquals("Unexpected store version", BDBConfigurationStore.VERSION, getStoreVersion(_environment));
         assertContent();
     }
 
@@ -115,7 +115,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(emptyEnvironment));
+            assertEquals("Unexpected store version", BDBConfigurationStore.VERSION, getStoreVersion(emptyEnvironment));
 
         }
         finally