@Override
    public void close() throws InterruptedException {
        beginShutdown("KafkaEventQueue#close");
        eventHandlerThread.join();
        log.info("closed event queue.");
    }
@Test
    public void testRejectedExecutionException() throws Exception {
        KafkaEventQueue queue = new KafkaEventQueue(Time.SYSTEM, logContext,
            "testRejectedExecutionException");
        queue.close();
        CompletableFuture<Void> future = new CompletableFuture<>();
        queue.append(new EventQueue.Event() {
                @Override
                public void run() {
                    future.complete(null);
                }

                @Override
                public void handleException(Throwable e) {
                    future.completeExceptionally(e);
                }
            });
        assertEquals(RejectedExecutionException.class, assertThrows(
            ExecutionException.class, () -> future.get()).getCause().getClass());
    }