@Override
    public synchronized CompletableFuture<Void> addConsumer(Consumer consumer) {
        if (IS_CLOSED_UPDATER.get(this) == TRUE) {
            log.warn("[{}] Dispatcher is already closed. Closing consumer {}", name, consumer);
            consumer.disconnect();
            return CompletableFuture.completedFuture(null);
        }
        return super.addConsumer(consumer).thenCompose(__ ->
                selector.addConsumer(consumer).handle((value, ex) -> {
                    if (ex != null) {
                        synchronized (NonPersistentStickyKeyDispatcherMultipleConsumers.this) {
                            consumerSet.removeAll(consumer);
                            consumerList.remove(consumer);
                        }
                        throw FutureUtil.wrapToCompletionException(ex);
                    } else {
                        return value;
                    }
                }));
    }
@Test(timeOut = 10000)
    public void testAddConsumerWhenClosed() throws Exception {
        nonpersistentDispatcher.close().get();
        Consumer consumer = mock(Consumer.class);
        nonpersistentDispatcher.addConsumer(consumer);
        verify(consumer, times(1)).disconnect();
        assertEquals(0, nonpersistentDispatcher.getConsumers().size());
        assertTrue(selector.getConsumerKeyHashRanges().isEmpty());
    }