QPID-5967: Set permitted nodes on a replica from application state of a master node

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

2 files changed, 103 insertions, 82 deletions

diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/replication/ReplicatedEnvironmentFacade.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/replication/ReplicatedEnvironmentFacade.java
index 92115dd39f..89fac0994f 100644
--- a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/replication/ReplicatedEnvironmentFacade.java
+++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/replication/ReplicatedEnvironmentFacade.java
@@ -1094,15 +1094,6 @@ public class ReplicatedEnvironmentFacade implements EnvironmentFacade, StateChan
         return environment;
     }
 
-    NodeState getRemoteNodeState(ReplicationNode repNode) throws IOException, ServiceConnectFailedException
-    {
-        if (repNode == null)
-        {
-            throw new IllegalArgumentException("Node cannot be null");
-        }
-        return new DbPing(repNode, (String)_configuration.getGroupName(), DB_PING_SOCKET_TIMEOUT).getNodeState();
-    }
-
     public int getNumberOfElectableGroupMembers()
     {
         if (_state.get() != State.OPEN)
@@ -1181,6 +1172,105 @@ public class ReplicatedEnvironmentFacade implements EnvironmentFacade, StateChan
         }
     }
 
+    Set<String> getPermittedNodes()
+    {
+        return Collections.unmodifiableSet(_permittedNodes);
+    }
+
+    public static NodeState getRemoteNodeState(String groupName, ReplicationNode repNode) throws IOException, ServiceConnectFailedException
+    {
+        if (repNode == null)
+        {
+            throw new IllegalArgumentException("Node cannot be null");
+        }
+        return new DbPing(repNode, groupName, DB_PING_SOCKET_TIMEOUT).getNodeState();
+    }
+
+    public static Set<String> convertApplicationStateBytesToPermittedNodeList(byte[] applicationState)
+    {
+        if (applicationState == null || applicationState.length == 0)
+        {
+            return Collections.emptySet();
+        }
+
+        ObjectMapper objectMapper = new ObjectMapper();
+        try
+        {
+            Map<String, Object> settings = objectMapper.readValue(applicationState, Map.class);
+            return new HashSet<String>((Collection<String>)settings.get(PERMITTED_NODE_LIST));
+        }
+        catch (Exception e)
+        {
+            throw new RuntimeException("Unexpected exception on de-serializing of application state", e);
+        }
+    }
+
+    public static void connectToHelperNodeAndCheckPermittedHosts(String nodeName, String hostPort, String groupName, String helperNodeName, String helperHostPort)
+    {
+        if (LOGGER.isDebugEnabled())
+        {
+            LOGGER.debug(String.format("Requesting state of the node '%s' at '%s'", helperNodeName, helperHostPort));
+        }
+
+        if (helperNodeName == null || "".equals(helperNodeName))
+        {
+            throw new IllegalConfigurationException(String.format("A helper node is not specified for node '%s'"
+                    + " joining the group '%s'", nodeName, groupName));
+        }
+
+        Collection<String> permittedNodes = null;
+        try
+        {
+            ReplicationNodeImpl node = new ReplicationNodeImpl(helperNodeName, helperHostPort);
+            NodeState state = getRemoteNodeState(groupName, node);
+            byte[] applicationState = state.getAppState();
+            permittedNodes = convertApplicationStateBytesToPermittedNodeList(applicationState);
+        }
+        catch (IOException e)
+        {
+            throw new IllegalConfigurationException(String.format("Cannot connect to '%s'", helperHostPort), e);
+        }
+        catch (ServiceConnectFailedException e)
+        {
+            throw new IllegalConfigurationException(String.format("Failure to connect to '%s'", helperHostPort), e);
+        }
+        catch (Exception e)
+        {
+            throw new RuntimeException(String.format("Unexpected exception on attempt to retrieve state from '%s' at '%s'",
+                    helperNodeName, helperHostPort), e);
+        }
+
+        if (LOGGER.isDebugEnabled())
+        {
+            LOGGER.debug(String.format("Attribute 'permittedNodes' on node '%s' is set to '%s'", helperNodeName, String.valueOf(permittedNodes)));
+        }
+
+        if (permittedNodes==null || !permittedNodes.contains(hostPort))
+        {
+            throw new IllegalConfigurationException(String.format("Node from '%s' is not permitted!", hostPort));
+        }
+    }
+
+    private void findMasterNodeStateAndApplyPermittedNodes(Collection<NodeState> nodeStates)
+    {
+        if (ReplicatedEnvironment.State.MASTER != _environment.getState())
+        {
+            for (NodeState nodeState : nodeStates)
+            {
+                if (nodeState.getNodeState() == ReplicatedEnvironment.State.MASTER)
+                {
+                    byte[] applicationState = nodeState.getAppState();
+                    Set<String> permittedNodes = convertApplicationStateBytesToPermittedNodeList(applicationState);
+                    if (!_permittedNodes.equals(permittedNodes))
+                    {
+                        setPermittedNodes(permittedNodes);
+                    }
+                    break;
+                }
+            }
+        }
+    }
+
     private void registerAppStateMonitorIfPermittedNodesSpecified()
     {
         if (!_permittedNodes.isEmpty())
@@ -1286,8 +1376,9 @@ public class ReplicatedEnvironmentFacade implements EnvironmentFacade, StateChan
                     executeDatabasePingerOnNodeChangesIfMaster(nodeStates);
 
                     notifyGroupListenerAboutNodeStates(nodeStates);
-                }
 
