@Override
  public Future<?> schedule(Executor executor, Runnable command, long delay, TimeUnit unit) {
    requireNonNull(executor);
    requireNonNull(command);
    requireNonNull(unit);

    if (scheduledExecutorService.isShutdown()) {
      return DisabledFuture.INSTANCE;
    }
    return scheduledExecutorService.schedule(() -> {
      try {
        executor.execute(command);
      } catch (Throwable t) {
        logger.log(Level.WARNING, "Exception thrown when submitting scheduled task", t);
        throw t;
      }
    }, delay, unit);
  }
@Test(dataProvider = "runnableSchedulers")
  public void scheduler_exception(Scheduler scheduler) {
    var thread = new AtomicReference<Thread>();
    Executor executor = task -> {
      thread.set(Thread.currentThread());
      throw new IllegalStateException();
    };
    var future = scheduler.schedule(executor, () -> {}, 1L, TimeUnit.NANOSECONDS);
    assertThat(future).isNotNull();
    await().untilAtomic(thread, is(not(nullValue())));

    if (thread.get() == Thread.currentThread()) {
      assertThat(logEvents()
          .withMessage("Exception thrown when submitting scheduled task")
          .withThrowable(IllegalStateException.class)
          .withLevel(WARN)
          .exclusively())
          .hasSize(1);
    }
  }