void notifyNewBufferSize(InputChannelID receiverId, int newBufferSize) {
        if (fatalError) {
            return;
        }

        // It is possible to receive new buffer size before the reader would be created since the
        // downstream task could calculate buffer size even using the data from one channel but it
        // sends new buffer size into all upstream even if they don't ready yet. In this case, just
        // ignore the new buffer size.
        NetworkSequenceViewReader reader = allReaders.get(receiverId);
        if (reader != null) {
            reader.notifyNewBufferSize(newBufferSize);
        }
    }
@Test
    void testNotifyNewBufferSize() throws Exception {
        // given: Result partition and the reader for subpartition 0.
        ResultPartition parent = createResultPartition();

        BufferAvailabilityListener bufferAvailabilityListener = new NoOpBufferAvailablityListener();
        ResultSubpartitionView view =
                parent.createSubpartitionView(
                        new ResultSubpartitionIndexSet(0), bufferAvailabilityListener);
        ResultPartition partition =
                TestingResultPartition.newBuilder()
                        .setCreateSubpartitionViewFunction((index, listener) -> view)
                        .build();

        InputChannelID receiverId = new InputChannelID();
        PartitionRequestQueue queue = new PartitionRequestQueue();
        CreditBasedSequenceNumberingViewReader reader =
                new CreditBasedSequenceNumberingViewReader(receiverId, 2, queue);
        EmbeddedChannel channel = new EmbeddedChannel(queue);

        reader.notifySubpartitionsCreated(partition, new ResultSubpartitionIndexSet(0));
        queue.notifyReaderCreated(reader);

        // when: New buffer size received.
        queue.notifyNewBufferSize(receiverId, 65);

        // and: New records emit.
        parent.emitRecord(ByteBuffer.allocate(128), 0);
        parent.emitRecord(ByteBuffer.allocate(10), 0);
        parent.emitRecord(ByteBuffer.allocate(60), 0);

        reader.notifyDataAvailable(view);
        channel.runPendingTasks();

        // then: Buffers of received size will be in outbound channel.
        Object data1 = channel.readOutbound();
        // The size can not be less than the first record in buffer.
        assertThat(((NettyMessage.BufferResponse) data1).buffer.getSize()).isEqualTo(128);
        Object data2 = channel.readOutbound();
        // The size should shrink up to notified buffer size.
        assertThat(((NettyMessage.BufferResponse) data2).buffer.getSize()).isEqualTo(65);
    }