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
    @Timeout(60)
    void testCreateSubpartitionReader() throws Exception {
        ManuallyTriggeredScheduledExecutorService ioExecutor =
                new ManuallyTriggeredScheduledExecutorService();
        readScheduler =
                new SortMergeResultPartitionReadScheduler(bufferPool, ioExecutor, new Object());

        SortMergeSubpartitionReader subpartitionReader =
                readScheduler.createSubpartitionReader(
                        new NoOpBufferAvailablityListener(), 0, partitionedFile);

        assertThat(readScheduler.isRunning()).isTrue();
        assertThat(readScheduler.getDataFileChannel().isOpen()).isTrue();
        assertThat(readScheduler.getIndexFileChannel().isOpen()).isTrue();
        assertThat(ioExecutor.numQueuedRunnables()).isEqualTo(1);

        int numBuffersRead = 0;
        while (numBuffersRead < numBuffersPerSubpartition) {
            ioExecutor.triggerAll();
            ResultSubpartition.BufferAndBacklog bufferAndBacklog =
                    subpartitionReader.getNextBuffer();
            if (bufferAndBacklog != null) {
                int numBytes = bufferAndBacklog.buffer().readableBytes();
                MemorySegment segment = MemorySegmentFactory.allocateUnpooledSegment(numBytes);
                Buffer fullBuffer =
                        ((CompositeBuffer) bufferAndBacklog.buffer()).getFullBufferData(segment);
                assertThat(ByteBuffer.wrap(dataBytes)).isEqualTo(fullBuffer.getNioBufferReadable());
                fullBuffer.recycleBuffer();
                ++numBuffersRead;
            }
        }
    }