@Override
    public PipelinedSubpartitionView createReadView(
            BufferAvailabilityListener availabilityListener) {
        synchronized (buffers) {
            checkState(!isReleased);
            checkState(
                    readView == null,
                    "Subpartition %s of is being (or already has been) consumed, "
                            + "but pipelined subpartitions can only be consumed once.",
                    getSubPartitionIndex(),
                    parent.getPartitionId());

            LOG.debug(
                    "{}: Creating read view for subpartition {} of partition {}.",
                    parent.getOwningTaskName(),
                    getSubPartitionIndex(),
                    parent.getPartitionId());

            readView = new PipelinedSubpartitionView(this, availabilityListener);
        }

        return readView;
    }
@TestTemplate
    void testIllegalReadViewRequest() throws Exception {
        final PipelinedSubpartition subpartition = createSubpartition();

        // Successful request
        assertThat(subpartition.createReadView(new NoOpBufferAvailablityListener())).isNotNull();

        assertThatThrownBy(() -> subpartition.createReadView(new NoOpBufferAvailablityListener()))
                .withFailMessage(
                        "Did not throw expected exception after duplicate notifyNonEmpty view request.")
                .isInstanceOf(IllegalStateException.class);
    }