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();  }
@Test public void submitBlocksWhenMaxTasksExceeded() throws Exception { int maxTasks = 3; MetricRegistry metrics = metrics(); final ComparisonExecutor executor = new ComparisonExecutor(executor(1), maxTasks, metrics); Gauge waitingToSubmit = metrics.getGauges().get("BlockedTasks"); assertEquals(0, waitingToSubmit.getValue()); final AtomicInteger successful = new AtomicInteger(0); final Consumer<Integer> onSuccess = (i) -> successful.incrementAndGet(); final AtomicInteger failed = new AtomicInteger(0); final Consumer<Throwable> onError = (t) -> failed.incrementAndGet(); final Phaser phaser = new Phaser(1); BlockingTask[] tasks = new BlockingTask[5]; for (int i=0; i<5; i++) tasks[i] = new BlockingTask(i); executor.submit(tasks[0], onSuccess, onError, phaser); executor.submit(tasks[1], onSuccess, onError, phaser); executor.submit(tasks[2], onSuccess, onError, phaser); assertEquals(0, waitingToSubmit.getValue()); final CountDownLatch latch = new CountDownLatch(2); Thread t1 = new Thread(() -> { latch.countDown(); executor.submit(tasks[3], onSuccess, onError, phaser);}); Thread t2 = new Thread(() -> { latch.countDown(); executor.submit(tasks[4], onSuccess, onError, phaser);}); t1.start(); t2.start(); latch.await(); assertThreadWaits(t1); assertThreadWaits(t2); assertEquals(2, waitingToSubmit.getValue()); tasks[0].latch.countDown(); t1.join(); assertThreadWaits(t2); assertEquals(1, waitingToSubmit.getValue()); tasks[1].latch.countDown(); t2.join(); assertEquals(0, waitingToSubmit.getValue()); for (int i=2; i<=4; i++) tasks[i].latch.countDown(); phaser.arriveAndAwaitAdvance(); assertEquals(5, successful.get()); assertEquals(0, failed.get()); }
@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()); }
@Test public void handleUncaughtExceptionInFailureCallback() { 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 AtomicInteger failures = new AtomicInteger(0); final Consumer<Integer> onSuccess = (i) -> successful.incrementAndGet(); final Consumer<Throwable> onError = (t) -> { failures.incrementAndGet(); throw new RuntimeException("UNCAUGHT"); }; final Phaser phaser = new Phaser(1); assertEquals(maxTasks, availableSlots.getValue()); RuntimeException toThrow = new RuntimeException("FAIL"); try { onError.accept(toThrow); fail("Failure callback should throw RuntimeException"); } catch (RuntimeException e) { failures.set(0); } 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(1, failures.get()); }
@Test public void handleUncaughtExceptionInSuccessCallback() { 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 AtomicInteger failures = new AtomicInteger(0); final Consumer<Integer> onSuccess = (i) -> { successful.incrementAndGet(); throw new RuntimeException("UNCAUGHT"); }; final Consumer<Throwable> onError = (t) -> failures.incrementAndGet(); final Phaser phaser = new Phaser(1); assertEquals(maxTasks, availableSlots.getValue()); try { onSuccess.accept(0); fail("Success callback should throw RuntimeException"); } catch (RuntimeException e) { successful.set(0); } BlockingTask task = new BlockingTask(0); 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(1, successful.get()); assertEquals(0, failures.get()); }
@Test public void handleRejectedExecutionException() { int maxTasks = 5; final AtomicInteger successful = new AtomicInteger(0); final AtomicInteger failures = new AtomicInteger(0); final AtomicInteger rejections = new AtomicInteger(0); final Consumer<Integer> onSuccess = (i) -> successful.incrementAndGet(); final Consumer<Throwable> onError = (t) -> failures.incrementAndGet(); MetricRegistry metrics = metrics(); ExecutorService rejectingExecutor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1), (r, executor) -> { rejections.incrementAndGet(); throw new RejectedExecutionException("REJECTED");}); ComparisonExecutor executor = new ComparisonExecutor(MoreExecutors.listeningDecorator(rejectingExecutor), maxTasks, metrics); Gauge availableSlots = metrics.getGauges().get("AvailableSlots"); final Phaser phaser = new Phaser(1); BlockingTask t0 = new BlockingTask(0); BlockingTask t1 = new BlockingTask(1); executor.submit(t0, onSuccess, onError, phaser); executor.submit(t1, onSuccess, onError, phaser); assertEquals(3, phaser.getUnarrivedParties()); assertEquals(maxTasks - 2, availableSlots.getValue()); executor.submit(new BlockingTask(2), onSuccess, onError, phaser); t0.latch.countDown(); t1.latch.countDown(); phaser.arriveAndAwaitAdvance(); assertEquals(maxTasks, availableSlots.getValue()); assertEquals(2, successful.get()); assertEquals(1, failures.get()); assertEquals(1, rejections.get()); }