@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) { } }
@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 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 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); }
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); }
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);  }