QPID-5715: Fix Virtual Host MBean creation in JMX management plugin

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

6 files changed, 306 insertions, 33 deletions

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 4aff32fe13..9602c2c29c 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
@@ -439,6 +439,7 @@ public class BDBHAVirtualHostNodeImpl extends AbstractVirtualHostNode<BDBHAVirtu
         if (virtualHost!= null)
         {
             virtualHost.close();
+            childRemoved(virtualHost);
         }
     }
 
@@ -473,6 +474,8 @@ public class BDBHAVirtualHostNodeImpl extends AbstractVirtualHostNode<BDBHAVirtu
                 LOGGER.error("Unexpected state change: " + state);
                 throw new IllegalStateException("Unexpected state change: " + state);
             }
+
+            attributeSet(ROLE, _role, state.name());
         }
     }
 
diff --git a/qpid/java/bdbstore/systests/src/main/java/org/apache/qpid/server/store/berkeleydb/HAClusterBlackboxTest.java b/qpid/java/bdbstore/systests/src/main/java/org/apache/qpid/server/store/berkeleydb/HAClusterBlackboxTest.java
index 0de61d68cd..faeb068f4d 100644
--- a/qpid/java/bdbstore/systests/src/main/java/org/apache/qpid/server/store/berkeleydb/HAClusterBlackboxTest.java
+++ b/qpid/java/bdbstore/systests/src/main/java/org/apache/qpid/server/store/berkeleydb/HAClusterBlackboxTest.java
@@ -20,6 +20,10 @@
 package org.apache.qpid.server.store.berkeleydb;
 
 import java.io.File;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
@@ -30,6 +34,7 @@ import org.apache.log4j.Logger;
 import org.apache.qpid.client.AMQConnection;
 import org.apache.qpid.jms.ConnectionListener;
 import org.apache.qpid.jms.ConnectionURL;
+import org.apache.qpid.server.virtualhostnode.berkeleydb.BDBHAVirtualHostNode;
 import org.apache.qpid.test.utils.QpidBrokerTestCase;
 import org.apache.qpid.test.utils.TestUtils;
 
@@ -115,6 +120,129 @@ public class HAClusterBlackboxTest extends QpidBrokerTestCase
         connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
     }
 
