@Override
    void releaseAllResources() throws IOException {
        if (isReleased.compareAndSet(false, true)) {

            final ArrayDeque<Buffer> releasedBuffers;
            synchronized (receivedBuffers) {
                releasedBuffers =
                        receivedBuffers.stream()
                                .map(sb -> sb.buffer)
                                .collect(Collectors.toCollection(ArrayDeque::new));
                receivedBuffers.clear();
            }
            bufferManager.releaseAllBuffers(releasedBuffers);

            // The released flag has to be set before closing the connection to ensure that
            // buffers received concurrently with closing are properly recycled.
            if (partitionRequestClient != null) {
                partitionRequestClient.close(this);
            } else {
                connectionManager.closeOpenChannelConnections(connectionId);
            }
        }
    }
@Test
    void testUnblockReleasedChannel() throws Exception {
        SingleInputGate inputGate = createSingleInputGate(1);
        RemoteInputChannel remoteChannel = createRemoteInputChannel(inputGate);

        remoteChannel.releaseAllResources();
        assertThatThrownBy(remoteChannel::resumeConsumption)
                .isInstanceOf(IllegalStateException.class);
    }