Bug 2526: Race condition may cause missing routes in RPC BucketStore
[controller.git] / opendaylight / md-sal / sal-remoterpc-connector / src / test / java / org / opendaylight / controller / remote / rpc / registry / RpcRegistryTest.java
index e6793741a3ec2ef9030a2e469058ab92de93c153..5b7b7e4fdc61dfaaa2377078853b691b48b07394 100644 (file)
 package org.opendaylight.controller.remote.rpc.registry;
 
-
-import akka.actor.ActorPath;
 import akka.actor.ActorRef;
-import akka.actor.ActorSelection;
 import akka.actor.ActorSystem;
-import akka.actor.ChildActorPath;
+import akka.actor.Address;
 import akka.actor.Props;
+import akka.japi.Pair;
 import akka.testkit.JavaTestKit;
-import com.google.common.base.Predicate;
-import com.typesafe.config.ConfigFactory;
-
+import com.google.common.util.concurrent.Uninterruptibles;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
 import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Test;
+import org.opendaylight.controller.remote.rpc.RemoteRpcProviderConfig;
 import org.opendaylight.controller.remote.rpc.RouteIdentifierImpl;
-import org.opendaylight.controller.remote.rpc.registry.gossip.Messages;
+import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.AddOrUpdateRoutes;
+import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.FindRouters;
+import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.FindRoutersReply;
+import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.RemoveRoutes;
+import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.SetLocalRouter;
+import org.opendaylight.controller.remote.rpc.registry.gossip.Bucket;
+import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetAllBuckets;
+import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetAllBucketsReply;
+import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketVersions;
+import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketVersionsReply;
 import org.opendaylight.controller.sal.connector.api.RpcRouter;
-import org.opendaylight.controller.utils.ConditionalProbe;
+import org.opendaylight.controller.sal.connector.api.RpcRouter.RouteIdentifier;
 import org.opendaylight.yangtools.yang.common.QName;
-import scala.concurrent.Await;
-import scala.concurrent.Future;
+import scala.concurrent.duration.Duration;
 import scala.concurrent.duration.FiniteDuration;
 
