SortMergeSubpartitionReader createSubpartitionReader(
            BufferAvailabilityListener availabilityListener,
            int targetSubpartition,
            PartitionedFile resultFile)
            throws IOException {
        synchronized (lock) {
            checkState(!isReleased, "Partition is already released.");

            PartitionedFileReader fileReader = createFileReader(resultFile, targetSubpartition);
            SortMergeSubpartitionReader subpartitionReader =
                    new SortMergeSubpartitionReader(availabilityListener, fileReader);
            if (allReaders.isEmpty()) {
                bufferPool.registerRequester(this);
            }
            allReaders.add(subpartitionReader);
            sortedReaders.add(subpartitionReader);
            subpartitionReader
                    .getReleaseFuture()
                    .thenRun(() -> releaseSubpartitionReader(subpartitionReader));

            mayTriggerReading();
            return subpartitionReader;
        }
    }
@Test
    void testOnReadBufferRequestError() throws Exception {
        ManuallyTriggeredScheduledExecutorService schedulerExecutor =
                new ManuallyTriggeredScheduledExecutorService();
        readScheduler =
                new SortMergeResultPartitionReadScheduler(
                        bufferPool, schedulerExecutor, new Object());
        SortMergeSubpartitionReader subpartitionReader =
                readScheduler.createSubpartitionReader(
                        new NoOpBufferAvailablityListener(), 0, partitionedFile);
        bufferPool.destroy();
        assertThat(schedulerExecutor.numQueuedRunnables()).isEqualTo(1);
        // we should trigger the scheduled task to handle the buffer request error.
        schedulerExecutor.trigger();

        waitUntilReadFinish();

        assertThat(subpartitionReader.isReleased()).isTrue();
        assertThat(subpartitionReader.getFailureCause()).isNotNull();
        assertThat(subpartitionReader.getAvailabilityAndBacklog(false).isAvailable()).isTrue();
        assertAllResourcesReleased();
    }