@Test
public void handleRejectedExecutionException() {

    // In the case that the underlying ExecutorService rejects a task submission, a permit
    // should be returned and the phaser notified
    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);
    // Submit an initial pair of tasks to ensure that the underlying work queue is full
    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());
    // Submit a third task which will be rejected by the executor service
    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());
}