-import javax.annotation.Nullable;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.TimeUnit;
-
-import static org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.SetLocalRouter;
-import static org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.AddOrUpdateRoutes;
-import static org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.RemoveRoutes;
-
 public class RpcRegistryTest {
 
-  private static ActorSystem node1;
-  private static ActorSystem node2;
-  private static ActorSystem node3;
-
-  private ActorRef registry1;
-  private ActorRef registry2;
-  private ActorRef registry3;
-
-  @BeforeClass
-  public static void setup() throws InterruptedException {
-    node1 = ActorSystem.create("opendaylight-rpc", ConfigFactory.load().getConfig("memberA"));
-    node2 = ActorSystem.create("opendaylight-rpc", ConfigFactory.load().getConfig("memberB"));
-    node3 = ActorSystem.create("opendaylight-rpc", ConfigFactory.load().getConfig("memberC"));
-  }
-
-  @AfterClass
-  public static void teardown() {
-    JavaTestKit.shutdownActorSystem(node1);
-    JavaTestKit.shutdownActorSystem(node2);
-    JavaTestKit.shutdownActorSystem(node3);
-    if (node1 != null)
-      node1.shutdown();
-    if (node2 != null)
-      node2.shutdown();
-    if (node3 != null)
-      node3.shutdown();
-
-  }
-
-  @Before
-  public void createRpcRegistry() throws InterruptedException {
-    registry1 = node1.actorOf(Props.create(RpcRegistry.class));
-    registry2 = node2.actorOf(Props.create(RpcRegistry.class));
-    registry3 = node3.actorOf(Props.create(RpcRegistry.class));
-  }
-
-  @After
-  public void stopRpcRegistry() throws InterruptedException {
-    if (registry1 != null)
-      node1.stop(registry1);
-    if (registry2 != null)
-      node2.stop(registry2);
-    if (registry3 != null)
-      node3.stop(registry3);
-  }
-
-  /**
-   * One node cluster.
-   * 1. Register rpc, ensure router can be found
-   * 2. Then remove rpc, ensure its deleted
-   *
-   * @throws URISyntaxException
-   * @throws InterruptedException
-   */
-  @Test
-  public void testAddRemoveRpcOnSameNode() throws URISyntaxException, InterruptedException {
-    validateSystemStartup();
-
-    final JavaTestKit mockBroker = new JavaTestKit(node1);
-
-    final ActorPath bucketStorePath = new ChildActorPath(registry1.path(), "store");
-
-    //install probe
-    final JavaTestKit probe1 = createProbeForMessage(
-        node1, bucketStorePath, Messages.BucketStoreMessages.UpdateBucket.class);
-
-    //Add rpc on node 1
-    registry1.tell(new SetLocalRouter(mockBroker.getRef()), mockBroker.getRef());
-    registry1.tell(getAddRouteMessage(), mockBroker.getRef());
-
-    //Bucket store should get an update bucket message. Updated bucket contains added rpc.
-    probe1.expectMsgClass(
-        FiniteDuration.apply(10, TimeUnit.SECONDS),
-        Messages.BucketStoreMessages.UpdateBucket.class);
-
-    //Now remove rpc
-    registry1.tell(getRemoveRouteMessage(), mockBroker.getRef());
-
-    //Bucket store should get an update bucket message. Rpc is removed in the updated bucket
-    probe1.expectMsgClass(
-        FiniteDuration.apply(10, TimeUnit.SECONDS),
-        Messages.BucketStoreMessages.UpdateBucket.class);
-
-
-  }
-
-
-  /**
-   * Three node cluster.
-   * 1. Register rpc on 1 node, ensure 2nd node gets updated
-   * 2. Remove rpc on 1 node, ensure 2nd node gets updated
-   *
-   * @throws URISyntaxException
-   * @throws InterruptedException
-   */
-  @Test
-  public void testRpcAddRemoveInCluster() throws URISyntaxException, InterruptedException {
+    private static ActorSystem node1;
+    private static ActorSystem node2;
+    private static ActorSystem node3;
 
-    validateSystemStartup();
-
-    final JavaTestKit mockBroker1 = new JavaTestKit(node1);
-
-    //install probe on node2's bucket store
-    final ActorPath bucketStorePath = new ChildActorPath(registry2.path(), "store");
-    final JavaTestKit probe2 = createProbeForMessage(
-        node2, bucketStorePath, Messages.BucketStoreMessages.UpdateRemoteBuckets.class);
+    private ActorRef registry1;
+    private ActorRef registry2;
+    private ActorRef registry3;
 
+    private int routeIdCounter = 1;
 
-    //Add rpc on node 1
-    registry1.tell(new SetLocalRouter(mockBroker1.getRef()), mockBroker1.getRef());
-    registry1.tell(getAddRouteMessage(), mockBroker1.getRef());
+    @BeforeClass
+    public static void staticSetup() throws InterruptedException {
+      RemoteRpcProviderConfig config1 = new RemoteRpcProviderConfig.Builder("memberA").build();
+      RemoteRpcProviderConfig config2 = new RemoteRpcProviderConfig.Builder("memberB").build();
+      RemoteRpcProviderConfig config3 = new RemoteRpcProviderConfig.Builder("memberC").build();
+      node1 = ActorSystem.create("opendaylight-rpc", config1.get());
+      node2 = ActorSystem.create("opendaylight-rpc", config2.get());
+      node3 = ActorSystem.create("opendaylight-rpc", config3.get());
+    }
 
-    //Bucket store on node2 should get a message to update its local copy of remote buckets
-    probe2.expectMsgClass(
-        FiniteDuration.apply(10, TimeUnit.SECONDS),
-        Messages.BucketStoreMessages.UpdateRemoteBuckets.class);
+    @AfterClass
+    public static void staticTeardown() {
+      JavaTestKit.shutdownActorSystem(node1);
+      JavaTestKit.shutdownActorSystem(node2);
+      JavaTestKit.shutdownActorSystem(node3);
+    }
 
-    //Now remove
-    registry1.tell(getRemoveRouteMessage(), mockBroker1.getRef());
+    @Before
+    public void setup() {
+        registry1 = node1.actorOf(Props.create(RpcRegistry.class));
+        registry2 = node2.actorOf(Props.create(RpcRegistry.class));
+        registry3 = node3.actorOf(Props.create(RpcRegistry.class));
+    }
 
-    //Bucket store on node2 should get a message to update its local copy of remote buckets
-    probe2.expectMsgClass(
-        FiniteDuration.apply(10, TimeUnit.SECONDS),
-        Messages.BucketStoreMessages.UpdateRemoteBuckets.class);
+    @After
+    public void teardown() {
+        if (registry1 != null) {
+            node1.stop(registry1);
+        }
+        if (registry2 != null) {
+            node2.stop(registry2);
+        }
+        if (registry3 != null) {
+            node3.stop(registry3);
+        }
+    }
 
-  }
+    /**
+     * One node cluster. 1. Register rpc, ensure router can be found 2. Then remove rpc, ensure its
+     * deleted
+     *
+     * @throws URISyntaxException
+     * @throws InterruptedException
+     */
+    @Test
+    public void testAddRemoveRpcOnSameNode() throws Exception {
 
-  /**
-   * Three node cluster.
-   * Register rpc on 2 nodes. Ensure 3rd gets updated.
-   *
-   * @throws Exception
-   */
-  @Test
-  public void testRpcAddedOnMultiNodes() throws Exception {
+        System.out.println("testAddRemoveRpcOnSameNode starting");
 
-    validateSystemStartup();
+        final JavaTestKit mockBroker = new JavaTestKit(node1);
 
-    final JavaTestKit mockBroker1 = new JavaTestKit(node1);
-    final JavaTestKit mockBroker2 = new JavaTestKit(node2);
-    final JavaTestKit mockBroker3 = new JavaTestKit(node3);
+        Address nodeAddress = node1.provider().getDefaultAddress();
 
-    registry3.tell(new SetLocalRouter(mockBroker3.getRef()), mockBroker3.getRef());
+        // Add rpc on node 1
+        registry1.tell(new SetLocalRouter(mockBroker.getRef()), mockBroker.getRef());
 
-    //install probe on node 3
-    final ActorPath bucketStorePath = new ChildActorPath(registry3.path(), "store");
-    final JavaTestKit probe3 = createProbeForMessage(
-        node3, bucketStorePath, Messages.BucketStoreMessages.UpdateRemoteBuckets.class);
+        List<RpcRouter.RouteIdentifier<?, ?, ?>> addedRouteIds = createRouteIds();
 
+        registry1.tell(new AddOrUpdateRoutes(addedRouteIds), mockBroker.getRef());
 
-    //Add rpc on node 1
-    registry1.tell(new SetLocalRouter(mockBroker1.getRef()), mockBroker1.getRef());
-    registry1.tell(getAddRouteMessage(), mockBroker1.getRef());
+        // Bucket store should get an update bucket message. Updated bucket contains added rpc.
 
-    probe3.expectMsgClass(
-        FiniteDuration.apply(10, TimeUnit.SECONDS),
-        Messages.BucketStoreMessages.UpdateRemoteBuckets.class);
+        Map<Address, Bucket> buckets = retrieveBuckets(registry1, mockBroker, nodeAddress);
+        verifyBucket(buckets.get(nodeAddress), addedRouteIds);
 
+        Map<Address, Long> versions = retrieveVersions(registry1, mockBroker);
+        Assert.assertEquals("Version for bucket " + nodeAddress, buckets.get(nodeAddress).getVersion(),
+                versions.get(nodeAddress));
 
-    //Add same rpc on node 2
-    registry2.tell(new SetLocalRouter(mockBroker2.getRef()), mockBroker2.getRef());
-    registry2.tell(getAddRouteMessage(), mockBroker2.getRef());
+        // Now remove rpc
+        registry1.tell(new RemoveRoutes(addedRouteIds), mockBroker.getRef());
 
-    probe3.expectMsgClass(
-        FiniteDuration.apply(10, TimeUnit.SECONDS),
-        Messages.BucketStoreMessages.UpdateRemoteBuckets.class);
-  }
+        // Bucket store should get an update bucket message. Rpc is removed in the updated bucket
 
-  private JavaTestKit createProbeForMessage(ActorSystem node, ActorPath subjectPath, final Class clazz) {
-    final JavaTestKit probe = new JavaTestKit(node);
+        verifyEmptyBucket(mockBroker, registry1, nodeAddress);
 
-    ConditionalProbe conditionalProbe =
-        new ConditionalProbe(probe.getRef(), new Predicate() {
-          @Override
-          public boolean apply(@Nullable Object input) {
-            return clazz.equals(input.getClass());
-          }
-        });
+        System.out.println("testAddRemoveRpcOnSameNode ending");
 
-    ActorSelection subject = node.actorSelection(subjectPath);
-    subject.tell(conditionalProbe, ActorRef.noSender());
+    }
+
+    /**
+     * Three node cluster. 1. Register rpc on 1 node, ensure 2nd node gets updated 2. Remove rpc on
+     * 1 node, ensure 2nd node gets updated
+     *
+     * @throws URISyntaxException
+     * @throws InterruptedException
+     */
+    @Test
+    public void testRpcAddRemoveInCluster() throws Exception {
 
-    return probe;
+        System.out.println("testRpcAddRemoveInCluster starting");
 
-  }
+        final JavaTestKit mockBroker1 = new JavaTestKit(node1);
+        final JavaTestKit mockBroker2 = new JavaTestKit(node2);
 
-  private void validateSystemStartup() throws InterruptedException {
+        List<RpcRouter.RouteIdentifier<?, ?, ?>> addedRouteIds = createRouteIds();
 
-    ActorPath gossiper1Path = new ChildActorPath(new ChildActorPath(registry1.path(), "store"), "gossiper");
-    ActorPath gossiper2Path = new ChildActorPath(new ChildActorPath(registry2.path(), "store"), "gossiper");
-    ActorPath gossiper3Path = new ChildActorPath(new ChildActorPath(registry3.path(), "store"), "gossiper");
+        Address node1Address = node1.provider().getDefaultAddress();
 
-    ActorSelection gossiper1 = node1.actorSelection(gossiper1Path);
-    ActorSelection gossiper2 = node2.actorSelection(gossiper2Path);
-    ActorSelection gossiper3 = node3.actorSelection(gossiper3Path);
+        // Add rpc on node 1
+        registry1.tell(new SetLocalRouter(mockBroker1.getRef()), mockBroker1.getRef());
+        registry1.tell(new AddOrUpdateRoutes(addedRouteIds), mockBroker1.getRef());
 
+        // Bucket store on node2 should get a message to update its local copy of remote buckets
 
-    if (!resolveReference(gossiper1, gossiper2, gossiper3))
-      Assert.fail("Could not find gossipers");
-  }
+        Map<Address, Bucket> buckets = retrieveBuckets(registry2, mockBroker2, node1Address);
+        verifyBucket(buckets.get(node1Address), addedRouteIds);
 
-  private Boolean resolveReference(ActorSelection... gossipers) {
+        // Now remove
+        registry1.tell(new RemoveRoutes(addedRouteIds), mockBroker1.getRef());
 
-    Boolean resolved = true;
-    for (int i = 0; i < 5; i++) {
+        // Bucket store on node2 should get a message to update its local copy of remote buckets.
+        // Wait for the bucket for node1 to be empty.
 
-      resolved = true;
-      System.out.println(System.currentTimeMillis() + " Resolving gossipers; trial #" + i);
+        verifyEmptyBucket(mockBroker2, registry2, node1Address);
 
-      for (ActorSelection gossiper : gossipers) {
-        ActorRef ref = null;
+        System.out.println("testRpcAddRemoveInCluster ending");
+    }
 
-        try {
-          Future<ActorRef> future = gossiper.resolveOne(new FiniteDuration(15000, TimeUnit.MILLISECONDS));
-          ref = Await.result(future, new FiniteDuration(10000, TimeUnit.MILLISECONDS));
-        } catch (Exception e) {
-          System.out.println("Could not find gossiper in attempt#" + i + ". Got exception " + e.getMessage());
+    private void verifyEmptyBucket(JavaTestKit testKit, ActorRef registry, Address address)
+            throws AssertionError {
+        Map<Address, Bucket> buckets;
+        int nTries = 0;
+        while(true) {
+            buckets = retrieveBuckets(registry1, testKit, address);
+
+            try {
+                verifyBucket(buckets.get(address), Collections.<RouteIdentifier<?, ?, ?>>emptyList());
+                break;
+            } catch (AssertionError e) {
+                if(++nTries >= 50) {
+                    throw e;
+                }
+            }
+
+            Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS);
         }
+    }
+
+    /**
+     * Three node cluster. Register rpc on 2 nodes. Ensure 3rd gets updated.
+     *
+     * @throws Exception
+     */
+    @Test
+    public void testRpcAddedOnMultiNodes() throws Exception {
+
+        final JavaTestKit mockBroker1 = new JavaTestKit(node1);
+        final JavaTestKit mockBroker2 = new JavaTestKit(node2);
+        final JavaTestKit mockBroker3 = new JavaTestKit(node3);
 
-        if (ref == null)
-          resolved = false;
-      }
+        registry3.tell(new SetLocalRouter(mockBroker3.getRef()), mockBroker3.getRef());
 
-      if (resolved) break;
+        // Add rpc on node 1
+        List<RpcRouter.RouteIdentifier<?, ?, ?>> addedRouteIds1 = createRouteIds();
+        registry1.tell(new SetLocalRouter(mockBroker1.getRef()), mockBroker1.getRef());
+        registry1.tell(new AddOrUpdateRoutes(addedRouteIds1), mockBroker1.getRef());
 
+        // Add rpc on node 2
+        List<RpcRouter.RouteIdentifier<?, ?, ?>> addedRouteIds2 = createRouteIds();
+        registry2.tell(new SetLocalRouter(mockBroker2.getRef()), mockBroker2.getRef());
+        registry2.tell(new AddOrUpdateRoutes(addedRouteIds2), mockBroker2.getRef());
+
+        Address node1Address = node1.provider().getDefaultAddress();
+        Address node2Address = node2.provider().getDefaultAddress();
+
+        Map<Address, Bucket> buckets = retrieveBuckets(registry3, mockBroker3, node1Address,
+                node2Address);
+
+        verifyBucket(buckets.get(node1Address), addedRouteIds1);
+        verifyBucket(buckets.get(node2Address), addedRouteIds2);
+
+        Map<Address, Long> versions = retrieveVersions(registry3, mockBroker3);
+        Assert.assertEquals("Version for bucket " + node1Address, buckets.get(node1Address).getVersion(),
+                versions.get(node1Address));
+        Assert.assertEquals("Version for bucket " + node2Address, buckets.get(node2Address).getVersion(),
+                versions.get(node2Address));
+
+        RouteIdentifier<?, ?, ?> routeID = addedRouteIds1.get(0);
+        registry3.tell(new FindRouters(routeID), mockBroker3.getRef());
+
+        FindRoutersReply reply = mockBroker3.expectMsgClass(Duration.create(3, TimeUnit.SECONDS),
+                FindRoutersReply.class);
+
+        List<Pair<ActorRef, Long>> respList = reply.getRouterWithUpdateTime();
+        Assert.assertEquals("getRouterWithUpdateTime size", 1, respList.size());
+
+        respList.get(0).first().tell("hello", ActorRef.noSender());
+        mockBroker1.expectMsgEquals(Duration.create(3, TimeUnit.SECONDS), "hello");
     }
-    return resolved;
-  }
 
-  private AddOrUpdateRoutes getAddRouteMessage() throws URISyntaxException {
-    return new AddOrUpdateRoutes(createRouteIds());
-  }
+    private Map<Address, Long> retrieveVersions(ActorRef bucketStore, JavaTestKit testKit) {
+        bucketStore.tell(new GetBucketVersions(), testKit.getRef());
+        GetBucketVersionsReply reply = testKit.expectMsgClass(Duration.create(3, TimeUnit.SECONDS),
+                GetBucketVersionsReply.class);
+        return reply.getVersions();
+    }
 
-  private RemoveRoutes getRemoveRouteMessage() throws URISyntaxException {
-    return new RemoveRoutes(createRouteIds());
-  }
+    private void verifyBucket(Bucket<RoutingTable> bucket, List<RouteIdentifier<?, ?, ?>> expRouteIds) {
+        RoutingTable table = bucket.getData();
+        Assert.assertNotNull("Bucket RoutingTable is null", table);
+        for(RouteIdentifier<?, ?, ?> r: expRouteIds) {
+            if(!table.contains(r)) {
+                Assert.fail("RoutingTable does not contain " + r + ". Actual: " + table);
+            }
+        }
 
-  private List<RpcRouter.RouteIdentifier<?, ?, ?>> createRouteIds() throws URISyntaxException {
-    QName type = new QName(new URI("/mockrpc"), "mockrpc");
-    List<RpcRouter.RouteIdentifier<?, ?, ?>> routeIds = new ArrayList<>();
-    routeIds.add(new RouteIdentifierImpl(null, type, null));
-    return routeIds;
-  }
+        Assert.assertEquals("RoutingTable size", expRouteIds.size(), table.size());
+    }
+
+    private Map<Address, Bucket> retrieveBuckets(ActorRef bucketStore, JavaTestKit testKit,
+            Address... addresses) {
+        int nTries = 0;
+        while(true) {
+            bucketStore.tell(new GetAllBuckets(), testKit.getRef());
+            GetAllBucketsReply reply = testKit.expectMsgClass(Duration.create(3, TimeUnit.SECONDS),
+                    GetAllBucketsReply.class);
+
+            Map<Address, Bucket> buckets = reply.getBuckets();
+            boolean foundAll = true;
+            for(Address addr: addresses) {
+                Bucket bucket = buckets.get(addr);
+                if(bucket  == null) {
+                    foundAll = false;
+                    break;
+                }
+            }
+
+            if(foundAll) {
+                return buckets;
+            }
+
+            if(++nTries >= 50) {
+                Assert.fail("Missing expected buckets for addresses: " + Arrays.toString(addresses)
+                        + ", Actual: " + buckets);
+            }
+
+            Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS);
+        }
+    }
 
+    @SuppressWarnings("unchecked")
+    @Test
+    public void testAddRoutesConcurrency() throws Exception {
+        final JavaTestKit testKit = new JavaTestKit(node1);
+
+        registry1.tell(new SetLocalRouter(testKit.getRef()), ActorRef.noSender());
+
+        final int nRoutes = 500;
+        final RouteIdentifier<?, ?, ?>[] added = new RouteIdentifier<?, ?, ?>[nRoutes];
+        for(int i = 0; i < nRoutes; i++) {
+            final RouteIdentifierImpl routeId = new RouteIdentifierImpl(null,
+                    new QName(new URI("/mockrpc"), "type" + i), null);
+            added[i] = routeId;
+
+            //Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
+            registry1.tell(new AddOrUpdateRoutes(Arrays.<RouteIdentifier<?, ?, ?>>asList(routeId)),
+                    ActorRef.noSender());
+        }
+
+        GetAllBuckets getAllBuckets = new GetAllBuckets();
+        FiniteDuration duration = Duration.create(3, TimeUnit.SECONDS);
+        int nTries = 0;
+        while(true) {
+            registry1.tell(getAllBuckets, testKit.getRef());
+            GetAllBucketsReply reply = testKit.expectMsgClass(duration, GetAllBucketsReply.class);
+
+            Bucket<RoutingTable> localBucket = reply.getBuckets().values().iterator().next();
+            RoutingTable table = localBucket.getData();
+            if(table != null && table.size() == nRoutes) {
+                for(RouteIdentifier<?, ?, ?> r: added) {
+                    Assert.assertEquals("RoutingTable contains " + r, true, table.contains(r));
+                }
+
+                break;
+            }
+
+            if(++nTries >= 50) {
+                Assert.fail("Expected # routes: " + nRoutes + ", Actual: " + table.size());
+            }
+
+            Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS);
+        }
+    }
+
+    private List<RpcRouter.RouteIdentifier<?, ?, ?>> createRouteIds() throws URISyntaxException {
+        QName type = new QName(new URI("/mockrpc"), "mockrpc" + routeIdCounter++);
+        List<RpcRouter.RouteIdentifier<?, ?, ?>> routeIds = new ArrayList<>();
+        routeIds.add(new RouteIdentifierImpl(null, type, null));
+        return routeIds;
+    }
 }