public synchronized void addWaiter(long delayMs, KafkaFutureImpl<Long> waiter) {
        long timeMs = time.milliseconds();
        if (delayMs <= 0) {
            waiter.complete(timeMs);
        } else {
            long triggerTimeMs = timeMs + delayMs;
            List<KafkaFutureImpl<Long>> futures = waiters.computeIfAbsent(triggerTimeMs, k -> new ArrayList<>());
            futures.add(waiter);
        }
    }
Futures which are waiting for a specified wall-clock time to arrive.
clients/src/test/java/org/apache/kafka/common/utils/MockScheduler.java