+    public void testTransferMasterFromLocalNode() throws Exception
+    {
+        final Connection connection = getConnection(_brokerFailoverUrl);
+
+        ((AMQConnection)connection).setConnectionListener(_failoverAwaitingListener);
+
+        final int activeBrokerPort = _clusterCreator.getBrokerPortNumberFromConnection(connection);
+        LOGGER.info("Active connection port " + activeBrokerPort);
+
+        final int inactiveBrokerPort = _clusterCreator.getPortNumberOfAnInactiveBroker(connection);
+        LOGGER.info("Update role attribute on inactive broker on port " + inactiveBrokerPort);
+
+        Map<String, Object> attributes = _clusterCreator.getNodeAttributes(inactiveBrokerPort);
+        assertEquals("Inactive broker has unexpeced role", "REPLICA", attributes.get(BDBHAVirtualHostNode.ROLE));
+        _clusterCreator.setNodeAttributes(inactiveBrokerPort, Collections.<String, Object>singletonMap(BDBHAVirtualHostNode.ROLE, "MASTER"));
+
+        _failoverAwaitingListener.assertFailoverOccurs(20000);
+        LOGGER.info("Listener has finished");
+
+        attributes = _clusterCreator.getNodeAttributes(inactiveBrokerPort);
+        assertEquals("Inactive broker has unexpeced role", "MASTER", attributes.get(BDBHAVirtualHostNode.ROLE));
+
+        assertProducingConsuming(connection);
+
+        _clusterCreator.awaitNodeToAttainRole(activeBrokerPort, "REPLICA");
+    }
+
+    public void testTransferMasterFromRemoteNode() throws Exception
+    {
+        final Connection connection = getConnection(_brokerFailoverUrl);
+
+        ((AMQConnection)connection).setConnectionListener(_failoverAwaitingListener);
+
+        final int activeBrokerPort = _clusterCreator.getBrokerPortNumberFromConnection(connection);
+        LOGGER.info("Active connection port " + activeBrokerPort);
+
+        final int inactiveBrokerPort = _clusterCreator.getPortNumberOfAnInactiveBroker(connection);
+        LOGGER.info("Update role attribute on inactive broker on port " + inactiveBrokerPort);
+
+        _clusterCreator.awaitNodeToAttainRole(activeBrokerPort, inactiveBrokerPort, "REPLICA");
+        Map<String, Object> attributes = _clusterCreator.getNodeAttributes(activeBrokerPort, inactiveBrokerPort);
+        assertEquals("Inactive broker has unexpeced role", "REPLICA", attributes.get(BDBHAVirtualHostNode.ROLE));
+
+        _clusterCreator.setNodeAttributes(activeBrokerPort, inactiveBrokerPort, Collections.<String, Object>singletonMap(BDBHAVirtualHostNode.ROLE, "MASTER"));
+
+        _failoverAwaitingListener.assertFailoverOccurs(20000);
+        LOGGER.info("Listener has finished");
+
+        attributes = _clusterCreator.getNodeAttributes(inactiveBrokerPort);
+        assertEquals("Inactive broker has unexpeced role", "MASTER", attributes.get(BDBHAVirtualHostNode.ROLE));
+
+        assertProducingConsuming(connection);
+
+        _clusterCreator.awaitNodeToAttainRole(activeBrokerPort, "REPLICA");
+    }
+
+    public void testQuorumOverride() throws Exception
+    {
+        final Connection connection = getConnection(_brokerFailoverUrl);
+
+        ((AMQConnection)connection).setConnectionListener(_failoverAwaitingListener);
+
+        Set<Integer> ports = _clusterCreator.getBrokerPortNumbersForNodes();
+        Iterator<Integer> iterator = ports.iterator();
+        Integer quorumOverridePort = iterator.next();
+        iterator.remove();
+
+        for (Integer p : ports)
+        {
+            _clusterCreator.stopNode(p);
+        }
+
+        Map<String, Object> attributes = _clusterCreator.getNodeAttributes(quorumOverridePort);
+        assertEquals("Broker has unexpeced quorum override", new Integer(0), attributes.get(BDBHAVirtualHostNode.QUORUM_OVERRIDE));
+        _clusterCreator.setNodeAttributes(quorumOverridePort, Collections.<String, Object>singletonMap(BDBHAVirtualHostNode.QUORUM_OVERRIDE, 1));
+
+        _failoverAwaitingListener.assertFailoverOccurs(20000);
+        LOGGER.info("Listener has finished");
+
+        attributes = _clusterCreator.getNodeAttributes(quorumOverridePort);
+        assertEquals("Broker has unexpeced quorum override", new Integer(1), attributes.get(BDBHAVirtualHostNode.QUORUM_OVERRIDE));
+
+        assertProducingConsuming(connection);
+    }
+
+    public void testPriority() throws Exception
+    {
+        final Connection connection = getConnection(_brokerFailoverUrl);
+
+        ((AMQConnection)connection).setConnectionListener(_failoverAwaitingListener);
+
+        final int activeBrokerPort = _clusterCreator.getBrokerPortNumberFromConnection(connection);
+        LOGGER.info("Active connection port " + activeBrokerPort);
+
+        int priority = 1;
+        Integer highestPriorityBrokerPort = null;
+        Set<Integer> ports = _clusterCreator.getBrokerPortNumbersForNodes();
+        for (Integer port : ports)
+        {
+            if (activeBrokerPort != port.intValue())
+            {
+                priority = priority + 1;
+                highestPriorityBrokerPort = port;
+                _clusterCreator.setNodeAttributes(port, port, Collections.<String, Object>singletonMap(BDBHAVirtualHostNode.PRIORITY, priority));
+                Map<String, Object> attributes = _clusterCreator.getNodeAttributes(port, port);
+                assertEquals("Broker has unexpeced priority", priority, attributes.get(BDBHAVirtualHostNode.PRIORITY));
+            }
+        }
+
+        LOGGER.info("Broker on port " + highestPriorityBrokerPort + " has the highest priority of " + priority);
+
+        LOGGER.info("Shutting down the MASTER");
+        _clusterCreator.stopNode(activeBrokerPort);
+
+        _failoverAwaitingListener.assertFailoverOccurs(20000);
+        LOGGER.info("Listener has finished");
+
+        Map<String, Object> attributes = _clusterCreator.getNodeAttributes(highestPriorityBrokerPort, highestPriorityBrokerPort);
+        assertEquals("Inactive broker has unexpeced role", "MASTER", attributes.get(BDBHAVirtualHostNode.ROLE));
+
+        assertProducingConsuming(connection);
+    }
+
     private final class FailoverAwaitingListener implements ConnectionListener
     {
         private final CountDownLatch _failoverLatch = new CountDownLatch(1);
diff --git a/qpid/java/bdbstore/systests/src/main/java/org/apache/qpid/server/store/berkeleydb/HAClusterManagementTest.java b/qpid/java/bdbstore/systests/src/main/java/org/apache/qpid/server/store/berkeleydb/HAClusterManagementTest.java
index 200f2c1087..c184ef8f8d 100644
--- a/qpid/java/bdbstore/systests/src/main/java/org/apache/qpid/server/store/berkeleydb/HAClusterManagementTest.java
+++ b/qpid/java/bdbstore/systests/src/main/java/org/apache/qpid/server/store/berkeleydb/HAClusterManagementTest.java
@@ -25,8 +25,11 @@ import static com.sleepycat.je.rep.ReplicatedEnvironment.State.REPLICA;
 import static com.sleepycat.je.rep.ReplicatedEnvironment.State.UNKNOWN;
 
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashSet;
 import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
 import java.util.Set;
 
 import javax.jms.Connection;
@@ -37,11 +40,14 @@ import javax.management.openmbean.TabularData;
 import org.apache.log4j.Logger;
 import org.apache.qpid.jms.ConnectionURL;
 import org.apache.qpid.management.common.mbeans.ManagedBroker;
+import org.apache.qpid.server.model.State;
+import org.apache.qpid.server.model.VirtualHost;
 import org.apache.qpid.server.store.berkeleydb.jmx.ManagedBDBHAMessageStore;
+import org.apache.qpid.server.virtualhostnode.berkeleydb.BDBHAVirtualHostNode;
+import org.apache.qpid.systest.rest.RestTestHelper;
 import org.apache.qpid.test.utils.JMXTestUtils;
 import org.apache.qpid.test.utils.QpidBrokerTestCase;
-
-import com.sleepycat.je.EnvironmentFailureException;
+import org.junit.Assert;
 
 /**
  * System test verifying the ability to control a cluster via the Management API.
@@ -214,6 +220,68 @@ public class HAClusterManagementTest extends QpidBrokerTestCase
         assertTrue("Unexpected designated primary after change", storeBean.getDesignatedPrimary());
     }
 
+    public void testVirtualHostMbeanOnMasterTransfer() throws Exception
+    {
+        Connection connection = getConnection(_brokerFailoverUrl);
+        int activeBrokerPort = _clusterCreator.getBrokerPortNumberFromConnection(connection);
+        LOGGER.info("Active connection port " + activeBrokerPort);
+        connection.close();
+
+        Set<Integer> ports = _clusterCreator.getBrokerPortNumbersForNodes();
+        ports.remove(activeBrokerPort);
+
+        int inactiveBrokerPort = ports.iterator().next();
+        LOGGER.info("Update role attribute on inactive broker on port " + inactiveBrokerPort);
+
+        ManagedBroker inactiveVirtualHostMBean = getManagedBrokerBeanForNodeAtBrokerPort(inactiveBrokerPort);
+
+        try
+        {
+            inactiveVirtualHostMBean.createNewQueue(getTestQueueName(), null, true);
+            fail("Exception not thrown");
+        }
+        catch (Exception e)
+        {
+            String message = e.getMessage();
+            assertEquals("The virtual hosts state of PASSIVE does not permit this operation.", message);
+        }
+
+        Map<String, Object> attributes = _clusterCreator.getNodeAttributes(inactiveBrokerPort);
+        assertEquals("Inactive broker has unexpeced role", "REPLICA", attributes.get(BDBHAVirtualHostNode.ROLE));
+        _clusterCreator.setNodeAttributes(inactiveBrokerPort, Collections.<String, Object>singletonMap(BDBHAVirtualHostNode.ROLE, "MASTER"));
+
+        _clusterCreator.awaitNodeToAttainRole(inactiveBrokerPort, "MASTER");
+
+        awaitVirtualHostAtNode(inactiveBrokerPort);
+
+        ManagedBroker activeVirtualHostMBean = getManagedBrokerBeanForNodeAtBrokerPort(inactiveBrokerPort);
+        activeVirtualHostMBean.createNewQueue(getTestQueueName() + inactiveBrokerPort, null, true);
+    }
+
+    public void awaitVirtualHostAtNode(int brokerPort) throws Exception
+    {
+        final long startTime = System.currentTimeMillis();
+        Map<String, Object> data = Collections.emptyMap();
+        String nodeName = _clusterCreator.getNodeNameForBrokerPort(brokerPort);
+        RestTestHelper restHelper = _clusterCreator.createRestTestHelper(brokerPort);
+        while(!State.ACTIVE.name().equals(data.get(VirtualHost.STATE)) && (System.currentTimeMillis() - startTime) < 30000)
+        {
+            LOGGER.debug("Awaiting virtual host '" + nodeName + "' to transit into active state");
+            List<Map<String, Object>> results= restHelper.getJsonAsList("virtualhost/" + nodeName + "/" + VIRTUAL_HOST);
+            if (results.size()== 1)
+            {
+                data = results.get(0);
+            }
+
+            if (!State.ACTIVE.name().equals(data.get(VirtualHost.STATE)))
+            {
+                Thread.sleep(1000);
+            }
+        }
+        Assert.assertEquals("Virtual host is not active", State.ACTIVE.name(), data.get(VirtualHost.STATE));
+        LOGGER.debug("Virtual host '" + nodeName + "' is in active state");
+    }
+
     private ManagedBDBHAMessageStore getStoreBeanForNodeAtBrokerPort(final int brokerPortNumber) throws Exception
     {
         _jmxUtils.open(brokerPortNumber);
diff --git a/qpid/java/bdbstore/systests/src/main/java/org/apache/qpid/server/store/berkeleydb/HAClusterTwoNodeTest.java b/qpid/java/bdbstore/systests/src/main/java/org/apache/qpid/server/store/berkeleydb/HAClusterTwoNodeTest.java
index b414905ddc..24296246b8 100644
--- a/qpid/java/bdbstore/systests/src/main/java/org/apache/qpid/server/store/berkeleydb/HAClusterTwoNodeTest.java
+++ b/qpid/java/bdbstore/systests/src/main/java/org/apache/qpid/server/store/berkeleydb/HAClusterTwoNodeTest.java
@@ -196,17 +196,4 @@ public class HAClusterTwoNodeTest extends QpidBrokerTestCase
         return storeBean;
     }
 
-    private void assertProducingConsuming(final Connection connection) throws JMSException, Exception
-    {
-        Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
-        Destination destination = session.createQueue(getTestQueueName());
-        MessageConsumer consumer = session.createConsumer(destination);
-        sendMessage(session, destination, 1);
-        connection.start();
-        Message m1 = consumer.receive(RECEIVE_TIMEOUT);
-        assertNotNull("Message 1 is not received", m1);
-        assertEquals("Unexpected first message received", 0, m1.getIntProperty(INDEX));
-        session.commit();
-    }
-
 }
diff --git a/qpid/java/bdbstore/systests/src/main/java/org/apache/qpid/server/store/berkeleydb/HAClusterWhiteboxTest.java b/qpid/java/bdbstore/systests/src/main/java/org/apache/qpid/server/store/berkeleydb/HAClusterWhiteboxTest.java
index 408643b98a..ef5cc7c464 100644
--- a/qpid/java/bdbstore/systests/src/main/java/org/apache/qpid/server/store/berkeleydb/HAClusterWhiteboxTest.java
+++ b/qpid/java/bdbstore/systests/src/main/java/org/apache/qpid/server/store/berkeleydb/HAClusterWhiteboxTest.java
@@ -235,22 +235,6 @@ public class HAClusterWhiteboxTest extends QpidBrokerTestCase
         killBroker(initialPortNumber); // kill awaits the death of the child
     }
 
-    private void assertProducingConsuming(final Connection connection) throws JMSException, Exception
-    {
-        Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
-        Destination destination = session.createQueue(getTestQueueName());
-        MessageConsumer consumer = session.createConsumer(destination);
-        sendMessage(session, destination, 2);
-        connection.start();
-        Message m1 = consumer.receive(RECEIVE_TIMEOUT);
-        assertNotNull("Message 1 is not received", m1);
-        assertEquals("Unexpected first message received", 0, m1.getIntProperty(INDEX));
-        Message m2 = consumer.receive(RECEIVE_TIMEOUT);
-        assertNotNull("Message 2 is not received", m2);
-        assertEquals("Unexpected second message received", 1, m2.getIntProperty(INDEX));
-        session.commit();
-    }
-
     private void closeConnection(final Connection initialConnection)
     {
         try
diff --git a/qpid/java/bdbstore/systests/src/main/java/org/apache/qpid/server/store/berkeleydb/HATestClusterCreator.java b/qpid/java/bdbstore/systests/src/main/java/org/apache/qpid/server/store/berkeleydb/HATestClusterCreator.java
index 96a75db2cc..48e15561d0 100644
--- a/qpid/java/bdbstore/systests/src/main/java/org/apache/qpid/server/store/berkeleydb/HATestClusterCreator.java
+++ b/qpid/java/bdbstore/systests/src/main/java/org/apache/qpid/server/store/berkeleydb/HATestClusterCreator.java
@@ -20,8 +20,10 @@
 package org.apache.qpid.server.store.berkeleydb;
 
 import java.io.File;
+import java.io.IOException;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
@@ -42,11 +44,17 @@ import org.apache.commons.lang.StringUtils;
 import org.apache.log4j.Logger;
 import org.apache.qpid.client.AMQConnection;
 import org.apache.qpid.client.AMQConnectionURL;
+import org.apache.qpid.server.management.plugin.HttpManagement;
+import org.apache.qpid.server.model.Plugin;
+import org.apache.qpid.server.model.Port;
 import org.apache.qpid.server.model.VirtualHostNode;
+import org.apache.qpid.server.virtualhostnode.berkeleydb.BDBHARemoteReplicationNode;
 import org.apache.qpid.server.virtualhostnode.berkeleydb.BDBHAVirtualHostNode;
+import org.apache.qpid.systest.rest.RestTestHelper;
 import org.apache.qpid.test.utils.QpidBrokerTestCase;
 import org.apache.qpid.test.utils.TestBrokerConfiguration;
 import org.apache.qpid.url.URLSyntaxException;
+import org.junit.Assert;
 
 import com.sleepycat.je.rep.ReplicationConfig;
 
@@ -119,6 +127,9 @@ public class HATestClusterCreator
 
             TestBrokerConfiguration brokerConfiguration = _testcase.getBrokerConfiguration(brokerPort);
             brokerConfiguration.addJmxManagementConfiguration();
+            brokerConfiguration.addHttpManagementConfiguration();
+            brokerConfiguration.setObjectAttribute(Plugin.class, TestBrokerConfiguration.ENTRY_NAME_HTTP_MANAGEMENT, HttpManagement.HTTP_BASIC_AUTHENTICATION_ENABLED, true);
+            brokerConfiguration.setObjectAttribute(Port.class, TestBrokerConfiguration.ENTRY_NAME_HTTP_PORT, Port.PORT, _testcase.getHttpManagementPort(brokerPort));
             brokerConfiguration.setObjectAttributes(VirtualHostNode.class, _virtualHostName, virtualHostNodeAttributes);
 
             brokerPort = _testcase.getNextAvailable(bdbPort + 1);
@@ -273,8 +284,7 @@ public class HATestClusterCreator
         return new AMQConnectionURL(String.format(MANY_BROKER_URL_FORMAT, _virtualHostName, brokerList, FAILOVER_CYCLECOUNT));
     }
 
-    public AMQConnectionURL getConnectionUrlForSingleNodeWithoutRetry(final int brokerPortNumber) throws
-                                                                                                  URLSyntaxException
+    public AMQConnectionURL getConnectionUrlForSingleNodeWithoutRetry(final int brokerPortNumber) throws URLSyntaxException
     {
         return getConnectionUrlForSingleNode(brokerPortNumber, false);
     }
@@ -380,4 +390,97 @@ public class HATestClusterCreator
         config.setSaved(false);
     }
 
+    public String getNodeNameForBrokerPort(final int brokerPort)
+    {
+        return getNodeNameForNodeAt(_brokerPortToBdbPortMap.get(brokerPort));
+    }
+
+    public void setNodeAttributes(int brokerPort, Map<String, Object> attributeMap)
+            throws Exception
+    {
+        setNodeAttributes(brokerPort, brokerPort, attributeMap);
+    }
+
+    public void setNodeAttributes(int localNodePort, int remoteNodePort, Map<String, Object> attributeMap)
+            throws Exception
+    {
+        RestTestHelper restHelper = createRestTestHelper(localNodePort);
+        String url = getNodeRestUrl(localNodePort, remoteNodePort);
+        int status = restHelper.submitRequest(url, "PUT", attributeMap);
+        if (status != 200)
+        {
+            throw new Exception("Unexpected http status when updating " + getNodeNameForBrokerPort(remoteNodePort) + " attribute's : " + status);
+        }
+    }
+
+    private String getNodeRestUrl(int localNodePort, int remoteNodePort)
+    {
+        String remoteNodeName = getNodeNameForBrokerPort(remoteNodePort);
+        String localNodeName = getNodeNameForBrokerPort(localNodePort);
+        String url = null;
+        if (localNodePort == remoteNodePort)
+        {
+            url = "/api/latest/virtualhostnode/" + localNodeName;
+        }
+        else
+        {
+            url = "/api/latest/replicationnode/" + localNodeName + "/" + remoteNodeName;
+        }
+        return url;
+    }
+
+    public Map<String, Object> getNodeAttributes(int brokerPort) throws IOException
+    {
+        return getNodeAttributes(brokerPort, brokerPort);
+    }
+
+    public Map<String, Object> getNodeAttributes(int localNodePort, int remoteNodePort) throws IOException
+    {
+        RestTestHelper restHelper = createRestTestHelper(localNodePort);
+        List<Map<String, Object>> results= restHelper.getJsonAsList(getNodeRestUrl(localNodePort, remoteNodePort));
+        int size = results.size();
+        if (size == 0)
+        {
+            return Collections.emptyMap();
+        }
+        else if (size == 1)
+        {
+            return results.get(0);
+        }
+        else
+        {
+            throw new RuntimeException("Unexpected number of nodes " + size);
+        }
+    }
+
+    public void awaitNodeToAttainRole(int brokerPort, String desiredRole) throws Exception
+    {
+        awaitNodeToAttainRole(brokerPort, brokerPort, desiredRole);
+    }
+
+    public void awaitNodeToAttainRole(int localNodePort, int remoteNodePort, String desiredRole) throws Exception
+    {
+        final long startTime = System.currentTimeMillis();
+        Map<String, Object> data = Collections.emptyMap();
+
+        while(!desiredRole.equals(data.get(BDBHARemoteReplicationNode.ROLE)) && (System.currentTimeMillis() - startTime) < 30000)
+        {
+            LOGGER.debug("Awaiting node '" + getNodeNameForBrokerPort(remoteNodePort) + "' to transit into " + desiredRole + " role");
+            data = getNodeAttributes(localNodePort, remoteNodePort);
+            if (!desiredRole.equals(data.get(BDBHARemoteReplicationNode.ROLE)))
+            {
+                Thread.sleep(1000);
+            }
+        }
+        LOGGER.debug("Node '" + getNodeNameForBrokerPort(remoteNodePort) + "' role is " + data.get(BDBHARemoteReplicationNode.ROLE));
+        Assert.assertEquals("Node is in unexpected role", desiredRole, data.get(BDBHARemoteReplicationNode.ROLE));
+    }
+
+    public RestTestHelper createRestTestHelper(int brokerPort)
+    {
+        int httpPort = _testcase.getHttpManagementPort(brokerPort);
+        RestTestHelper helper = new RestTestHelper(httpPort);
+        helper.setUsernameAndPassword("webadmin", "webadmin");
+        return helper;
+    }
 }