public ConcurrentLongHashMap<CompletableFuture<Producer>> getProducers() {
        return producers;
    }
@Test(timeOut = 30000)
    public void testBrokerClosedProducerClientRecreatesProducerThenSendCommand() throws Exception {
        resetChannel();
        setChannelConnected();
        setConnectionVersion(ProtocolVersion.v5.getValue());
        serverCnx.cancelKeepAliveTask();

        String producerName = "my-producer";

        ByteBuf clientCommand1 = Commands.newProducer(successTopicName, 1 /* producer id */, 1 /* request id */,
                producerName, Collections.emptyMap(), false);
        channel.writeInbound(clientCommand1);
        assertThat(getResponse()).isInstanceOf(CommandProducerSuccess.class);

        // Call disconnect method on producer to trigger activity similar to unloading
        Producer producer = serverCnx.getProducers().get(1).get();
        assertNotNull(producer);
        producer.disconnect();
        channel.runPendingTasks();
        assertThat(getResponse()).isInstanceOf(CommandCloseProducer.class);

        // Send message and expect no response
        sendMessage();

        assertTrue(channel.outboundMessages().isEmpty());

        // Move clock forward to trigger scheduled clean up task
        ByteBuf createProducer2 = Commands.newProducer(successTopicName, 1 /* producer id */, 1 /* request id */,
                producerName, Collections.emptyMap(), false);
        channel.writeInbound(createProducer2);
        assertTrue(getResponse() instanceof CommandProducerSuccess);

        // Send message and expect success
        sendMessage();

        assertTrue(getResponse() instanceof CommandSendReceipt);
        channel.finish();
    }