@Override
    public boolean isActive() {
        return isActive;
    }
@Test(timeOut = 30000)
    public void testClientClosedProducerThenSendsMessageAndGetsClosed() 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);
        assertTrue(getResponse() instanceof CommandProducerSuccess);

        ByteBuf closeProducer = Commands.newCloseProducer(1,2);
        channel.writeInbound(closeProducer);
        assertTrue(getResponse() instanceof CommandSuccess);

        // Send message and get disconnected
        sendMessage();
        Awaitility.await().atMost(10, TimeUnit.SECONDS).until(() -> !channel.isActive());
        channel.finish();
    }