@Test public void handleTaskFailure() { int maxTasks = 5; MetricRegistry metrics = metrics(); ComparisonExecutor executor = new ComparisonExecutor(executor(1), maxTasks, metrics); Gauge availableSlots = metrics.getGauges().get("AvailableSlots"); final AtomicInteger successful = new AtomicInteger(0); final Consumer<Integer> onSuccess = (i) -> successful.incrementAndGet(); AtomicReference<Throwable> thrown = new AtomicReference<>(); final Consumer<Throwable> onError = thrown::set; final Phaser phaser = new Phaser(1); assertEquals(maxTasks, availableSlots.getValue()); RuntimeException toThrow = new RuntimeException("FAIL"); BlockingTask task = new BlockingTask(0, toThrow); executor.submit(task, onSuccess, onError, phaser); assertEquals(maxTasks - 1, availableSlots.getValue()); assertEquals(2, phaser.getUnarrivedParties()); task.latch.countDown(); phaser.arriveAndAwaitAdvance(); assertEquals(maxTasks, availableSlots.getValue()); assertEquals(0, successful.get()); assertEquals(toThrow, thrown.get()); }
public <T> void submit(final Callable<T> callable, final Consumer<T> onSuccess, final Consumer<Throwable> onError, final Phaser phaser) { phaser.register(); semaphore.acquireUninterruptibly(); try { Futures.addCallback(executor.submit(callable), new FutureCallback<T>() { public void onSuccess(T result) { fireThenReleaseAndArrive(onSuccess, result, phaser); } public void onFailure(Throwable t) { fireThenReleaseAndArrive(onError, t, phaser); } }, MoreExecutors.directExecutor()); } catch (RejectedExecutionException e) { fireThenReleaseAndArrive(onError, e, phaser); } }
ComparisonExecutor { public <T> void submit(final Callable<T> callable, final Consumer<T> onSuccess, final Consumer<Throwable> onError, final Phaser phaser) { phaser.register(); semaphore.acquireUninterruptibly(); try { Futures.addCallback(executor.submit(callable), new FutureCallback<T>() { public void onSuccess(T result) { fireThenReleaseAndArrive(onSuccess, result, phaser); } public void onFailure(Throwable t) { fireThenReleaseAndArrive(onError, t, phaser); } }, MoreExecutors.directExecutor()); } catch (RejectedExecutionException e) { fireThenReleaseAndArrive(onError, e, phaser); } } }
ComparisonExecutor { public <T> void submit(final Callable<T> callable, final Consumer<T> onSuccess, final Consumer<Throwable> onError, final Phaser phaser) { phaser.register(); semaphore.acquireUninterruptibly(); try { Futures.addCallback(executor.submit(callable), new FutureCallback<T>() { public void onSuccess(T result) { fireThenReleaseAndArrive(onSuccess, result, phaser); } public void onFailure(Throwable t) { fireThenReleaseAndArrive(onError, t, phaser); } }, MoreExecutors.directExecutor()); } catch (RejectedExecutionException e) { fireThenReleaseAndArrive(onError, e, phaser); } } @VisibleForTesting  ComparisonExecutor(ListeningExecutorService executor, int maxTasks, MetricRegistry metrics); }
ComparisonExecutor { public <T> void submit(final Callable<T> callable, final Consumer<T> onSuccess, final Consumer<Throwable> onError, final Phaser phaser) { phaser.register(); semaphore.acquireUninterruptibly(); try { Futures.addCallback(executor.submit(callable), new FutureCallback<T>() { public void onSuccess(T result) { fireThenReleaseAndArrive(onSuccess, result, phaser); } public void onFailure(Throwable t) { fireThenReleaseAndArrive(onError, t, phaser); } }, MoreExecutors.directExecutor()); } catch (RejectedExecutionException e) { fireThenReleaseAndArrive(onError, e, phaser); } } @VisibleForTesting  ComparisonExecutor(ListeningExecutorService executor, int maxTasks, MetricRegistry metrics); void submit(final Callable<T> callable,
                           final Consumer<T> onSuccess,
                           final Consumer<Throwable> onError,
                           final Phaser phaser); void shutdown(); }
ComparisonExecutor { public <T> void submit(final Callable<T> callable, final Consumer<T> onSuccess, final Consumer<Throwable> onError, final Phaser phaser) { phaser.register(); semaphore.acquireUninterruptibly(); try { Futures.addCallback(executor.submit(callable), new FutureCallback<T>() { public void onSuccess(T result) { fireThenReleaseAndArrive(onSuccess, result, phaser); } public void onFailure(Throwable t) { fireThenReleaseAndArrive(onError, t, phaser); } }, MoreExecutors.directExecutor()); } catch (RejectedExecutionException e) { fireThenReleaseAndArrive(onError, e, phaser); } } @VisibleForTesting  ComparisonExecutor(ListeningExecutorService executor, int maxTasks, MetricRegistry metrics); void submit(final Callable<T> callable,
                           final Consumer<T> onSuccess,
                           final Consumer<Throwable> onError,
                           final Phaser phaser); void shutdown();  }