@Override
    public <T> Future<T> schedule(final ScheduledExecutorService executor,
                                  final Callable<T> callable, long delayMs) {
        final KafkaFutureImpl<T> future = new KafkaFutureImpl<>();
        KafkaFutureImpl<Long> waiter = new KafkaFutureImpl<>();
        waiter.thenApply((KafkaFuture.BaseFunction<Long, Void>) now -> {
            executor.submit((Callable<Void>) () -> {
                // Note: it is possible that we'll execute Callable#call right after
                // the future is cancelled.  This is a valid sequence of events
                // that the author of the Callable needs to be able to handle.
                //
                // Note 2: If the future is cancelled, we will not remove the waiter
                // from this MockTime object.  This small bit of inefficiency is acceptable
                // in testing code (at least we aren't polling!)
                if (!future.isCancelled()) {
                    try {
                        log.trace("Invoking {} at {}", callable, now);
                        future.complete(callable.call());
                    } catch (Throwable throwable) {
                        future.completeExceptionally(throwable);
                    }
                }
                return null;
            });
            return null;
        });
        log.trace("Scheduling {} for {} ms from now.", callable, delayMs);
        addWaiter(delayMs, waiter);
        return future;
    }
Futures which are waiting for a specified wall-clock time to arrive.
clients/src/test/java/org/apache/kafka/common/utils/MockScheduler.java