QPID-5410: [Java Broker/BDB]. Introduce a thin facade (EnvironmentFacade) between the BDBMessage and BDB JE's Environment/ReplicatedEnvironment. The motivation behind this facade is principally HA; there are a number of cases where JE requires the ReplicatedEnvironment is recreated. The facade layer allows for this to be done transparently from the upper tiers (the BDBMessageStore). The facade has two implementations StandardFacade used in the non-HA use case, and ReplicatedEnvironmentFacade in the HA case.

Key changes:

* BDBHAVirtualHost is now responsible for the creation of ReplicatedEnvironmentFacade
* BDBMessageStore reverts to a single implementation without knowledge of HA.
* BDBMessageStore now interacts with JE via the facade.
* BDBHAMessageStoreManagerMBean interrogates the facade
* ReplicatedEnvironmentFacade monitors the group for changes in state (nodes becoming uncontactable etc), if such a state change is detected, the DatabasePinger
fires a single transaction to determine if quorum still exists.  If quorum does not exist, the environment is restarted, thus transition the environment into
 the UNKNOWN state.

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/java-broker-bdb-ha2@1576697 13f79535-47bb-0310-9956-ffa450edef68

3 files changed, 56 insertions, 62 deletions

diff --git a/qpid/java/bdbstore/jmx/src/main/java/org/apache/qpid/server/store/berkeleydb/jmx/BDBHAMessageStoreManagerMBean.java b/qpid/java/bdbstore/jmx/src/main/java/org/apache/qpid/server/store/berkeleydb/jmx/BDBHAMessageStoreManagerMBean.java
index aa4ddd8181..f36c1ecc6f 100644
--- a/qpid/java/bdbstore/jmx/src/main/java/org/apache/qpid/server/store/berkeleydb/jmx/BDBHAMessageStoreManagerMBean.java
+++ b/qpid/java/bdbstore/jmx/src/main/java/org/apache/qpid/server/store/berkeleydb/jmx/BDBHAMessageStoreManagerMBean.java
@@ -36,19 +36,12 @@ import javax.management.openmbean.TabularDataSupport;
 import javax.management.openmbean.TabularType;
 
 import org.apache.log4j.Logger;
-import org.apache.qpid.server.store.StoreException;
 import org.apache.qpid.server.jmx.AMQManagedObject;
 import org.apache.qpid.server.jmx.ManagedObject;
-import org.apache.qpid.server.store.berkeleydb.BDBHAMessageStore;
+import org.apache.qpid.server.store.berkeleydb.replication.ReplicatedEnvironmentFacade;
 
 /**
  * Management mbean for BDB HA.
- * <p>
- * At runtime, the classloader loading this clas must have visibility of the other Qpid JMX classes. This is
- * currently arranged through OSGI using the <b>fragment</b> feature so that this bundle shares the
- * same classloader as broker-plugins-management-jmx.  See the <b>Fragment-Host:</b> header within the MANIFEST.MF
- * of this bundle.
- * </p>
  */
 public class BDBHAMessageStoreManagerMBean extends AMQManagedObject implements ManagedBDBHAMessageStore
 {
@@ -63,7 +56,7 @@ public class BDBHAMessageStoreManagerMBean extends AMQManagedObject implements M
         try
         {
             GROUP_MEMBER_ATTRIBUTE_TYPES = new OpenType<?>[] {SimpleType.STRING, SimpleType.STRING};
-            final String[] itemNames = new String[] {BDBHAMessageStore.GRP_MEM_COL_NODE_NAME, BDBHAMessageStore.GRP_MEM_COL_NODE_HOST_PORT};
+            final String[] itemNames = new String[] {ReplicatedEnvironmentFacade.GRP_MEM_COL_NODE_NAME, ReplicatedEnvironmentFacade.GRP_MEM_COL_NODE_HOST_PORT};
             final String[] itemDescriptions = new String[] {"Unique node name", "Node host / port "};
             GROUP_MEMBER_ROW = new CompositeType("GroupMember", "Replication group member",
                                                 itemNames,
@@ -71,7 +64,7 @@ public class BDBHAMessageStoreManagerMBean extends AMQManagedObject implements M
                                                 GROUP_MEMBER_ATTRIBUTE_TYPES );
             GROUP_MEMBERS_TABLE = new TabularType("GroupMembers", "Replication group memebers",
                                                 GROUP_MEMBER_ROW,
-                                                new String[] {BDBHAMessageStore.GRP_MEM_COL_NODE_NAME});
+                                                new String[] {ReplicatedEnvironmentFacade.GRP_MEM_COL_NODE_NAME});
         }
         catch (final OpenDataException ode)
         {
@@ -79,44 +72,46 @@ public class BDBHAMessageStoreManagerMBean extends AMQManagedObject implements M
         }
     }
 
