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

        long producerId = 1;
        PulsarClientImpl pulsarClient = mock(PulsarClientImpl.class);
        when(pulsarClient.getConfiguration()).thenReturn(conf);
        ProducerImpl producer = mock(ProducerImpl.class);
        when(producer.getClient()).thenReturn(pulsarClient);
        cnx.registerProducer(producerId, producer);
        assertEquals(cnx.producers.size(), 1);

        CommandCloseProducer closeProducerCmd = new CommandCloseProducer().setProducerId(producerId).setRequestId(1);
        cnx.handleCloseProducer(closeProducerCmd);
        assertEquals(cnx.producers.size(), 0);

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

        eventLoop.shutdownGracefully();
    }