@SuppressWarnings("unchecked") @Test public void testExecuteOnExecutesNoCallbacksInPool() throws Exception { ServiceEndPoint FOO = mock(ServiceEndPoint.class); ServiceEndPoint BAR = mock(ServiceEndPoint.class); ServiceEndPoint BAZ = mock(ServiceEndPoint.class); when(_mockPool.getAllEndPoints()).thenReturn(Lists.newArrayList(FOO, BAR, BAZ)); ServiceEndPointPredicate predicate = mock(ServiceEndPointPredicate.class); when(predicate.apply(same(FOO))).thenReturn(false); when(predicate.apply(same(BAR))).thenReturn(false); when(predicate.apply(same(BAZ))).thenReturn(false); AsyncServicePool<Service> pool = newAsyncPool(MoreExecutors.newDirectExecutorService()); pool.executeOn(predicate, NEVER_RETRY, mock(ServiceCallback.class)); verify(_mockPool, never()).executeOnEndPoint(any(ServiceEndPoint.class), any(ServiceCallback.class)); }
@Override public <R> Collection<Future<R>> executeOn(ServiceEndPointPredicate predicate, final RetryPolicy retry, final ServiceCallback<S, R> callback) { Collection<Future<R>> futures = Lists.newArrayList(); Iterable<ServiceEndPoint> endPoints = _pool.getAllEndPoints(); if (Iterables.isEmpty(endPoints)) { throw new NoAvailableHostsException(String.format("No hosts discovered for service %s", _pool.getServiceName())); } for (final ServiceEndPoint endPoint : endPoints) { if (!predicate.apply(endPoint)) { continue; } Future<R> future = _executor.submit(new Callable<R>() { @Override public R call() throws Exception { Timer.Context timer = _executionTime.time(); final long start = _ticker.read(); int numAttempts = 0; try { Exception lastException; do { try { R result = _pool.executeOnEndPoint(endPoint, callback); _numExecuteSuccesses.mark(); return result; } catch (Exception e) { _numExecuteFailures.mark(); if (!_pool.isRetriableException(e)) { throw e; } lastException = e; LOG.info("Retriable exception from end point: " + endPoint, e); } } while (retry.allowRetry(++numAttempts, TimeUnit.NANOSECONDS.toMillis(_ticker.read() - start))); throw new MaxRetriesException(lastException); } finally { timer.stop(); } } }); futures.add(future); } _executeBatchSize.update(futures.size()); return futures; }
AsyncServicePool implements com.bazaarvoice.ostrich.AsyncServicePool<S> { @Override public <R> Collection<Future<R>> executeOn(ServiceEndPointPredicate predicate, final RetryPolicy retry, final ServiceCallback<S, R> callback) { Collection<Future<R>> futures = Lists.newArrayList(); Iterable<ServiceEndPoint> endPoints = _pool.getAllEndPoints(); if (Iterables.isEmpty(endPoints)) { throw new NoAvailableHostsException(String.format("No hosts discovered for service %s", _pool.getServiceName())); } for (final ServiceEndPoint endPoint : endPoints) { if (!predicate.apply(endPoint)) { continue; } Future<R> future = _executor.submit(new Callable<R>() { @Override public R call() throws Exception { Timer.Context timer = _executionTime.time(); final long start = _ticker.read(); int numAttempts = 0; try { Exception lastException; do { try { R result = _pool.executeOnEndPoint(endPoint, callback); _numExecuteSuccesses.mark(); return result; } catch (Exception e) { _numExecuteFailures.mark(); if (!_pool.isRetriableException(e)) { throw e; } lastException = e; LOG.info("Retriable exception from end point: " + endPoint, e); } } while (retry.allowRetry(++numAttempts, TimeUnit.NANOSECONDS.toMillis(_ticker.read() - start))); throw new MaxRetriesException(lastException); } finally { timer.stop(); } } }); futures.add(future); } _executeBatchSize.update(futures.size()); return futures; } }
AsyncServicePool implements com.bazaarvoice.ostrich.AsyncServicePool<S> { @Override public <R> Collection<Future<R>> executeOn(ServiceEndPointPredicate predicate, final RetryPolicy retry, final ServiceCallback<S, R> callback) { Collection<Future<R>> futures = Lists.newArrayList(); Iterable<ServiceEndPoint> endPoints = _pool.getAllEndPoints(); if (Iterables.isEmpty(endPoints)) { throw new NoAvailableHostsException(String.format("No hosts discovered for service %s", _pool.getServiceName())); } for (final ServiceEndPoint endPoint : endPoints) { if (!predicate.apply(endPoint)) { continue; } Future<R> future = _executor.submit(new Callable<R>() { @Override public R call() throws Exception { Timer.Context timer = _executionTime.time(); final long start = _ticker.read(); int numAttempts = 0; try { Exception lastException; do { try { R result = _pool.executeOnEndPoint(endPoint, callback); _numExecuteSuccesses.mark(); return result; } catch (Exception e) { _numExecuteFailures.mark(); if (!_pool.isRetriableException(e)) { throw e; } lastException = e; LOG.info("Retriable exception from end point: " + endPoint, e); } } while (retry.allowRetry(++numAttempts, TimeUnit.NANOSECONDS.toMillis(_ticker.read() - start))); throw new MaxRetriesException(lastException); } finally { timer.stop(); } } }); futures.add(future); } _executeBatchSize.update(futures.size()); return futures; } AsyncServicePool(Ticker ticker, ServicePool<S> pool, boolean shutdownPoolOnClose,
                            ExecutorService executor, boolean shutdownExecutorOnClose, MetricRegistry metrics); }
