ExecutorImpl implements Executor { boolean scheduleDelayedTask(String action, Task task, Duration delay) { Instant now = Instant.now(); String actionId = LogManager.ID_GENERATOR.next(now); logger.debug("submit delayed task, action={}, id={}, delay={}", action, actionId, delay); Callable<Void> execution = execution(actionId, action, now.plus(delay), () -> { task.execute(); return null; }); Runnable delayedTask = () -> { try { submitTask(action, execution); } catch (Throwable e) { logger.error(e.getMessage(), e); } }; try { scheduler.schedule(delayedTask, delay.toMillis(), TimeUnit.MILLISECONDS); return true; } catch (RejectedExecutionException e) { logger.warn(errorCode("TASK_REJECTED"), "reject task due to server is shutting down, action={}", action, e); } return false; } ExecutorImpl(int poolSize, String name, LogManager logManager); void shutdown(); void awaitTermination(long timeoutInMs); @Override Future<T> submit(String action, Callable<T> task); @Override void submit(String action, Task task, Duration delay);  }
@Test void scheduleDelayedTask() throws InterruptedException { executor.scheduler = ThreadPools.singleThreadScheduler("test-"); boolean scheduled = executor.scheduleDelayedTask("action", () -> { ActionLog actionLog = LogManager.CURRENT_ACTION_LOG.get(); assertThat(actionLog.action).isEqualTo("task:action"); assertThat(actionLog.context).doesNotContainKey("root_action"); }, Duration.ZERO); executor.scheduler.shutdown(); executor.scheduler.awaitTermination(10, TimeUnit.SECONDS); assertThat(scheduled).isTrue(); }