@Override
    public void close(RemoteInputChannel inputChannel) throws IOException {

        clientHandler.removeInputChannel(inputChannel);

        if (closeReferenceCounter.updateAndGet(count -> Math.max(count - 1, 0)) == 0
                && !canBeReused()) {
            closeConnection();
        } else {
            clientHandler.cancelRequestFor(inputChannel.getInputChannelId());
        }
    }
@TestTemplate
    void testRetriggerPartitionRequest() throws Exception {
        final long deadline = System.currentTimeMillis() + 30_000L; // 30 secs

        final CreditBasedPartitionRequestClientHandler handler =
                new CreditBasedPartitionRequestClientHandler();
        final EmbeddedChannel channel = new EmbeddedChannel(handler);
        final PartitionRequestClient client =
                createPartitionRequestClient(channel, handler, connectionReuseEnabled);

        final int numExclusiveBuffers = 2;
        final NetworkBufferPool networkBufferPool = new NetworkBufferPool(10, 32);
        final SingleInputGate inputGate = createSingleInputGate(1, networkBufferPool);
        final RemoteInputChannel inputChannel =
                InputChannelBuilder.newBuilder()
                        .setConnectionManager(
                                mockConnectionManagerWithPartitionRequestClient(client))
                        .setPartitionRequestListenerTimeout(1)
                        .setMaxBackoff(2)
                        .buildRemoteChannel(inputGate);

        try {
            inputGate.setInputChannels(inputChannel);
            final BufferPool bufferPool = networkBufferPool.createBufferPool(6, 6);
            inputGate.setBufferPool(bufferPool);
            inputGate.setupChannels();

            // first subpartition request
            inputChannel.requestSubpartitions();

            assertThat(channel.isWritable()).isTrue();
            Object readFromOutbound = channel.readOutbound();
            assertThat(readFromOutbound).isInstanceOf(PartitionRequest.class);
            assertThat(((PartitionRequest) readFromOutbound).receiverId)
                    .isEqualTo(inputChannel.getInputChannelId());
            assertThat(((PartitionRequest) readFromOutbound).credit).isEqualTo(numExclusiveBuffers);

            // retrigger subpartition request, e.g. due to failures
            inputGate.retriggerPartitionRequest(
                    inputChannel.getPartitionId().getPartitionId(), inputChannel.getChannelInfo());
            runAllScheduledPendingTasks(channel, deadline);

            readFromOutbound = channel.readOutbound();
            assertThat(readFromOutbound).isInstanceOf(PartitionRequest.class);
            assertThat(((PartitionRequest) readFromOutbound).receiverId)
                    .isEqualTo(inputChannel.getInputChannelId());
            assertThat(((PartitionRequest) readFromOutbound).credit).isEqualTo(numExclusiveBuffers);

            // retrigger subpartition request once again, e.g. due to failures
            inputGate.retriggerPartitionRequest(
                    inputChannel.getPartitionId().getPartitionId(), inputChannel.getChannelInfo());
            runAllScheduledPendingTasks(channel, deadline);

            readFromOutbound = channel.readOutbound();
            assertThat(readFromOutbound).isInstanceOf(PartitionRequest.class);
            assertThat(((PartitionRequest) readFromOutbound).receiverId)
                    .isEqualTo(inputChannel.getInputChannelId());
            assertThat(((PartitionRequest) readFromOutbound).credit).isEqualTo(numExclusiveBuffers);

            assertThat((Object) channel.readOutbound()).isNull();
        } finally {
            // Release all the buffer resources
            inputGate.close();

            networkBufferPool.destroyAllBufferPools();
            networkBufferPool.destroy();
        }
    }