-    private final BDBHAMessageStore _store;
+    private final ReplicatedEnvironmentFacade _replicatedEnvironmentFacade;
+    private final String _objectName;
 
-    protected BDBHAMessageStoreManagerMBean(BDBHAMessageStore store, ManagedObject parent) throws JMException
+    protected BDBHAMessageStoreManagerMBean(String virtualHostName, ReplicatedEnvironmentFacade replicatedEnvironmentFacade, ManagedObject parent) throws JMException
     {
         super(ManagedBDBHAMessageStore.class, ManagedBDBHAMessageStore.TYPE, ((AMQManagedObject)parent).getRegistry());
-        LOGGER.debug("Creating BDBHAMessageStoreManagerMBean");
-        _store = store;
+        LOGGER.debug("Creating BDBHAMessageStoreManagerMBean for " + virtualHostName);
+        _replicatedEnvironmentFacade = replicatedEnvironmentFacade;
+        _objectName = ObjectName.quote(virtualHostName);
         register();
     }
 
     @Override
     public String getObjectInstanceName()
     {
-        return ObjectName.quote(_store.getName());
+        return _objectName;
     }
 
     @Override
     public String getGroupName()
     {
-        return _store.getGroupName();
+        return _replicatedEnvironmentFacade.getGroupName();
     }
 
     @Override
     public String getNodeName()
     {
-        return _store.getNodeName();
+        return _replicatedEnvironmentFacade.getNodeName();
     }
 
     @Override
     public String getNodeHostPort()
     {
-        return _store.getNodeHostPort();
+        return _replicatedEnvironmentFacade.getHostPort();
     }
 
     @Override
     public String getHelperHostPort()
     {
-        return _store.getHelperHostPort();
+        return _replicatedEnvironmentFacade.getHelperHostPort();
     }
 
     @Override
