@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 testContextIsPassed() throws Exception {
    final NoOpCheckable target1 = new NoOpCheckable();
    final FakeTimer timer = new FakeTimer();
    ThrottledAsyncChecker<Boolean, Boolean> checker =
        new ThrottledAsyncChecker<>(timer, MIN_ERROR_CHECK_GAP, 0,
            getExecutorService());

    assertTrue(checker.schedule(target1, true).isPresent());
    waitTestCheckableCheckCount(target1, 1L);
    timer.advance(MIN_ERROR_CHECK_GAP + 1);
    assertTrue(checker.schedule(target1, false).isPresent());
    waitTestCheckableCheckCount(target1, 2L);

  }