@Override
    protected ResultSubpartitionView createSubpartitionView(
            int subpartitionIndex, BufferAvailabilityListener availabilityListener)
            throws IOException {
        synchronized (lock) {
            checkElementIndex(subpartitionIndex, numSubpartitions, "Subpartition not found.");
            checkState(!isReleased(), "Partition released.");
            checkState(isFinished(), "Trying to read unfinished blocking partition.");

            if (!resultFile.isReadable()) {
                throw new PartitionNotFoundException(getPartitionId());
            }

            return readScheduler.createSubpartitionReader(
                    availabilityListener, subpartitionIndex, resultFile);
        }
    }
@TestTemplate
    void testReadUnfinishedPartition() throws Exception {
        BufferPool bufferPool = globalPool.createBufferPool(10, 10);
        SortMergeResultPartition partition = createSortMergedPartition(10, bufferPool);
        assertThatThrownBy(
                        () ->
                                partition.createSubpartitionView(
                                        new ResultSubpartitionIndexSet(0), listener))
                .isInstanceOf(IllegalStateException.class);
        bufferPool.lazyDestroy();
    }