@Test
public void submitBlocksWhenMaxTasksExceeded() throws Exception {

    // submit maxTasks, then assert that further submission blocks until tasks are processed
    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);
    // Ensure that the submission itself does not block before the max number of tasks are submitted
    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());
    // Now submit another pair of tasks which should block as the executor is fully occupied
    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();
    // wait for both to attempt submission
    latch.await();
    assertThreadWaits(t1);
    assertThreadWaits(t2);
    assertEquals(2, waitingToSubmit.getValue());
    // Let the first waiting task complete, which should allow t1 to complete its submission
    tasks[0].latch.countDown();
    t1.join();
    // the second submission should still be waiting on a slot
    assertThreadWaits(t2);
    assertEquals(1, waitingToSubmit.getValue());
    // Let another task complete, allowing t2 to complete its submission
    tasks[1].latch.countDown();
    t2.join();
    assertEquals(0, waitingToSubmit.getValue());
    // Let all tasks complete, wait for them to do so then verify counters
    for (int i=2; i<=4; i++)
        tasks[i].latch.countDown();
    phaser.arriveAndAwaitAdvance();
    assertEquals(5, successful.get());
    assertEquals(0, failed.get());
}