@Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        try {
            decodeMsg(msg);
        } catch (Throwable t) {
            notifyAllChannelsOfErrorAndClose(t);
        }
    }
@Test
    void testNotifyCreditAvailableAfterReleased() throws Exception {
        final CreditBasedPartitionRequestClientHandler handler =
                new CreditBasedPartitionRequestClientHandler();
        final EmbeddedChannel channel = new EmbeddedChannel(handler);
        final PartitionRequestClient client =
                new NettyPartitionRequestClient(
                        channel,
                        handler,
                        mock(ConnectionID.class),
                        mock(PartitionRequestClientFactory.class));

        final NetworkBufferPool networkBufferPool = new NetworkBufferPool(10, 32);
        final SingleInputGate inputGate = createSingleInputGate(1, networkBufferPool);
        final RemoteInputChannel inputChannel = createRemoteInputChannel(inputGate, client);
        try {
            inputGate.setInputChannels(inputChannel);
            final BufferPool bufferPool = networkBufferPool.createBufferPool(6, 6);
            inputGate.setBufferPool(bufferPool);
            inputGate.setupChannels();

            inputChannel.requestSubpartitions();

            // This should send the partition request
            Object readFromOutbound = channel.readOutbound();
            assertThat(readFromOutbound).isInstanceOf(PartitionRequest.class);
            assertThat(((PartitionRequest) readFromOutbound).credit).isEqualTo(2);

            // Trigger request floating buffers via buffer response to notify credits available
            final BufferResponse bufferResponse =
                    createBufferResponse(
                            TestBufferFactory.createBuffer(32),
                            0,
                            inputChannel.getInputChannelId(),
                            1,
                            new NetworkBufferAllocator(handler));
            handler.channelRead(mock(ChannelHandlerContext.class), bufferResponse);

            assertThat(inputChannel.getUnannouncedCredit()).isEqualTo(2);

            // Release the input channel
            inputGate.close();

            // it should send a close request after releasing the input channel,
            // but will not notify credits for a released input channel.
            readFromOutbound = channel.readOutbound();
            assertThat(readFromOutbound).isInstanceOf(CloseRequest.class);

            channel.runPendingTasks();
            assertThat((Object) channel.readOutbound()).isNull();
        } finally {
            releaseResource(inputGate, networkBufferPool);
            channel.close();
        }
    }