@Test
public void handleTaskFailure() {

    // Ensure that the failure callback is fired, a permit for task submission
    // returned and the phaser notified when a task throws
    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());
}