AsyncServicePool implements com.bazaarvoice.ostrich.AsyncServicePool<S> { @Override public <R> Collection<Future<R>> executeOnAll(RetryPolicy retry, ServiceCallback<S, R> callback) { return executeOn(ALL_END_POINTS, retry, callback); } 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();  }
@SuppressWarnings("unchecked") @Test public void testExecuteAllSubmitsMultipleCallablesToExecutor() { List<ServiceEndPoint> endPoints = Lists.newArrayList( mock(ServiceEndPoint.class), mock(ServiceEndPoint.class), mock(ServiceEndPoint.class) ); when(_mockPool.getAllEndPoints()).thenReturn(endPoints); AsyncServicePool<Service> pool = newAsyncPool(); pool.executeOnAll(NEVER_RETRY, mock(ServiceCallback.class)); verify(_mockExecutor, times(endPoints.size())).submit(any(Callable.class)); }
@SuppressWarnings("unchecked") @Test public void testExecuteAllExecutesCallbacksInPool() 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)); AsyncServicePool<Service> pool = newAsyncPool(MoreExecutors.newDirectExecutorService()); ServiceCallback<Service, Void> callback = (ServiceCallback<Service, Void>) mock(ServiceCallback.class); pool.executeOnAll(NEVER_RETRY, callback); verify(_mockPool).executeOnEndPoint(same(FOO), same(callback)); verify(_mockPool).executeOnEndPoint(same(BAR), same(callback)); verify(_mockPool).executeOnEndPoint(same(BAZ), same(callback)); }
@SuppressWarnings("unchecked") @Test public void testExecuteAllReturnsValueInFuture() 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)); when(_mockPool.executeOnEndPoint(same(FOO), any(ServiceCallback.class))).thenReturn("FOO"); when(_mockPool.executeOnEndPoint(same(BAR), any(ServiceCallback.class))).thenReturn("BAR"); when(_mockPool.executeOnEndPoint(same(BAZ), any(ServiceCallback.class))).thenReturn("BAZ"); AsyncServicePool<Service> pool = newAsyncPool(MoreExecutors.newDirectExecutorService()); Collection<Future<String>> futures = pool.executeOnAll(NEVER_RETRY, mock(ServiceCallback.class)); assertEquals(3, futures.size()); Set<String> results = Sets.newHashSet(); for (Future<String> future : futures) { results.add(future.get()); } assertEquals(Sets.newHashSet("FOO", "BAR", "BAZ"), results); }
@SuppressWarnings("unchecked") @Test public void testExecuteAllWrapsNonRetriableExceptionInFuture() throws Exception { RuntimeException exception = mock(RuntimeException.class); when(_mockPool.getAllEndPoints()).thenReturn(Lists.newArrayList(mock(ServiceEndPoint.class))); when(_mockPool.executeOnEndPoint(any(ServiceEndPoint.class), any(ServiceCallback.class))).thenThrow(exception); when(_mockPool.isRetriableException(any(Exception.class))).thenReturn(false); AsyncServicePool<Service> pool = newAsyncPool(MoreExecutors.newDirectExecutorService()); Collection<Future<Void>> futures = pool.executeOnAll(NEVER_RETRY, mock(ServiceCallback.class)); assertEquals(1, futures.size()); Future<Void> future = futures.iterator().next(); try { future.get(10, TimeUnit.SECONDS); fail(); } catch(ExecutionException e) { assertSame(exception, e.getCause()); } }
@SuppressWarnings("unchecked") @Test public void testExecuteAllPropagatesMaxRetriesExceptionWhenOutOfRetries() throws Exception { IllegalArgumentException exception = mock(IllegalArgumentException.class); when(_mockPool.getAllEndPoints()).thenReturn(Lists.newArrayList(mock(ServiceEndPoint.class))); when(_mockPool.executeOnEndPoint(any(ServiceEndPoint.class), any(ServiceCallback.class))).thenThrow(exception); when(_mockPool.isRetriableException(any(Exception.class))).thenReturn(true); AsyncServicePool<Service> pool = newAsyncPool(MoreExecutors.newDirectExecutorService()); Collection<Future<Void>> futures = pool.executeOnAll(NEVER_RETRY, mock(ServiceCallback.class)); assertEquals(1, futures.size()); Future<Void> future = futures.iterator().next(); try { future.get(10, TimeUnit.SECONDS); fail(); } catch(ExecutionException e) { assertTrue(e.getCause() instanceof MaxRetriesException); MaxRetriesException mre = (MaxRetriesException) e.getCause(); assertTrue(mre.getCause() instanceof IllegalArgumentException); } }
@SuppressWarnings("unchecked") @Test public void testRetriableExceptionIsRetried() throws Exception { when(_mockPool.getAllEndPoints()).thenReturn(Lists.newArrayList(mock(ServiceEndPoint.class))); when(_mockPool.executeOnEndPoint(any(ServiceEndPoint.class), any(ServiceCallback.class))) .thenThrow(RuntimeException.class).thenReturn(null); when(_mockPool.isRetriableException(any(Exception.class))).thenReturn(true); AsyncServicePool<Service> pool = newAsyncPool(MoreExecutors.newDirectExecutorService()); RetryPolicy retry = mock(RetryPolicy.class); when(retry.allowRetry(anyInt(), anyLong())).thenReturn(true); Collection<Future<Void>> futures = pool.executeOnAll(retry, mock(ServiceCallback.class)); assertEquals(1, futures.size()); Future<Void> future = futures.iterator().next(); future.get(10, TimeUnit.SECONDS); }