ConsistentHashRoutingService extends ServerSetRoutingService { @Override public void startService() { super.startService(); this.hashedWheelTimer.start(); } ConsistentHashRoutingService(ServerSetWatcher serverSetWatcher,
                                 int numReplicas,
                                 int blackoutSeconds,
                                 StatsReceiver statsReceiver); @Deprecated static ConsistentHashRoutingService of(ServerSetWatcher serverSetWatcher, int numReplicas); static Builder newBuilder(); @Override void startService(); @Override void stopService(); @Override Set<SocketAddress> getHosts(); @Override SocketAddress getHost(String key, RoutingContext rContext); @Override void removeHost(SocketAddress host, Throwable reason);  }
@Test(timeout = 60000) public void testPerformServerSetChangeOnServerSet() throws Exception { TestServerSetWatcher serverSetWatcher = new TestServerSetWatcher(); ConsistentHashRoutingService routingService = new ConsistentHashRoutingService( serverSetWatcher, 997, Integer.MAX_VALUE, NullStatsReceiver.get()); int basePort = 3180; int numHosts = 4; Set<DLSocketAddress> addresses1 = Sets.newConcurrentHashSet(); Set<DLSocketAddress> addresses2 = Sets.newConcurrentHashSet(); Set<DLSocketAddress> addresses3 = Sets.newConcurrentHashSet(); for (int i = 0; i < numHosts; i++) { InetSocketAddress inetAddress = new InetSocketAddress("127.0.0.1", basePort + i); DLSocketAddress dsa = new DLSocketAddress(i, inetAddress); addresses1.add(dsa); } for (int i = 0; i < numHosts; i++) { InetSocketAddress inetAddress = new InetSocketAddress("127.0.0.1", basePort + numHosts + i); DLSocketAddress dsa = new DLSocketAddress(i+2, inetAddress); addresses2.add(dsa); } for (int i = 0; i < numHosts; i++) { InetSocketAddress inetAddress = new InetSocketAddress("127.0.0.1", basePort + 10 + i); DLSocketAddress dsa = new DLSocketAddress(i, inetAddress); addresses3.add(dsa); } final List<SocketAddress> leftAddresses = Lists.newArrayList(); final List<SocketAddress> joinAddresses = Lists.newArrayList(); RoutingService.RoutingListener routingListener = new RoutingService.RoutingListener() { @Override public void onServerLeft(SocketAddress address) { synchronized (leftAddresses) { leftAddresses.add(address); leftAddresses.notifyAll(); } } @Override public void onServerJoin(SocketAddress address) { synchronized (joinAddresses) { joinAddresses.add(address); joinAddresses.notifyAll(); } } }; routingService.registerListener(routingListener); serverSetWatcher.notifyChanges(ImmutableSet.copyOf(addresses1)); routingService.startService(); synchronized (joinAddresses) { while (joinAddresses.size() < numHosts) { joinAddresses.wait(); } } synchronized (joinAddresses) { assertEquals(numHosts, joinAddresses.size()); } synchronized (leftAddresses) { assertEquals(0, leftAddresses.size()); } assertEquals(numHosts, routingService.shardId2Address.size()); assertEquals(numHosts, routingService.address2ShardId.size()); for (int i = 0; i < numHosts; i++) { InetSocketAddress inetAddress = new InetSocketAddress("127.0.0.1", basePort + i); assertTrue(routingService.address2ShardId.containsKey(inetAddress)); int shardId = routingService.address2ShardId.get(inetAddress); assertEquals(i, shardId); SocketAddress sa = routingService.shardId2Address.get(shardId); assertNotNull(sa); assertEquals(inetAddress, sa); } serverSetWatcher.notifyChanges(ImmutableSet.copyOf(addresses2)); synchronized (joinAddresses) { while (joinAddresses.size() < numHosts + 2) { joinAddresses.wait(); } } synchronized (leftAddresses) { while (leftAddresses.size() < 2) { leftAddresses.wait(); } } assertEquals(numHosts + 2, routingService.shardId2Address.size()); assertEquals(numHosts + 2, routingService.address2ShardId.size()); for (int i = 0; i < 2; i++) { InetSocketAddress inetAddress = new InetSocketAddress("127.0.0.1", basePort + i); assertTrue(routingService.address2ShardId.containsKey(inetAddress)); int shardId = routingService.address2ShardId.get(inetAddress); assertEquals(i, shardId); SocketAddress sa = routingService.shardId2Address.get(shardId); assertNotNull(sa); assertEquals(inetAddress, sa); } for (int i = 0; i < numHosts; i++) { InetSocketAddress inetAddress = new InetSocketAddress("127.0.0.1", basePort + numHosts + i); assertTrue(routingService.address2ShardId.containsKey(inetAddress)); int shardId = routingService.address2ShardId.get(inetAddress); assertEquals(i+2, shardId); SocketAddress sa = routingService.shardId2Address.get(shardId); assertNotNull(sa); assertEquals(inetAddress, sa); } serverSetWatcher.notifyChanges(ImmutableSet.copyOf(addresses3)); synchronized (joinAddresses) { while (joinAddresses.size() < numHosts + 2 + numHosts) { joinAddresses.wait(); } } synchronized (leftAddresses) { while (leftAddresses.size() < 2 + numHosts) { leftAddresses.wait(); } } assertEquals(numHosts + 2, routingService.shardId2Address.size()); assertEquals(numHosts + 2, routingService.address2ShardId.size()); for (int i = 0; i < numHosts; i++) { InetSocketAddress inetAddress = new InetSocketAddress("127.0.0.1", basePort + 10 + i); assertTrue(routingService.address2ShardId.containsKey(inetAddress)); int shardId = routingService.address2ShardId.get(inetAddress); assertEquals(i, shardId); SocketAddress sa = routingService.shardId2Address.get(shardId); assertNotNull(sa); assertEquals(inetAddress, sa); } for (int i = 0; i < 2; i++) { InetSocketAddress inetAddress = new InetSocketAddress("127.0.0.1", basePort + numHosts + 2 + i); assertTrue(routingService.address2ShardId.containsKey(inetAddress)); int shardId = routingService.address2ShardId.get(inetAddress); assertEquals(numHosts + i, shardId); SocketAddress sa = routingService.shardId2Address.get(shardId); assertNotNull(sa); assertEquals(inetAddress, sa); } }
@Test(timeout = 60000) public void testPerformServerSetChangeOnServerSet() throws Exception { TestServerSetWatcher serverSetWatcher = new TestServerSetWatcher(); ConsistentHashRoutingService routingService = new ConsistentHashRoutingService( serverSetWatcher, 997, Integer.MAX_VALUE, NullStatsReceiver.get()); int basePort = 3180; int numHosts = 4; Set<DLSocketAddress> addresses1 = Sets.newConcurrentHashSet(); Set<DLSocketAddress> addresses2 = Sets.newConcurrentHashSet(); Set<DLSocketAddress> addresses3 = Sets.newConcurrentHashSet(); for (int i = 0; i < numHosts; i++) { InetSocketAddress inetAddress = new InetSocketAddress("127.0.0.1", basePort + i); DLSocketAddress dsa = new DLSocketAddress(i, inetAddress); addresses1.add(dsa); } for (int i = 0; i < numHosts; i++) { InetSocketAddress inetAddress = new InetSocketAddress("127.0.0.1", basePort + numHosts + i); DLSocketAddress dsa = new DLSocketAddress(i + 2, inetAddress); addresses2.add(dsa); } for (int i = 0; i < numHosts; i++) { InetSocketAddress inetAddress = new InetSocketAddress("127.0.0.1", basePort + 10 + i); DLSocketAddress dsa = new DLSocketAddress(i, inetAddress); addresses3.add(dsa); } final List<SocketAddress> leftAddresses = Lists.newArrayList(); final List<SocketAddress> joinAddresses = Lists.newArrayList(); RoutingService.RoutingListener routingListener = new RoutingService.RoutingListener() { @Override public void onServerLeft(SocketAddress address) { synchronized (leftAddresses) { leftAddresses.add(address); leftAddresses.notifyAll(); } } @Override public void onServerJoin(SocketAddress address) { synchronized (joinAddresses) { joinAddresses.add(address); joinAddresses.notifyAll(); } } }; routingService.registerListener(routingListener); serverSetWatcher.notifyChanges(ImmutableSet.copyOf(addresses1)); routingService.startService(); synchronized (joinAddresses) { while (joinAddresses.size() < numHosts) { joinAddresses.wait(); } } synchronized (joinAddresses) { assertEquals(numHosts, joinAddresses.size()); } synchronized (leftAddresses) { assertEquals(0, leftAddresses.size()); } assertEquals(numHosts, routingService.shardId2Address.size()); assertEquals(numHosts, routingService.address2ShardId.size()); for (int i = 0; i < numHosts; i++) { InetSocketAddress inetAddress = new InetSocketAddress("127.0.0.1", basePort + i); assertTrue(routingService.address2ShardId.containsKey(inetAddress)); int shardId = routingService.address2ShardId.get(inetAddress); assertEquals(i, shardId); SocketAddress sa = routingService.shardId2Address.get(shardId); assertNotNull(sa); assertEquals(inetAddress, sa); } serverSetWatcher.notifyChanges(ImmutableSet.copyOf(addresses2)); synchronized (joinAddresses) { while (joinAddresses.size() < numHosts + 2) { joinAddresses.wait(); } } synchronized (leftAddresses) { while (leftAddresses.size() < 2) { leftAddresses.wait(); } } assertEquals(numHosts + 2, routingService.shardId2Address.size()); assertEquals(numHosts + 2, routingService.address2ShardId.size()); for (int i = 0; i < 2; i++) { InetSocketAddress inetAddress = new InetSocketAddress("127.0.0.1", basePort + i); assertTrue(routingService.address2ShardId.containsKey(inetAddress)); int shardId = routingService.address2ShardId.get(inetAddress); assertEquals(i, shardId); SocketAddress sa = routingService.shardId2Address.get(shardId); assertNotNull(sa); assertEquals(inetAddress, sa); } for (int i = 0; i < numHosts; i++) { InetSocketAddress inetAddress = new InetSocketAddress("127.0.0.1", basePort + numHosts + i); assertTrue(routingService.address2ShardId.containsKey(inetAddress)); int shardId = routingService.address2ShardId.get(inetAddress); assertEquals(i + 2, shardId); SocketAddress sa = routingService.shardId2Address.get(shardId); assertNotNull(sa); assertEquals(inetAddress, sa); } serverSetWatcher.notifyChanges(ImmutableSet.copyOf(addresses3)); synchronized (joinAddresses) { while (joinAddresses.size() < numHosts + 2 + numHosts) { joinAddresses.wait(); } } synchronized (leftAddresses) { while (leftAddresses.size() < 2 + numHosts) { leftAddresses.wait(); } } assertEquals(numHosts + 2, routingService.shardId2Address.size()); assertEquals(numHosts + 2, routingService.address2ShardId.size()); for (int i = 0; i < numHosts; i++) { InetSocketAddress inetAddress = new InetSocketAddress("127.0.0.1", basePort + 10 + i); assertTrue(routingService.address2ShardId.containsKey(inetAddress)); int shardId = routingService.address2ShardId.get(inetAddress); assertEquals(i, shardId); SocketAddress sa = routingService.shardId2Address.get(shardId); assertNotNull(sa); assertEquals(inetAddress, sa); } for (int i = 0; i < 2; i++) { InetSocketAddress inetAddress = new InetSocketAddress("127.0.0.1", basePort + numHosts + 2 + i); assertTrue(routingService.address2ShardId.containsKey(inetAddress)); int shardId = routingService.address2ShardId.get(inetAddress); assertEquals(numHosts + i, shardId); SocketAddress sa = routingService.shardId2Address.get(shardId); assertNotNull(sa); assertEquals(inetAddress, sa); } }