@Override
    public void beginShutdown(String source) {
        lock.lock();
        try {
            if (shuttingDown) {
                log.debug("{}: Event queue is already shutting down.", source);
                return;
            }
            log.info("{}: shutting down event queue.", source);
            shuttingDown = true;
            eventHandler.cond.signal();
        } finally {
            lock.unlock();
        }
    }
@Test
    public void testShutdownBeforeDeferred() throws Exception {
        try (KafkaEventQueue queue = new KafkaEventQueue(Time.SYSTEM, logContext, "testShutdownBeforeDeferred")) {
            final AtomicInteger count = new AtomicInteger(0);
            CompletableFuture<Integer> future = new CompletableFuture<>();
            queue.scheduleDeferred("myDeferred",
                    __ -> OptionalLong.of(Time.SYSTEM.nanoseconds() + HOURS.toNanos(1)),
                    new FutureEvent<>(future, () -> count.getAndAdd(1)));
            queue.beginShutdown("testShutdownBeforeDeferred");
            assertEquals(RejectedExecutionException.class, assertThrows(ExecutionException.class, () -> future.get()).getCause().getClass());
            assertEquals(0, count.get());
        }
    }