@Nullable
    public Buffer requestBuffer() {
        return bufferManager.requestBuffer();
    }
@Test
    void testRequestBuffer() throws Exception {
        BufferPool bufferPool = new TestBufferPool();
        SingleInputGate inputGate = createSingleInputGate(bufferPool);

        RemoteInputChannel remoteChannel1 = createRemoteInputChannel(inputGate, 0, 2);
        RemoteInputChannel remoteChannel2 = createRemoteInputChannel(inputGate, 1, 0);
        inputGate.setup();
        remoteChannel1.requestSubpartitions();
        remoteChannel2.requestSubpartitions();

        remoteChannel1.onSenderBacklog(2);
        remoteChannel2.onSenderBacklog(2);

        for (int i = 4; i >= 0; --i) {
            assertThat(remoteChannel1.getNumberOfRequiredBuffers()).isEqualTo(i);
            remoteChannel1.requestBuffer();
        }

        for (int i = 2; i >= 0; --i) {
            assertThat(remoteChannel2.getNumberOfRequiredBuffers()).isEqualTo(i);
            remoteChannel2.requestBuffer();
        }
    }