test_class.file: spark-job/src/test/java/org/apache/cassandra/diff/ComparisonExecutorTest.java
test_class.identifier: ComparisonExecutorTest
test_case.body: @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());
      }
test_case.class_method_signature: ComparisonExecutorTest.submitBlocksWhenMaxTasksExceeded()
test_case.full_signature: @Test public void submitBlocksWhenMaxTasksExceeded()
test_case.identifier: submitBlocksWhenMaxTasksExceeded
test_case.invocations: metrics executor get getGauges assertEquals getValue incrementAndGet incrementAndGet submit submit submit assertEquals getValue countDown submit countDown submit start start await assertThreadWaits assertThreadWaits assertEquals getValue countDown join assertThreadWaits assertEquals getValue countDown join assertEquals getValue countDown arriveAndAwaitAdvance assertEquals get assertEquals get
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void submitBlocksWhenMaxTasksExceeded()
focal_class.declarator: executor
focal_class.modifier: private final
focal_class.original_string: private final ListeningExecutorService executor;
focal_class.type: ListeningExecutorService
focal_class.var_name: executor
focal_class.declarator: semaphore
focal_class.modifier: private final
focal_class.original_string: private final Semaphore semaphore;
focal_class.type: Semaphore
focal_class.var_name: semaphore
focal_class.file: spark-job/src/main/java/org/apache/cassandra/diff/ComparisonExecutor.java
focal_class.identifier: ComparisonExecutor
focal_class.class_method_signature: ComparisonExecutor.newExecutor(int maxConcurrentTasks, MetricRegistry metricRegistry)
focal_class.full_signature: static ComparisonExecutor newExecutor(int maxConcurrentTasks, MetricRegistry metricRegistry)
focal_class.identifier: newExecutor
focal_class.modifiers: static
focal_class.parameters: (int maxConcurrentTasks, MetricRegistry metricRegistry)
focal_class.return: ComparisonExecutor
focal_class.signature: ComparisonExecutor newExecutor(int maxConcurrentTasks, MetricRegistry metricRegistry)
focal_class.class_method_signature: ComparisonExecutor.ComparisonExecutor(ListeningExecutorService executor, int maxTasks, MetricRegistry metrics)
focal_class.full_signature: @VisibleForTesting  ComparisonExecutor(ListeningExecutorService executor, int maxTasks, MetricRegistry metrics)
focal_class.identifier: ComparisonExecutor
focal_class.modifiers: @VisibleForTesting
focal_class.parameters: (ListeningExecutorService executor, int maxTasks, MetricRegistry metrics)
focal_class.signature:  ComparisonExecutor(ListeningExecutorService executor, int maxTasks, MetricRegistry metrics)
focal_class.class_method_signature: ComparisonExecutor.submit(final Callable<T> callable,
                             final Consumer<T> onSuccess,
                             final Consumer<Throwable> onError,
                             final Phaser phaser)
focal_class.full_signature: public void submit(final Callable<T> callable,
                             final Consumer<T> onSuccess,
                             final Consumer<Throwable> onError,
                             final Phaser phaser)
focal_class.identifier: submit
focal_class.modifiers: public
focal_class.parameters: (final Callable<T> callable,
                             final Consumer<T> onSuccess,
                             final Consumer<Throwable> onError,
                             final Phaser phaser)
focal_class.return: void
focal_class.signature: void submit(final Callable<T> callable,
                             final Consumer<T> onSuccess,
                             final Consumer<Throwable> onError,
                             final Phaser phaser)
focal_class.class_method_signature: ComparisonExecutor.fireThenReleaseAndArrive(Consumer<T> callback, T argument, Phaser phaser)
focal_class.full_signature: private void fireThenReleaseAndArrive(Consumer<T> callback, T argument, Phaser phaser)
focal_class.identifier: fireThenReleaseAndArrive
focal_class.modifiers: private
focal_class.parameters: (Consumer<T> callback, T argument, Phaser phaser)
focal_class.return: void
focal_class.signature: void fireThenReleaseAndArrive(Consumer<T> callback, T argument, Phaser phaser)
focal_class.class_method_signature: ComparisonExecutor.shutdown()
focal_class.full_signature: public void shutdown()
focal_class.identifier: shutdown
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void shutdown()
focal_method.body: 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);
          }
  
      }
focal_method.class_method_signature: ComparisonExecutor.submit(final Callable<T> callable,
                             final Consumer<T> onSuccess,
                             final Consumer<Throwable> onError,
                             final Phaser phaser)
focal_method.full_signature: public void submit(final Callable<T> callable,
                             final Consumer<T> onSuccess,
                             final Consumer<Throwable> onError,
                             final Phaser phaser)
focal_method.identifier: submit
focal_method.invocations: register acquireUninterruptibly addCallback submit fireThenReleaseAndArrive fireThenReleaseAndArrive directExecutor fireThenReleaseAndArrive
focal_method.modifiers: public
focal_method.parameters: (final Callable<T> callable,
                             final Consumer<T> onSuccess,
                             final Consumer<Throwable> onError,
                             final Phaser phaser)
focal_method.return: void
focal_method.signature: void submit(final Callable<T> callable,
                             final Consumer<T> onSuccess,
                             final Consumer<Throwable> onError,
                             final Phaser phaser)