@Override
    protected void handleCloseConsumer(CommandCloseConsumer closeConsumer) {
        final long consumerId = closeConsumer.getConsumerId();
        log.info("[{}] Broker notification of closed consumer: {}, assignedBrokerUrl: {}, assignedBrokerUrlTls: {}",
                remoteAddress, consumerId,
                closeConsumer.hasAssignedBrokerServiceUrl() ? closeConsumer.getAssignedBrokerServiceUrl() : null,
                closeConsumer.hasAssignedBrokerServiceUrlTls() ? closeConsumer.getAssignedBrokerServiceUrlTls() : null);
        ConsumerImpl<?> consumer = consumers.remove(consumerId);
        if (consumer != null) {
            String brokerServiceUrl = getBrokerServiceUrl(closeConsumer, consumer);
            Optional<URI> hostUri = parseUri(brokerServiceUrl,
                                             closeConsumer.hasRequestId() ? closeConsumer.getRequestId() : null);
            Optional<Long> initialConnectionDelayMs = hostUri.map(__ -> 0L);
            consumer.connectionClosed(this, initialConnectionDelayMs, hostUri);
        } else {
            log.warn("[{}] Consumer with id {} not found while closing consumer", remoteAddress, consumerId);
        }
    }
@Test
    public void testHandleCloseConsumer() {
        ThreadFactory threadFactory = new DefaultThreadFactory("testHandleCloseConsumer");
        EventLoopGroup eventLoop = EventLoopUtil.newEventLoopGroup(1, false, threadFactory);
        ClientConfigurationData conf = new ClientConfigurationData();
        ClientCnx cnx = new ClientCnx(InstrumentProvider.NOOP, conf, eventLoop);

        long consumerId = 1;
        PulsarClientImpl pulsarClient = mock(PulsarClientImpl.class);
        when(pulsarClient.getConfiguration()).thenReturn(conf);
        ConsumerImpl consumer = mock(ConsumerImpl.class);
        when(consumer.getClient()).thenReturn(pulsarClient);
        cnx.registerConsumer(consumerId, consumer);
        assertEquals(cnx.consumers.size(), 1);

        CommandCloseConsumer closeConsumer = new CommandCloseConsumer().setConsumerId(consumerId).setRequestId(1);
        cnx.handleCloseConsumer(closeConsumer);
        assertEquals(cnx.consumers.size(), 0);

        verify(consumer).connectionClosed(cnx, Optional.empty(), Optional.empty());

        eventLoop.shutdownGracefully();
    }