@Override
    public ScheduledFuture<?> schedule(String name, Runnable task, long delayMs, long periodMs) {
        log.debug("Scheduling task {} with initial delay {} ms and period {} ms.", name, delayMs, periodMs);
        synchronized (this) {
            if (isStarted()) {
                Runnable runnable = () -> {
                    try {
                        log.trace("Beginning execution of scheduled task '{}'.", name);
                        task.run();
                    } catch (Throwable t) {
                        log.error("Uncaught exception in scheduled task '{}'", name, t);
                    } finally {
                        log.trace("Completed execution of scheduled task '{}'.", name);
                    }
                };
                if (periodMs > 0)
                    return executor.scheduleAtFixedRate(runnable, delayMs, periodMs, TimeUnit.MILLISECONDS);
                else
                    return executor.schedule(runnable, delayMs, TimeUnit.MILLISECONDS);
            } else {
                log.info("Kafka scheduler is not running at the time task '{}' is scheduled. The task is ignored.", name);
                return new NoOpScheduledFutureTask();
            }
        }
    }
A scheduler based on java.util.concurrent.ScheduledThreadPoolExecutor
<br>
It has a pool of kafka-scheduler- threads that do the actual work.
server-common/src/main/java/org/apache/kafka/server/util/KafkaScheduler.java