@Override
  public synchronized Optional<ListenableFuture<V>> schedule(
      Checkable<K, V> target, K context) {
    if (checksInProgress.containsKey(target)) {
      return Optional.empty();
    }

    final LastCheckResult<V> result = completedChecks.get(target);
    if (result != null) {
      final long msSinceLastCheck = timer.monotonicNow() - result.completedAt;
      if (msSinceLastCheck < minMsBetweenChecks) {
        LOG.debug("Skipped checking {}. Time since last check {}ms " +
                "is less than the min gap {}ms.",
            target, msSinceLastCheck, minMsBetweenChecks);
        return Optional.empty();
      }
    }

    LOG.info("Scheduling a check for {}", target);
    final ListenableFuture<V> lfWithoutTimeout = executorService.submit(
        new Callable<V>() {
          @Override
          public V call() throws Exception {
            return target.check(context);
          }
        });
    final ListenableFuture<V> lf;

    if (diskCheckTimeout > 0) {
      lf = TimeoutFuture
          .create(lfWithoutTimeout, diskCheckTimeout, TimeUnit.MILLISECONDS,
              scheduledExecutorService);
    } else {
      lf = lfWithoutTimeout;
    }

    checksInProgress.put(target, lf);
    addResultCachingCallback(target, lf);
    return Optional.of(lf);
  }
@Test(timeout=60000)
  public void testExceptionIsPropagated() throws Exception {
    final ThrowingCheckable target = new ThrowingCheckable();
    final FakeTimer timer = new FakeTimer();
    ThrottledAsyncChecker<Boolean, Boolean> checker =
        new ThrottledAsyncChecker<>(timer, MIN_ERROR_CHECK_GAP, 0,
            getExecutorService());

    final Optional<ListenableFuture<Boolean>> olf =
        checker.schedule(target, true);
    assertTrue(olf.isPresent());
    try {
      olf.get().get();
      fail("Failed to get expected ExecutionException");
    } catch(ExecutionException ee) {
      assertTrue(ee.getCause() instanceof DummyException);
    }
  }