@Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        try {
            decodeMsg(msg);
        } catch (Throwable t) {
            notifyAllChannelsOfErrorAndClose(t);
        }
    }
@Test
    void testNotifyCreditAvailable() throws Exception {
        final CreditBasedPartitionRequestClientHandler handler =
                new CreditBasedPartitionRequestClientHandler();
        final NetworkBufferAllocator allocator = new NetworkBufferAllocator(handler);
        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(2, networkBufferPool);
        final RemoteInputChannel[] inputChannels = new RemoteInputChannel[2];
        inputChannels[0] = createRemoteInputChannel(inputGate, client);
        inputChannels[1] = createRemoteInputChannel(inputGate, client);
        try {
            inputGate.setInputChannels(inputChannels);
            final BufferPool bufferPool = networkBufferPool.createBufferPool(6, 6);
            inputGate.setBufferPool(bufferPool);
            inputGate.setupChannels();

            inputChannels[0].requestSubpartitions();
            inputChannels[1].requestSubpartitions();

            // The two input channels should send partition requests
            assertThat(channel.isWritable()).isTrue();
            Object readFromOutbound = channel.readOutbound();
            assertThat(readFromOutbound).isInstanceOf(PartitionRequest.class);
            assertThat(inputChannels[0].getInputChannelId())
                    .isEqualTo(((PartitionRequest) readFromOutbound).receiverId);
            assertThat(((PartitionRequest) readFromOutbound).credit).isEqualTo(2);

            readFromOutbound = channel.readOutbound();
            assertThat(readFromOutbound).isInstanceOf(PartitionRequest.class);
            assertThat(inputChannels[1].getInputChannelId())
                    .isEqualTo(((PartitionRequest) readFromOutbound).receiverId);
            assertThat(((PartitionRequest) readFromOutbound).credit).isEqualTo(2);

            // The buffer response will take one available buffer from input channel, and it will
            // trigger
            // requesting (backlog + numExclusiveBuffers - numAvailableBuffers) floating buffers
            final BufferResponse bufferResponse1 =
                    createBufferResponse(
                            TestBufferFactory.createBuffer(32),
                            0,
                            inputChannels[0].getInputChannelId(),
                            1,
                            allocator);
            final BufferResponse bufferResponse2 =
                    createBufferResponse(
                            TestBufferFactory.createBuffer(32),
                            0,
                            inputChannels[1].getInputChannelId(),
                            1,
                            allocator);
            handler.channelRead(mock(ChannelHandlerContext.class), bufferResponse1);
            handler.channelRead(mock(ChannelHandlerContext.class), bufferResponse2);

            assertThat(inputChannels[0].getUnannouncedCredit()).isEqualTo(2);
            assertThat(inputChannels[1].getUnannouncedCredit()).isEqualTo(2);

            channel.runPendingTasks();

            // The two input channels should notify credits availability via the writable channel
            readFromOutbound = channel.readOutbound();
            assertThat(readFromOutbound).isInstanceOf(AddCredit.class);
            assertThat(inputChannels[0].getInputChannelId())
                    .isEqualTo(((AddCredit) readFromOutbound).receiverId);
            assertThat(((AddCredit) readFromOutbound).credit).isEqualTo(2);

            readFromOutbound = channel.readOutbound();
            assertThat(readFromOutbound).isInstanceOf(AddCredit.class);
            assertThat(inputChannels[1].getInputChannelId())
                    .isEqualTo(((AddCredit) readFromOutbound).receiverId);
            assertThat(((AddCredit) readFromOutbound).credit).isEqualTo(2);
            assertThat((Object) channel.readOutbound()).isNull();

            ByteBuf channelBlockingBuffer = blockChannel(channel);

            // Trigger notify credits availability via buffer response on the condition of an
            // un-writable channel
            final BufferResponse bufferResponse3 =
                    createBufferResponse(
                            TestBufferFactory.createBuffer(32),
                            1,
                            inputChannels[0].getInputChannelId(),
                            1,
                            allocator);
            handler.channelRead(mock(ChannelHandlerContext.class), bufferResponse3);

            assertThat(inputChannels[0].getUnannouncedCredit()).isOne();
            assertThat(inputChannels[1].getUnannouncedCredit()).isZero();

            channel.runPendingTasks();

            // The input channel will not notify credits via un-writable channel
            assertThat(channel.isWritable()).isFalse();
            assertThat((Object) channel.readOutbound()).isNull();

            // Flush the buffer to make the channel writable again
            channel.flush();
            assertThat(channelBlockingBuffer).isSameAs(channel.readOutbound());

            // The input channel should notify credits via channel's writability changed event
            assertThat(channel.isWritable()).isTrue();
            readFromOutbound = channel.readOutbound();
            assertThat(readFromOutbound).isInstanceOf(AddCredit.class);
            assertThat(((AddCredit) readFromOutbound).credit).isOne();
            assertThat(inputChannels[0].getUnannouncedCredit()).isZero();
            assertThat(inputChannels[1].getUnannouncedCredit()).isZero();

            // no more messages
            assertThat((Object) channel.readOutbound()).isNull();
        } finally {
            releaseResource(inputGate, networkBufferPool);
            channel.close();
        }
    }