@Override
  public void execute(Runnable command) {
    _delegateExecutor.execute(toAccountingRunnable(command));
  }
@Test
  public void testBoundsWithinThreadCount()
      throws BrokenBarrierException, InterruptedException {
    SchedulerGroupAccountant accountant = mock(SchedulerGroupAccountant.class);
    // Test below relies on jobs > limit
    final int limit = 3;
    final int jobs = 5;
    // we want total threads > limit
    Executor es = Executors.newFixedThreadPool(2 * limit);
    final BoundedAccountingExecutor bes = new BoundedAccountingExecutor(es, limit, accountant);
    final Syncer syncer = new Syncer();
    // barrier parties: all the executables plus 1 for main testing thread
    // startup barrier pauses main thread till all the threads have started
    // validation barrier allows for validation to complete before proceeding further
    syncer._startupBarrier = new CyclicBarrier(limit + 1);
    syncer._validationBarrier = new CyclicBarrier(limit + 1);

    // start adding jobs in new thread. We need to add jobs in new thread
    // because the thread adding jobs is expected to block at limit
    new Thread(new Runnable() {
      @Override
      public void run() {
        for (int i = 0; i < jobs; i++) {
          bes.execute(new Runnable() {
            @Override
            public void run() {
              try {
                _running.incrementAndGet();
                syncer._startupBarrier.await();
                syncer._validationBarrier.await();
                _running.decrementAndGet();
              } catch (Exception e) {
                throw new RuntimeException(e);
              }
            }
          });
        }
      }
    }).start();

    syncer._startupBarrier.await();
    // At this point, 'limit' jobs should have executed the startupBarrier.await() call.
    // The other jobs should be waiting on a semaphore for permits inside the BoundedAccountingExecutor.execute()
    // so they have not executed running.incrementAndGet() yet.
    assertEquals(_running.get(), limit);
    verify(accountant, times(limit)).incrementThreads();
    // reset will clear the counts on incrementThreads
    reset(accountant);
    final int pendingJobs = jobs - limit;
    // Before the pendingJobs get to startupBarrier, reset it to a different value
    // since we cannot change the limit of the CyclicBarrier once created.
    // The new limit will be pending jobs plus the await we will call in this thread.
    syncer._startupBarrier = new CyclicBarrier(pendingJobs + 1);

    // Now let the running threads complete and call running.decrementAndGet. As
    // they exit, the pending jobs will acquire permits and start to increment
    // the running counter and wait on startupBarrier.await().
    syncer._validationBarrier.await();
    // verify additional jobs are run as soon as current job finishes
    syncer._validationBarrier = new CyclicBarrier(pendingJobs + 1);
    // When we run the test in a small number of cores, it is possible that the running jobs
    // have not yet gotten to execute running.decrementAndGet(), but the pending jobs have already
    // done the increment. So, we need to wait until we check the running counter to equal the
    // pending jobs.
    TestUtils.waitForCondition(aVoid -> _running.get() == pendingJobs, 10_000,
        "Invalid number of running jobs" + _running.get());

    // Now that there are no jobs running, we can let the new ones in.
    // All the pending jobs will wait on the validationBarrier after we let them pass
    // the startupbarrier below.
    syncer._startupBarrier.await();
    verify(accountant, times(pendingJobs)).incrementThreads();
    syncer._validationBarrier.await();
  }