class 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);

}