HostSelectionWithFallback { public void initWithHosts(Map<Host, HostConnectionPool<CL>> hPools) { List<HostToken> allHostTokens = tokenSupplier.getTokens(hPools.keySet()); Map<HostToken, HostConnectionPool<CL>> tokenPoolMap = new HashMap<HostToken, HostConnectionPool<CL>>(); for (HostToken hToken : allHostTokens) { hostTokens.put(hToken.getHost(), hToken); tokenPoolMap.put(hToken, hPools.get(hToken.getHost())); } Map<HostToken, HostConnectionPool<CL>> localPools = getHostPoolsForRack(tokenPoolMap, localRack); localSelector.initWithHosts(localPools); if (localSelector.isTokenAware()) { replicationFactor.set(HostUtils.calculateReplicationFactorForDC(allHostTokens, cpConfig.getLocalDataCenter(), localRack)); } Set<String> remoteRacks = hPools.keySet().stream().map(h -> h.getRack()).filter(rack -> rack != null && !rack.equals(localRack)).collect(Collectors.toSet()); for (String rack : remoteRacks) { Map<HostToken, HostConnectionPool<CL>> dcPools = getHostPoolsForRack(tokenPoolMap, rack); HostSelectionStrategy<CL> remoteSelector = selectorFactory.vendPoolSelectionStrategy(); remoteSelector.initWithHosts(dcPools); remoteRackSelectors.put(rack, remoteSelector); } remoteRackNames.swapWithList(remoteRackSelectors.keySet()); topology.set(createTokenPoolTopology(allHostTokens)); } HostSelectionWithFallback(ConnectionPoolConfiguration config, ConnectionPoolMonitor monitor); Connection<CL> getConnection(BaseOperation<CL, ?> op, int duration, TimeUnit unit); Connection<CL> getConnectionUsingRetryPolicy(BaseOperation<CL, ?> op, int duration, TimeUnit unit, RetryPolicy retry); Collection<Connection<CL>> getConnectionsToRing(TokenRackMapper tokenRackMapper, int duration, TimeUnit unit); void initWithHosts(Map<Host, HostConnectionPool<CL>> hPools); void addHost(Host host, HostConnectionPool<CL> hostPool); void removeHost(Host host); TokenPoolTopology createTokenPoolTopology(List<HostToken> allHostTokens); TokenPoolTopology getTokenPoolTopology(); Long getTokenForKey(String key); @Override String toString();  }
@Test public void testGetConnectionsFromRingNormal() throws Exception { HostSelectionWithFallback<Integer> selection = new HostSelectionWithFallback<Integer>(cpConfig, cpMonitor); Map<Host, HostConnectionPool<Integer>> pools = new HashMap<Host, HostConnectionPool<Integer>>(); for (Host host : hosts) { poolStatus.put(host, new AtomicBoolean(true)); pools.put(host, getMockHostConnectionPool(host, poolStatus.get(host))); } selection.initWithHosts(pools); Collection<String> hostnames = runConnectionsToRingTest(selection); verifyExactly(hostnames, "h1", "h2"); }
@Test public void testGetConnectionsFromRingWhenPrimaryHostPoolInactive() throws Exception { HostSelectionWithFallback<Integer> selection = new HostSelectionWithFallback<Integer>(cpConfig, cpMonitor); Map<Host, HostConnectionPool<Integer>> pools = new HashMap<Host, HostConnectionPool<Integer>>(); for (Host host : hosts) { poolStatus.put(host, new AtomicBoolean(true)); pools.put(host, getMockHostConnectionPool(host, poolStatus.get(host))); } selection.initWithHosts(pools); poolStatus.get(h1).set(false); Collection<String> hostnames = runConnectionsToRingTest(selection); verifyPresent(hostnames, "h2"); verifyAtLeastOnePresent(hostnames, "h3", "h5"); selection.initWithHosts(pools); poolStatus.get(h1).set(true); poolStatus.get(h2).set(false); hostnames = runConnectionsToRingTest(selection); verifyPresent(hostnames, "h1"); verifyAtLeastOnePresent(hostnames, "h4", "h6"); selection.initWithHosts(pools); poolStatus.get(h1).set(false); poolStatus.get(h2).set(false); hostnames = runConnectionsToRingTest(selection); verifyAtLeastOnePresent(hostnames, "h3", "h5"); verifyAtLeastOnePresent(hostnames, "h4", "h6"); selection.initWithHosts(pools); poolStatus.get(h1).set(false); poolStatus.get(h2).set(false); poolStatus.get(h3).set(false); hostnames = runConnectionsToRingTest(selection); verifyAtLeastOnePresent(hostnames, "h4", "h6"); verifyPresent(hostnames, "h5"); selection.initWithHosts(pools); poolStatus.get(h1).set(false); poolStatus.get(h2).set(false); poolStatus.get(h3).set(false); poolStatus.get(h4).set(false); hostnames = runConnectionsToRingTest(selection); verifyExactly(hostnames, "h5", "h6"); }
@Test public void testGetConnectionsFromRingWhenHostDown() throws Exception { HostSelectionWithFallback<Integer> selection = new HostSelectionWithFallback<Integer>(cpConfig, cpMonitor); Map<Host, HostConnectionPool<Integer>> pools = new HashMap<Host, HostConnectionPool<Integer>>(); for (Host host : hosts) { poolStatus.put(host, new AtomicBoolean(true)); pools.put(host, getMockHostConnectionPool(host, poolStatus.get(host))); } selection.initWithHosts(pools); h1.setStatus(Status.Down); Collection<String> hostnames = runConnectionsToRingTest(selection); verifyPresent(hostnames, "h2"); verifyAtLeastOnePresent(hostnames, "h3", "h5"); selection.initWithHosts(pools); h1.setStatus(Status.Up); h2.setStatus(Status.Down); hostnames = runConnectionsToRingTest(selection); verifyPresent(hostnames, "h1"); verifyAtLeastOnePresent(hostnames, "h4", "h6"); selection.initWithHosts(pools); h1.setStatus(Status.Down); h2.setStatus(Status.Down); hostnames = runConnectionsToRingTest(selection); verifyAtLeastOnePresent(hostnames, "h3", "h5"); verifyAtLeastOnePresent(hostnames, "h4", "h6"); selection.initWithHosts(pools); h1.setStatus(Status.Down); h2.setStatus(Status.Down); h3.setStatus(Status.Down); hostnames = runConnectionsToRingTest(selection); verifyAtLeastOnePresent(hostnames, "h4", "h6"); verifyPresent(hostnames, "h5"); selection.initWithHosts(pools); h1.setStatus(Status.Down); h2.setStatus(Status.Down); h3.setStatus(Status.Down); h4.setStatus(Status.Down); hostnames = runConnectionsToRingTest(selection); verifyExactly(hostnames, "h5", "h6"); }