AsyncServicePool implements com.bazaarvoice.ostrich.AsyncServicePool<S> { @Override public <R> Collection<Future<R>> executeOn(ServiceEndPointPredicate predicate, final RetryPolicy retry, final ServiceCallback<S, R> callback) { Collection<Future<R>> futures = Lists.newArrayList(); Iterable<ServiceEndPoint> endPoints = _pool.getAllEndPoints(); if (Iterables.isEmpty(endPoints)) { throw new NoAvailableHostsException(String.format("No hosts discovered for service %s", _pool.getServiceName())); } for (final ServiceEndPoint endPoint : endPoints) { if (!predicate.apply(endPoint)) { continue; } Future<R> future = _executor.submit(new Callable<R>() { @Override public R call() throws Exception { Timer.Context timer = _executionTime.time(); final long start = _ticker.read(); int numAttempts = 0; try { Exception lastException; do { try { R result = _pool.executeOnEndPoint(endPoint, callback); _numExecuteSuccesses.mark(); return result; } catch (Exception e) { _numExecuteFailures.mark(); if (!_pool.isRetriableException(e)) { throw e; } lastException = e; LOG.info("Retriable exception from end point: " + endPoint, e); } } while (retry.allowRetry(++numAttempts, TimeUnit.NANOSECONDS.toMillis(_ticker.read() - start))); throw new MaxRetriesException(lastException); } finally { timer.stop(); } } }); futures.add(future); } _executeBatchSize.update(futures.size()); return futures; } AsyncServicePool(Ticker ticker, ServicePool<S> pool, boolean shutdownPoolOnClose,
                            ExecutorService executor, boolean shutdownExecutorOnClose, MetricRegistry metrics); @Override void close(); @Override Future<R> execute(final RetryPolicy retryPolicy, final ServiceCallback<S, R> callback); @Override Future<R> execute(final PartitionContext partitionContext, final RetryPolicy retryPolicy,
                                 final ServiceCallback<S, R> callback); @Override Collection<Future<R>> executeOnAll(RetryPolicy retry, ServiceCallback<S, R> callback); @Override Collection<Future<R>> executeOn(ServiceEndPointPredicate predicate, final RetryPolicy retry,
                                               final ServiceCallback<S, R> callback); @Override int getNumValidEndPoints(); @Override int getNumBadEndPoints(); }
AsyncServicePool implements com.bazaarvoice.ostrich.AsyncServicePool<S> { @Override public <R> Collection<Future<R>> executeOn(ServiceEndPointPredicate predicate, final RetryPolicy retry, final ServiceCallback<S, R> callback) { Collection<Future<R>> futures = Lists.newArrayList(); Iterable<ServiceEndPoint> endPoints = _pool.getAllEndPoints(); if (Iterables.isEmpty(endPoints)) { throw new NoAvailableHostsException(String.format("No hosts discovered for service %s", _pool.getServiceName())); } for (final ServiceEndPoint endPoint : endPoints) { if (!predicate.apply(endPoint)) { continue; } Future<R> future = _executor.submit(new Callable<R>() { @Override public R call() throws Exception { Timer.Context timer = _executionTime.time(); final long start = _ticker.read(); int numAttempts = 0; try { Exception lastException; do { try { R result = _pool.executeOnEndPoint(endPoint, callback); _numExecuteSuccesses.mark(); return result; } catch (Exception e) { _numExecuteFailures.mark(); if (!_pool.isRetriableException(e)) { throw e; } lastException = e; LOG.info("Retriable exception from end point: " + endPoint, e); } } while (retry.allowRetry(++numAttempts, TimeUnit.NANOSECONDS.toMillis(_ticker.read() - start))); throw new MaxRetriesException(lastException); } finally { timer.stop(); } } }); futures.add(future); } _executeBatchSize.update(futures.size()); return futures; } AsyncServicePool(Ticker ticker, ServicePool<S> pool, boolean shutdownPoolOnClose,
                            ExecutorService executor, boolean shutdownExecutorOnClose, MetricRegistry metrics); @Override void close(); @Override Future<R> execute(final RetryPolicy retryPolicy, final ServiceCallback<S, R> callback); @Override Future<R> execute(final PartitionContext partitionContext, final RetryPolicy retryPolicy,
                                 final ServiceCallback<S, R> callback); @Override Collection<Future<R>> executeOnAll(RetryPolicy retry, ServiceCallback<S, R> callback); @Override Collection<Future<R>> executeOn(ServiceEndPointPredicate predicate, final RetryPolicy retry,
                                               final ServiceCallback<S, R> callback); @Override int getNumValidEndPoints(); @Override int getNumBadEndPoints();  }