MutexableOneThreadTaskExecutor extends OneThreadTaskExecutor { @SuppressWarnings("unchecked") public void clearAndExecuteCommand(RequestResponseCommand<?> command) { logger.info("[clearAndExecuteCommand] {}", command); clear(); super.executeCommand(command); } MutexableOneThreadTaskExecutor(Executor executors, ScheduledExecutorService scheduled); @Override void executeCommand(Command<?> command); @SuppressWarnings("unchecked") void clearAndExecuteCommand(RequestResponseCommand<?> command);  }
@Test public void durableCall2() throws TimeoutException, InterruptedException { AtomicReference<String> state = new AtomicReference<>(); AtomicInteger taskDone = new AtomicInteger(); AtomicInteger counter = new AtomicInteger(); int taskNum = 100; CyclicBarrier barrier = new CyclicBarrier(taskNum); CountDownLatch latch = new CountDownLatch(taskNum); List<CommandFuture> futures = Lists.newCopyOnWriteArrayList(); for (int i = 0; i < taskNum; i++) { int finalI = i; executors.execute(new Runnable() { @Override public void run() { try { barrier.await(); } catch (InterruptedException e) { e.printStackTrace(); } catch (BrokenBarrierException e) { e.printStackTrace(); } StateCommand command = new StateCommand("expected-" + finalI, state, finalI); futures.add(command.future()); command.future().addListener(commandFuture -> { taskDone.incrementAndGet(); }); oneThreadTaskExecutor.clearAndExecuteCommand(command); latch.countDown(); } }); } latch.await(); sleep(300); oneThreadTaskExecutor.clearAndExecuteCommand(new StateCommand("expected", state, 12345)); sleep(200); waitConditionUntilTimeOut(()->oneThreadTaskExecutor.tasks.isEmpty(), 1000); try { waitConditionUntilTimeOut(()->taskDone.get() >= taskNum, 10000); } catch (Exception ignore) { for (CommandFuture future : futures) { if (!future.isDone()) { logger.error("[NOT DONE] {}", future.command().getName()); } } logger.error("[taskDone] {}", taskDone, ignore); throw ignore; } Assert.assertTrue(oneThreadTaskExecutor.tasks.isEmpty()); Assert.assertEquals("expected", state.get()); logger.info("[task done]{}", taskDone); }