ConnectionPoolImpl implements ConnectionPool<CL>, TopologyView { @Override public Future<Boolean> start() throws DynoException { if (started.get()) { return getEmptyFutureTask(false); } HostSupplier hostSupplier = cpConfiguration.getHostSupplier(); if (hostSupplier == null) { throw new DynoException("Host supplier not configured!"); } HostStatusTracker hostStatus = hostsUpdater.refreshHosts(); cpMonitor.setHostCount(hostStatus.getHostCount()); Collection<Host> hostsUp = hostStatus.getActiveHosts(); if (hostsUp == null || hostsUp.isEmpty()) { throw new NoAvailableHostsException("No available hosts when starting connection pool"); } final ExecutorService threadPool = Executors.newFixedThreadPool(Math.max(10, hostsUp.size())); final List<Future<Void>> futures = new ArrayList<Future<Void>>(); for (final Host host : hostsUp) { futures.add(threadPool.submit(new Callable<Void>() { @Override public Void call() throws Exception { addHost(host, false); return null; } })); } try { for (Future<Void> future : futures) { try { future.get(); } catch (InterruptedException e) { } catch (ExecutionException e) { throw new RuntimeException(e); } } } finally { threadPool.shutdownNow(); } boolean success = started.compareAndSet(false, true); if (success) { idling.set(false); idleThreadPool.shutdownNow(); selectionStrategy = initSelectionStrategy(); cpHealthTracker.start(); connPoolThreadPool.scheduleWithFixedDelay(new Runnable() { @Override public void run() { try { HostStatusTracker hostStatus = hostsUpdater.refreshHosts(); cpMonitor.setHostCount(hostStatus.getHostCount()); Logger.debug(hostStatus.toString()); updateHosts(hostStatus.getActiveHosts(), hostStatus.getInactiveHosts()); } catch (Throwable throwable) { Logger.error("Failed to update hosts cache", throwable); } } }, 15 * 1000, 30 * 1000, TimeUnit.MILLISECONDS); MonitorConsole.getInstance().registerConnectionPool(this); registerMonitorConsoleMBean(MonitorConsole.getInstance()); } return getEmptyFutureTask(true); } ConnectionPoolImpl(ConnectionFactory<CL> cFactory, ConnectionPoolConfiguration cpConfig,
                              ConnectionPoolMonitor cpMon);  ConnectionPoolImpl(ConnectionFactory<CL> cFactory, ConnectionPoolConfiguration cpConfig,
                              ConnectionPoolMonitor cpMon, Type type); String getName(); ConnectionPoolMonitor getMonitor(); ConnectionPoolHealthTracker<CL> getHealthTracker(); @Override boolean isIdle(); @Override boolean addHost(Host host); boolean addHost(Host host, boolean refreshLoadBalancer); @Override boolean removeHost(Host host); @Override boolean isHostUp(Host host); @Override boolean hasHost(Host host); @Override List<HostConnectionPool<CL>> getActivePools(); @Override List<HostConnectionPool<CL>> getPools(); @Override Future<Boolean> updateHosts(Collection<Host> hostsUp, Collection<Host> hostsDown); @Override HostConnectionPool<CL> getHostPool(Host host); @Override OperationResult<R> executeWithFailover(Operation<CL, R> op); @Override Collection<OperationResult<R>> executeWithRing(TokenRackMapper tokenRackMapper, Operation<CL, R> op); Connection<CL> getConnectionForOperation(BaseOperation<CL, R> baseOperation); @Override void shutdown(); @Override Future<Boolean> start(); @Override void idle(); @Override ConnectionPoolConfiguration getConfiguration(); @Override ListenableFuture<OperationResult<R>> executeAsync(AsyncOperation<CL, R> op); TokenPoolTopology getTopology(); @Override Map<String, List<TokenPoolTopology.TokenStatus>> getTopologySnapshot(); @Override Long getTokenForKey(String key);  }
@Test public void testHostEvictionDueToErrorRates() throws Exception { ErrorRateMonitorConfigImpl errConfig = new ErrorRateMonitorConfigImpl(); errConfig.checkFrequency = 1; errConfig.window = 1; errConfig.suppressWindow = 60; errConfig.addThreshold(10, 1, 100); final AtomicReference<String> badHost = new AtomicReference<String>(); final ConnectionFactory<TestClient> badConnectionFactory = new ConnectionFactory<TestClient>() { @Override public Connection<TestClient> createConnection(final HostConnectionPool<TestClient> pool) throws DynoConnectException { return new TestConnection(pool) { @Override public <R> OperationResult<R> execute(Operation<TestClient, R> op) throws DynoException { if (pool.getHost().getHostAddress().equals(badHost.get())) { throw new FatalConnectionException("Fail for bad host"); } return super.execute(op); } }; } @Override public Connection<TestClient> createConnectionWithDataStore(HostConnectionPool<TestClient> pool) throws DynoConnectException { return null; } @Override public Connection<TestClient> createConnectionWithConsistencyLevel(HostConnectionPool<TestClient> pool, String consistency) throws DynoConnectException { return null; } }; final ConnectionPoolImpl<TestClient> pool = new ConnectionPoolImpl<TestClient>(badConnectionFactory, cpConfig, cpMonitor); hostSupplierHosts.add(host1); hostSupplierHosts.add(host2); hostSupplierHosts.add(host3); pool.start(); final Callable<Void> testLogic = new Callable<Void>() { @Override public Void call() throws Exception { Thread.sleep(2000); badHost.set("host2"); Thread.sleep(2000); return null; } }; runTest(pool, testLogic); Assert.assertTrue("Total ops: " + client.ops.get(), client.ops.get() > 0); Assert.assertTrue("Total errors: " + cpMonitor.getOperationFailureCount(), cpMonitor.getOperationFailureCount() > 0); Assert.assertEquals(3 * cpConfig.getMaxConnsPerHost(), cpMonitor.getConnectionCreatedCount()); Assert.assertEquals(0, cpMonitor.getConnectionCreateFailedCount()); Assert.assertEquals(3 * cpConfig.getMaxConnsPerHost(), cpMonitor.getConnectionClosedCount()); Assert.assertEquals(client.ops.get() + cpMonitor.getOperationFailureCount(), cpMonitor.getConnectionBorrowedCount()); Assert.assertEquals(client.ops.get() + cpMonitor.getOperationFailureCount(), cpMonitor.getConnectionReturnedCount()); checkHostStats(host1); checkHostStats(host3); HostConnectionStats h2Stats = cpMonitor.getHostStats().get(host2); Assert.assertEquals(cpMonitor.getOperationFailureCount(), h2Stats.getOperationErrorCount()); }
@Test(expected = NoAvailableHostsException.class) public void testHostsDownDuringStartup() { final ConnectionPoolImpl<TestClient> pool = new ConnectionPoolImpl<TestClient>(connFactory, cpConfig, cpMonitor); hostSupplierHosts.add(new HostBuilder().setHostname("host1_down").setPort(8080).setRack("localRack").setStatus(Status.Down).createHost()); hostSupplierHosts.add(new HostBuilder().setHostname("host2_down").setPort(8080).setRack("localRack").setStatus(Status.Down).createHost()); hostSupplierHosts.add(new HostBuilder().setHostname("host3_down").setPort(8080).setRack("localRack").setStatus(Status.Down).createHost()); pool.start(); }