@@ -124,7 +119,7 @@ public class BDBHAMessageStoreManagerMBean extends AMQManagedObject implements M
     {
         try
         {
-            return _store.getDurability();
+            return _replicatedEnvironmentFacade.getDurability();
         }
         catch (RuntimeException e)
         {
@@ -137,7 +132,7 @@ public class BDBHAMessageStoreManagerMBean extends AMQManagedObject implements M
     @Override
     public boolean getCoalescingSync() throws IOException, JMException
     {
-        return _store.isCoalescingSync();
+        return _replicatedEnvironmentFacade.isCoalescingSync();
     }
 
     @Override
@@ -145,7 +140,7 @@ public class BDBHAMessageStoreManagerMBean extends AMQManagedObject implements M
     {
         try
         {
-            return _store.getNodeState();
+            return _replicatedEnvironmentFacade.getNodeState();
         }
         catch (RuntimeException e)
         {
@@ -159,7 +154,7 @@ public class BDBHAMessageStoreManagerMBean extends AMQManagedObject implements M
     {
         try
         {
-            return _store.isDesignatedPrimary();
+            return _replicatedEnvironmentFacade.isDesignatedPrimary();
         }
         catch (RuntimeException e)
         {
@@ -172,7 +167,7 @@ public class BDBHAMessageStoreManagerMBean extends AMQManagedObject implements M
     public TabularData getAllNodesInGroup() throws IOException, JMException
     {
         final TabularDataSupport data = new TabularDataSupport(GROUP_MEMBERS_TABLE);
-        final List<Map<String, String>> members = _store.getGroupMembers();
+        final List<Map<String, String>> members = _replicatedEnvironmentFacade.getGroupMembers();
 
         for (Map<String, String> map : members)
         {
@@ -187,9 +182,9 @@ public class BDBHAMessageStoreManagerMBean extends AMQManagedObject implements M
     {
         try
         {
-            _store.removeNodeFromGroup(nodeName);
+            _replicatedEnvironmentFacade.removeNodeFromGroup(nodeName);
         }
-        catch (StoreException e)
+        catch (RuntimeException e)
         {
             LOGGER.error("Failed to remove node " + nodeName + " from group", e);
             throw new JMException(e.getMessage());
@@ -201,11 +196,11 @@ public class BDBHAMessageStoreManagerMBean extends AMQManagedObject implements M
     {
         try
         {
-            _store.setDesignatedPrimary(primary);
+            _replicatedEnvironmentFacade.setDesignatedPrimary(primary);
         }
-        catch (StoreException e)
+        catch (RuntimeException e)
         {
-            LOGGER.error("Failed to set node " + _store.getNodeName() + " as designated primary", e);
+            LOGGER.error("Failed to set node " + _replicatedEnvironmentFacade.getNodeName() + " as designated primary", e);
             throw new JMException(e.getMessage());
         }
     }
@@ -215,9 +210,9 @@ public class BDBHAMessageStoreManagerMBean extends AMQManagedObject implements M
     {
         try
         {
-            _store.updateAddress(nodeName, newHostName, newPort);
+            _replicatedEnvironmentFacade.updateAddress(nodeName, newHostName, newPort);
         }
-        catch(StoreException e)
+        catch(RuntimeException e)
         {
             LOGGER.error("Failed to update address for node " + nodeName + " to " + newHostName + ":" + newPort, e);
             throw new JMException(e.getMessage());
diff --git a/qpid/java/bdbstore/jmx/src/main/java/org/apache/qpid/server/store/berkeleydb/jmx/BDBHAMessageStoreManagerMBeanProvider.java b/qpid/java/bdbstore/jmx/src/main/java/org/apache/qpid/server/store/berkeleydb/jmx/BDBHAMessageStoreManagerMBeanProvider.java
index 0492350a25..16199d30a3 100644
--- a/qpid/java/bdbstore/jmx/src/main/java/org/apache/qpid/server/store/berkeleydb/jmx/BDBHAMessageStoreManagerMBeanProvider.java
+++ b/qpid/java/bdbstore/jmx/src/main/java/org/apache/qpid/server/store/berkeleydb/jmx/BDBHAMessageStoreManagerMBeanProvider.java
@@ -28,11 +28,12 @@ import org.apache.qpid.server.jmx.MBeanProvider;
 import org.apache.qpid.server.jmx.ManagedObject;
 import org.apache.qpid.server.model.ConfiguredObject;
 import org.apache.qpid.server.model.VirtualHost;
-import org.apache.qpid.server.store.berkeleydb.BDBHAMessageStore;
+import org.apache.qpid.server.store.berkeleydb.BDBMessageStore;
+import org.apache.qpid.server.store.berkeleydb.replication.ReplicatedEnvironmentFacade;
 
 /**
  * This provide will create a {@link BDBHAMessageStoreManagerMBean} if the child is a virtual
- * host and of type {@link BDBHAMessageStore#TYPE}.
+ * host and of type {@link ReplicatedEnvironmentFacade#TYPE}.
  *
  */
 public class BDBHAMessageStoreManagerMBeanProvider implements MBeanProvider
@@ -48,7 +49,7 @@ public class BDBHAMessageStoreManagerMBeanProvider implements MBeanProvider
     public boolean isChildManageableByMBean(ConfiguredObject child)
     {
         return (child instanceof VirtualHost
-            && BDBHAMessageStore.TYPE.equals(child.getAttribute(VirtualHost.STORE_TYPE)));
+            && ReplicatedEnvironmentFacade.TYPE.equals(child.getAttribute(VirtualHost.STORE_TYPE)));
     }
 
     @Override
@@ -56,14 +57,15 @@ public class BDBHAMessageStoreManagerMBeanProvider implements MBeanProvider
     {
         VirtualHost virtualHostChild = (VirtualHost) child;
 
-        BDBHAMessageStore messageStore = (BDBHAMessageStore) virtualHostChild.getMessageStore();
+        BDBMessageStore messageStore = (BDBMessageStore) virtualHostChild.getMessageStore();
 
         if (LOGGER.isDebugEnabled())
         {
             LOGGER.debug("Creating mBean for child " + child);
         }
 
-        return new BDBHAMessageStoreManagerMBean(messageStore, (ManagedObject) parent);
+        ReplicatedEnvironmentFacade replicatedEnvironmentFacade = (ReplicatedEnvironmentFacade)messageStore.getEnvironmentFacade();
+        return new BDBHAMessageStoreManagerMBean(virtualHostChild.getName(), replicatedEnvironmentFacade, (ManagedObject) parent);
     }
 
     @Override
diff --git a/qpid/java/bdbstore/jmx/src/test/java/org/apache/qpid/server/store/berkeleydb/jmx/BDBHAMessageStoreManagerMBeanTest.java b/qpid/java/bdbstore/jmx/src/test/java/org/apache/qpid/server/store/berkeleydb/jmx/BDBHAMessageStoreManagerMBeanTest.java
index 0d963ebdae..fa16d1061a 100644
--- a/qpid/java/bdbstore/jmx/src/test/java/org/apache/qpid/server/store/berkeleydb/jmx/BDBHAMessageStoreManagerMBeanTest.java
+++ b/qpid/java/bdbstore/jmx/src/test/java/org/apache/qpid/server/store/berkeleydb/jmx/BDBHAMessageStoreManagerMBeanTest.java
@@ -37,10 +37,9 @@ import javax.management.openmbean.TabularData;
 
 import junit.framework.TestCase;
 
-import org.apache.qpid.server.store.StoreException;
 import org.apache.qpid.server.jmx.AMQManagedObject;
 import org.apache.qpid.server.jmx.ManagedObjectRegistry;
-import org.apache.qpid.server.store.berkeleydb.BDBHAMessageStore;
+import org.apache.qpid.server.store.berkeleydb.replication.ReplicatedEnvironmentFacade;
 
 public class BDBHAMessageStoreManagerMBeanTest extends TestCase
 {
@@ -53,7 +52,7 @@ public class BDBHAMessageStoreManagerMBeanTest extends TestCase
     private static final String TEST_STORE_NAME = "testStoreName";
     private static final boolean TEST_DESIGNATED_PRIMARY_FLAG = false;
 
-    private BDBHAMessageStore _store;
+    private ReplicatedEnvironmentFacade _replicatedEnvironmentFacade;
     private BDBHAMessageStoreManagerMBean _mBean;
     private AMQManagedObject _mBeanParent;
 
@@ -62,10 +61,10 @@ public class BDBHAMessageStoreManagerMBeanTest extends TestCase
     {
         super.setUp();
 
-        _store = mock(BDBHAMessageStore.class);
+        _replicatedEnvironmentFacade = mock(ReplicatedEnvironmentFacade.class);
         _mBeanParent = mock(AMQManagedObject.class);
         when(_mBeanParent.getRegistry()).thenReturn(mock(ManagedObjectRegistry.class));
-        _mBean = new BDBHAMessageStoreManagerMBean(_store, _mBeanParent);
+        _mBean = new BDBHAMessageStoreManagerMBean(TEST_STORE_NAME, _replicatedEnvironmentFacade, _mBeanParent);
     }
 
     @Override
@@ -76,64 +75,62 @@ public class BDBHAMessageStoreManagerMBeanTest extends TestCase
 
     public void testObjectName() throws Exception
     {
-        when(_store.getName()).thenReturn(TEST_STORE_NAME);
-
         String expectedObjectName = "org.apache.qpid:type=BDBHAMessageStore,name=" + ObjectName.quote(TEST_STORE_NAME);
         assertEquals(expectedObjectName, _mBean.getObjectName().toString());
     }
 
     public void testGroupName() throws Exception
     {
-        when(_store.getGroupName()).thenReturn(TEST_GROUP_NAME);
+        when(_replicatedEnvironmentFacade.getGroupName()).thenReturn(TEST_GROUP_NAME);
 
         assertEquals(TEST_GROUP_NAME, _mBean.getAttribute(ManagedBDBHAMessageStore.ATTR_GROUP_NAME));
     }
 
     public void testNodeName() throws Exception
     {
-        when(_store.getNodeName()).thenReturn(TEST_NODE_NAME);
+        when(_replicatedEnvironmentFacade.getNodeName()).thenReturn(TEST_NODE_NAME);
 
         assertEquals(TEST_NODE_NAME, _mBean.getAttribute(ManagedBDBHAMessageStore.ATTR_NODE_NAME));
     }
 
     public void testNodeHostPort() throws Exception
     {
-        when(_store.getNodeHostPort()).thenReturn(TEST_NODE_HOST_PORT);
+        when(_replicatedEnvironmentFacade.getHostPort()).thenReturn(TEST_NODE_HOST_PORT);
 
         assertEquals(TEST_NODE_HOST_PORT, _mBean.getAttribute(ManagedBDBHAMessageStore.ATTR_NODE_HOST_PORT));
     }
 
     public void testHelperHostPort() throws Exception
     {
-        when(_store.getHelperHostPort()).thenReturn(TEST_HELPER_HOST_PORT);
+        when(_replicatedEnvironmentFacade.getHelperHostPort()).thenReturn(TEST_HELPER_HOST_PORT);
 
         assertEquals(TEST_HELPER_HOST_PORT, _mBean.getAttribute(ManagedBDBHAMessageStore.ATTR_HELPER_HOST_PORT));
     }
 
     public void testDurability() throws Exception
     {
-        when(_store.getDurability()).thenReturn(TEST_DURABILITY);
+        when(_replicatedEnvironmentFacade.getDurability()).thenReturn(TEST_DURABILITY);
 
         assertEquals(TEST_DURABILITY, _mBean.getAttribute(ManagedBDBHAMessageStore.ATTR_DURABILITY));
     }
 
     public void testCoalescingSync() throws Exception
     {
-        when(_store.isCoalescingSync()).thenReturn(true);
+        when(_replicatedEnvironmentFacade.isCoalescingSync()).thenReturn(true);
 
         assertEquals(true, _mBean.getAttribute(ManagedBDBHAMessageStore.ATTR_COALESCING_SYNC));
     }
 
     public void testNodeState() throws Exception
     {
-        when(_store.getNodeState()).thenReturn(TEST_NODE_STATE);
+        when(_replicatedEnvironmentFacade.getNodeState()).thenReturn(TEST_NODE_STATE);
 
         assertEquals(TEST_NODE_STATE, _mBean.getAttribute(ManagedBDBHAMessageStore.ATTR_NODE_STATE));
     }
 
     public void testDesignatedPrimaryFlag() throws Exception
     {
-        when(_store.isDesignatedPrimary()).thenReturn(TEST_DESIGNATED_PRIMARY_FLAG);
+        when(_replicatedEnvironmentFacade.isDesignatedPrimary()).thenReturn(TEST_DESIGNATED_PRIMARY_FLAG);
 
         assertEquals(TEST_DESIGNATED_PRIMARY_FLAG, _mBean.getAttribute(ManagedBDBHAMessageStore.ATTR_DESIGNATED_PRIMARY));
     }
@@ -141,29 +138,29 @@ public class BDBHAMessageStoreManagerMBeanTest extends TestCase
     public void testGroupMembersForGroupWithOneNode() throws Exception
     {
         List<Map<String, String>> members = Collections.singletonList(createTestNodeResult());
-        when(_store.getGroupMembers()).thenReturn(members);
+        when(_replicatedEnvironmentFacade.getGroupMembers()).thenReturn(members);
 
         final TabularData resultsTable = _mBean.getAllNodesInGroup();
 
-        assertTableHasHeadingsNamed(resultsTable, BDBHAMessageStore.GRP_MEM_COL_NODE_NAME, BDBHAMessageStore.GRP_MEM_COL_NODE_HOST_PORT);
+        assertTableHasHeadingsNamed(resultsTable, ReplicatedEnvironmentFacade.GRP_MEM_COL_NODE_NAME, ReplicatedEnvironmentFacade.GRP_MEM_COL_NODE_HOST_PORT);
 
         final int numberOfDataRows = resultsTable.size();
         assertEquals("Unexpected number of data rows", 1 ,numberOfDataRows);
         final CompositeData row = (CompositeData) resultsTable.values().iterator().next();
-        assertEquals(TEST_NODE_NAME, row.get(BDBHAMessageStore.GRP_MEM_COL_NODE_NAME));
-        assertEquals(TEST_NODE_HOST_PORT, row.get(BDBHAMessageStore.GRP_MEM_COL_NODE_HOST_PORT));
+        assertEquals(TEST_NODE_NAME, row.get(ReplicatedEnvironmentFacade.GRP_MEM_COL_NODE_NAME));
+        assertEquals(TEST_NODE_HOST_PORT, row.get(ReplicatedEnvironmentFacade.GRP_MEM_COL_NODE_HOST_PORT));
     }
 
     public void testRemoveNodeFromReplicationGroup() throws Exception
     {
         _mBean.removeNodeFromGroup(TEST_NODE_NAME);
 
-        verify(_store).removeNodeFromGroup(TEST_NODE_NAME);
+        verify(_replicatedEnvironmentFacade).removeNodeFromGroup(TEST_NODE_NAME);
     }
 
     public void testRemoveNodeFromReplicationGroupWithError() throws Exception
     {
-        doThrow(new StoreException("mocked exception")).when(_store).removeNodeFromGroup(TEST_NODE_NAME);
+        doThrow(new RuntimeException("mocked exception")).when(_replicatedEnvironmentFacade).removeNodeFromGroup(TEST_NODE_NAME);
 
         try
         {
@@ -180,12 +177,12 @@ public class BDBHAMessageStoreManagerMBeanTest extends TestCase
     {
         _mBean.setDesignatedPrimary(true);
 
-        verify(_store).setDesignatedPrimary(true);
+        verify(_replicatedEnvironmentFacade).setDesignatedPrimary(true);
     }
 
     public void testSetAsDesignatedPrimaryWithError() throws Exception
     {
-        doThrow(new StoreException("mocked exception")).when(_store).setDesignatedPrimary(true);
+        doThrow(new RuntimeException("mocked exception")).when(_replicatedEnvironmentFacade).setDesignatedPrimary(true);
 
         try
         {
@@ -205,7 +202,7 @@ public class BDBHAMessageStoreManagerMBeanTest extends TestCase
 
         _mBean.updateAddress(TEST_NODE_NAME, newHostName, newPort);
 
-        verify(_store).updateAddress(TEST_NODE_NAME, newHostName, newPort);
+        verify(_replicatedEnvironmentFacade).updateAddress(TEST_NODE_NAME, newHostName, newPort);
     }
 
     private void assertTableHasHeadingsNamed(final TabularData resultsTable, String... headingNames)
@@ -220,8 +217,8 @@ public class BDBHAMessageStoreManagerMBeanTest extends TestCase
     private Map<String, String> createTestNodeResult()
     {
         Map<String, String> items = new HashMap<String, String>();
-        items.put(BDBHAMessageStore.GRP_MEM_COL_NODE_NAME, TEST_NODE_NAME);
-        items.put(BDBHAMessageStore.GRP_MEM_COL_NODE_HOST_PORT, TEST_NODE_HOST_PORT);
+        items.put(ReplicatedEnvironmentFacade.GRP_MEM_COL_NODE_NAME, TEST_NODE_NAME);
+        items.put(ReplicatedEnvironmentFacade.GRP_MEM_COL_NODE_HOST_PORT, TEST_NODE_HOST_PORT);
         return items;
     }
 }