@Test
public void handleUncaughtExceptionInSuccessCallback() {

    // Ensure that if the success callback throws, a permit for submission is
    // still returned and the phaser notified
    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) {
        // expected - reset failure count
        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());
}