test_class.file: spark-job/src/test/java/org/apache/cassandra/diff/ComparisonExecutorTest.java
test_class.identifier: ComparisonExecutorTest
test_case.body: @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());
      }
test_case.class_method_signature: ComparisonExecutorTest.handleRejectedExecutionException()
test_case.full_signature: @Test public void handleRejectedExecutionException()
test_case.identifier: handleRejectedExecutionException
test_case.invocations: incrementAndGet incrementAndGet metrics incrementAndGet listeningDecorator get getGauges submit submit assertEquals getUnarrivedParties assertEquals getValue submit countDown countDown arriveAndAwaitAdvance assertEquals getValue assertEquals get assertEquals get assertEquals get
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void handleRejectedExecutionException()
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)