+                    findMasterNodeStateAndApplyPermittedNodes(nodeStates.values());
+                }
             }
             finally
             {
@@ -1384,7 +1475,7 @@ public class ReplicatedEnvironmentFacade implements EnvironmentFacade, StateChan
                         NodeState nodeStateObject = null;
                         try
                         {
-                            nodeStateObject = getRemoteNodeState(node);
+                            nodeStateObject = getRemoteNodeState((String)_configuration.getGroupName(), node);
                         }
                         catch (IOException | ServiceConnectFailedException e )
                         {
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 cacb04736c..5489493f74 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
@@ -20,11 +20,9 @@
  */
 package org.apache.qpid.server.virtualhostnode.berkeleydb;
 
-import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.security.PrivilegedAction;
 import java.util.Collection;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
@@ -43,10 +41,8 @@ import com.sleepycat.je.rep.ReplicatedEnvironment;
 import com.sleepycat.je.rep.ReplicationNode;
 import com.sleepycat.je.rep.StateChangeEvent;
 import com.sleepycat.je.rep.StateChangeListener;
-import com.sleepycat.je.rep.util.DbPing;
 import com.sleepycat.je.rep.util.ReplicationGroupAdmin;
 import com.sleepycat.je.rep.utilint.HostPortPair;
-import com.sleepycat.je.rep.utilint.ServiceDispatcher;
 import org.apache.log4j.Logger;
 
 import org.apache.qpid.server.configuration.IllegalConfigurationException;
@@ -73,7 +69,6 @@ import org.apache.qpid.server.store.berkeleydb.replication.ReplicationGroupListe
 import org.apache.qpid.server.util.ServerScopedRuntimeException;
 import org.apache.qpid.server.virtualhost.berkeleydb.BDBHAVirtualHostImpl;
 import org.apache.qpid.server.virtualhostnode.AbstractVirtualHostNode;
-import org.codehaus.jackson.map.ObjectMapper;
 
 @ManagedObject( category = false, type = BDBHAVirtualHostNodeImpl.VIRTUAL_HOST_NODE_TYPE )
 public class BDBHAVirtualHostNodeImpl extends AbstractVirtualHostNode<BDBHAVirtualHostNodeImpl> implements
@@ -263,7 +258,7 @@ public class BDBHAVirtualHostNodeImpl extends AbstractVirtualHostNode<BDBHAVirtu
         {
             try
             {
-                connectToHelperNodeAndCheckPermittedHosts(getHelperNodeName(), getHelperAddress(), getAddress());
+                ReplicatedEnvironmentFacade.connectToHelperNodeAndCheckPermittedHosts(getName(), getAddress(), getGroupName(), getHelperNodeName(), getHelperAddress());
             }
             catch(IllegalConfigurationException e)
             {
@@ -706,71 +701,6 @@ public class BDBHAVirtualHostNodeImpl extends AbstractVirtualHostNode<BDBHAVirtu
         return getAddress().equals(getHelperAddress());
     }
 
-    private void connectToHelperNodeAndCheckPermittedHosts(String helperNodeName, String helperHostPort, String hostPort)
-    {
-        if (LOGGER.isDebugEnabled())
-        {
-            LOGGER.debug(String.format("Requesting state of the node '%s' at '%s'", helperNodeName, helperHostPort));
-        }
-
-        if (_helperNodeName == null || "".equals(_helperNodeName))
-        {
-            throw new IllegalConfigurationException(String.format("An attribute '%s' is not set in node '%s'"
-                    + " on joining the group '%s'", HELPER_NODE_NAME, getName(), getGroupName()));
-        }
-
-        Collection<String> permittedNodes = null;
-        try
-        {
-            ReplicatedEnvironmentFacade.ReplicationNodeImpl node = new ReplicatedEnvironmentFacade.ReplicationNodeImpl(helperNodeName, helperHostPort);
-            NodeState state = new DbPing(node, getGroupName(), ReplicatedEnvironmentFacade.DB_PING_SOCKET_TIMEOUT).getNodeState();
-            byte[] applicationState = state.getAppState();
-            permittedNodes = bytesToPermittedNodeList(applicationState);
-        }
-        catch (IOException e)
-        {
-            throw new IllegalConfigurationException(String.format("Cannot connect to '%s'", helperHostPort), e);
-        }
-        catch (ServiceDispatcher.ServiceConnectFailedException e)
-        {
-            throw new IllegalConfigurationException(String.format("Failure to connect to '%s'", helperHostPort), e);
-        }
-        catch (Exception e)
-        {
-            throw new RuntimeException(String.format("Unexpected exception on attempt to retrieve state from '%s' at '%s'",
-                    helperNodeName, helperHostPort), e);
-        }
-
-        if (LOGGER.isDebugEnabled())
-        {
-            LOGGER.debug(String.format("Attribute 'permittedNodes' on node '%s' is set to '%s'", helperNodeName, String.valueOf(permittedNodes)));
-        }
-
-        if (permittedNodes != null && !permittedNodes.isEmpty() && !permittedNodes.contains(hostPort))
-        {
-            throw new IllegalConfigurationException(String.format("Node from '%s' is not permitted!", hostPort));
-        }
-    }
-
-    private Collection<String> bytesToPermittedNodeList(byte[] applicationState)
-    {
-        if (applicationState == null || applicationState.length == 0)
-        {
-            return Collections.emptySet();
-        }
-
-        ObjectMapper objectMapper = new ObjectMapper();
-        try
-        {
-            Map<String, Object> settings = objectMapper.readValue(applicationState, Map.class);
-            return (Collection<String>)settings.get(ReplicatedEnvironmentFacade.PERMITTED_NODE_LIST);
-        }
-        catch (Exception e)
-        {
-            throw new RuntimeException("Unexpected exception on de-serializing of application state", e);
-        }
-    }
-
     private class RemoteNodesDiscoverer implements ReplicationGroupListener
     {
         @Override