BoundedExecutor implements Executor { @Override public void execute(Runnable task) { checkState(!failed.get(), "BoundedExecutor is in a failed state"); queue.add(task); int size = queueSize.incrementAndGet(); if (size <= maxThreads) { try { coreExecutor.execute(this::drainQueue); } catch (Throwable e) { failed.set(true); log.error("BoundedExecutor state corrupted due to underlying executor failure"); throw e; } } } BoundedExecutor(Executor coreExecutor, int maxThreads); @Override void execute(Runnable task);  }
@Test public void testCounter() { int maxThreads = 1; BoundedExecutor boundedExecutor = new BoundedExecutor(executorService, maxThreads); int stageTasks = 100_000; int totalTasks = stageTasks * 2; AtomicInteger counter = new AtomicInteger(); CountDownLatch initializeLatch = new CountDownLatch(maxThreads); CountDownLatch startLatch = new CountDownLatch(1); CountDownLatch completeLatch = new CountDownLatch(totalTasks); for (int i = 0; i < stageTasks; i++) { boundedExecutor.execute(() -> { try { initializeLatch.countDown(); Uninterruptibles.awaitUninterruptibly(startLatch); int initialCount = counter.get(); counter.set(initialCount + 1); } finally { completeLatch.countDown(); } }); } assertTrue(Uninterruptibles.awaitUninterruptibly(initializeLatch, 1, TimeUnit.MINUTES)); startLatch.countDown(); for (int i = 0; i < stageTasks; i++) { boundedExecutor.execute(() -> { try { int initialCount = counter.get(); counter.set(initialCount + 1); } finally { completeLatch.countDown(); } }); } assertTrue(Uninterruptibles.awaitUninterruptibly(completeLatch, 1, TimeUnit.MINUTES)); assertEquals(counter.get(), totalTasks); }
@Test public void testExecutorCorruptionDetection() { AtomicBoolean reject = new AtomicBoolean(); Executor executor = command -> { if (reject.get()) { throw new RejectedExecutionException(); } executorService.execute(command); }; BoundedExecutor boundedExecutor = new BoundedExecutor(executor, 1); reject.set(true); try { boundedExecutor.execute(() -> fail("Should not be run")); fail("Execute should fail"); } catch (Exception e) { } reject.set(false); try { boundedExecutor.execute(() -> fail("Should not be run")); fail("Execute should still fail"); } catch (Exception e) { } }