HConnectionManager { public void operateWithFailover(Operation<?> op) throws HectorException { final Object timerToken = timer.start(op.stopWatchTagName); int retries = Math.min(op.failoverPolicy.numRetries, hostPools.size()); HClient client = null; HClientPool pool = null; boolean success = false; boolean retryable = false; Set<CassandraHost> excludeHosts = new HashSet<CassandraHost>(); while ( !success ) { try { pool = getClientFromLBPolicy(excludeHosts); client = pool.borrowClient(); if ( op.credentials != null && !op.credentials.isEmpty() && !client.isAlreadyAuthenticated(op.credentials)) { client.getCassandra().login(new AuthenticationRequest(op.credentials)); client.setAuthenticated(op.credentials); } Cassandra.Client c = client.getCassandra(op.keyspaceName); op.executeAndSetResult(c, pool.getCassandraHost()); success = true; client.updateLastSuccessTime(); timer.stop(timerToken, op.stopWatchTagName, true); break; } catch (Exception ex) { CassandraHost host = getHost(pool, client); HectorException he = exceptionsTranslator.translate(ex, host); if ( he instanceof HUnavailableException) { throw he; } else if (he instanceof HInvalidRequestException || he instanceof HCassandraInternalException) { closeClient(client); throw he; } else if (he instanceof HectorTransportException) { closeClient(client); markHostAsDown(pool.getCassandraHost()); excludeHosts.add(pool.getCassandraHost()); retryable = op.failoverPolicy.shouldRetryFor(HectorTransportException.class); monitor.incCounter(Counter.RECOVERABLE_TRANSPORT_EXCEPTIONS); } else if (he instanceof HTimedOutException ) { doTimeoutCheck(pool.getCassandraHost()); retryable = op.failoverPolicy.shouldRetryFor(HTimedOutException.class); monitor.incCounter(Counter.RECOVERABLE_TIMED_OUT_EXCEPTIONS); client.close(); } else if (he instanceof HPoolExhaustedException) { if (pool.getExhaustedTime() >= pool.getCassandraHost().getMaxExhaustedTimeBeforeMarkingAsDown()) { markHostAsDown(pool.getCassandraHost()); log.warn("Client pool for {} was exhausted for {} ms and was marked as down", pool.getCassandraHost(), pool.getExhaustedTime()); } if (hostPools.isEmpty()) { throw he; } excludeHosts.add(pool.getCassandraHost()); retryable = op.failoverPolicy.shouldRetryFor(HPoolExhaustedException.class); monitor.incCounter(Counter.POOL_EXHAUSTED); } else if ( he instanceof HPoolRecoverableException ) { retryable = op.failoverPolicy.shouldRetryFor(HPoolRecoverableException.class);; if ( hostPools.size() == 1 ) { throw he; } monitor.incCounter(Counter.POOL_EXHAUSTED); excludeHosts.add(pool.getCassandraHost()); } else { retryable = false; } if ( retries <= 0 || retryable == false) throw he; log.warn("Could not fullfill request on this host {}", client); log.warn("Exception: ", he); monitor.incCounter(Counter.SKIP_HOST_SUCCESS); sleepBetweenHostSkips(op.failoverPolicy); } finally { --retries; if ( !success ) { monitor.incCounter(op.failCounter); timer.stop(timerToken, op.stopWatchTagName, false); } releaseClient(client); client = null; } } } HConnectionManager(String clusterName, CassandraHostConfigurator cassandraHostConfigurator); void doAddNodes(); boolean addCassandraHost(CassandraHost cassandraHost); boolean removeCassandraHost(CassandraHost cassandraHost); boolean suspendCassandraHost(CassandraHost cassandraHost); boolean unsuspendCassandraHost(CassandraHost cassandraHost); Set<CassandraHost> getSuspendedCassandraHosts(); Set<CassandraHost> getHosts(); List<String> getStatusPerPool(); void operateWithFailover(Operation<?> op); HOpTimer getTimer(); void setTimer(HOpTimer timer); void addListener(String listenerName, ConnectionManagerListener listener); void removeListener(String listenerName); void removeAllListeners(); Set<CassandraHost> getDownedHosts(); Collection<HClientPool> getActivePools(); long createClock(); String getClusterName(); void shutdown(); void setCassandraHostRetryDelay(int retryDelay);  }
@Test(expected=HTimedOutException.class) public void testTimedOutOperateWithFailover() { setupClient(); FailoverPolicy fp = FailoverPolicy.ON_FAIL_TRY_ONE_NEXT_AVAILABLE; connectionManager.operateWithFailover(new TimeoutOp(fp)); }