@Test public void testBoundsWithinThreadCount() throws BrokenBarrierException, InterruptedException { SchedulerGroupAccountant accountant = mock(SchedulerGroupAccountant.class); final int limit = 3; final int jobs = 5; Executor es = Executors.newFixedThreadPool(2 * limit); final BoundedAccountingExecutor bes = new BoundedAccountingExecutor(es, limit, accountant); final Syncer syncer = new Syncer(); syncer.startupBarrier = new CyclicBarrier(limit + 1); syncer.validationBarrier = new CyclicBarrier(limit + 1); 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(); assertEquals(running.get(), limit); verify(accountant, times(limit)).incrementThreads(); reset(accountant); final int pendingJobs = jobs - limit; syncer.startupBarrier = new CyclicBarrier(pendingJobs + 1); syncer.validationBarrier.await(); syncer.validationBarrier = new CyclicBarrier(pendingJobs + 1); TestUtils.waitForCondition(aVoid -> running.get() == pendingJobs, 10_000, "Invalid number of running jobs" + running.get()); syncer.startupBarrier.await(); verify(accountant, times(pendingJobs)).incrementThreads(); syncer.validationBarrier.await(); }
@Override public void execute(Runnable command) { delegateExecutor.execute(toAccountingRunnable(command)); }
BoundedAccountingExecutor extends QueryExecutorService { @Override public void execute(Runnable command) { delegateExecutor.execute(toAccountingRunnable(command)); } }
BoundedAccountingExecutor extends QueryExecutorService { @Override public void execute(Runnable command) { delegateExecutor.execute(toAccountingRunnable(command)); } BoundedAccountingExecutor(@Nonnull Executor s, int bounds, @Nonnull SchedulerGroupAccountant accountant); }
BoundedAccountingExecutor extends QueryExecutorService { @Override public void execute(Runnable command) { delegateExecutor.execute(toAccountingRunnable(command)); } BoundedAccountingExecutor(@Nonnull Executor s, int bounds, @Nonnull SchedulerGroupAccountant accountant); @Override void execute(Runnable command); @Override void releaseWorkers(); }
BoundedAccountingExecutor extends QueryExecutorService { @Override public void execute(Runnable command) { delegateExecutor.execute(toAccountingRunnable(command)); } BoundedAccountingExecutor(@Nonnull Executor s, int bounds, @Nonnull SchedulerGroupAccountant accountant); @Override void execute(Runnable command); @Override void releaseWorkers();  }