@Override
    public Reader createReader(ResultSubpartitionView subpartitionView) throws IOException {
        checkState(!fileChannel.isOpen());

        final FileChannel fc = FileChannel.open(filePath, StandardOpenOption.READ);
        return new FileBufferReader(fc, memorySegmentSize, subpartitionView);
    }
@TestTemplate
    void testRecycleBufferForNotifyingSubpartitionView() throws Exception {
        final int numberOfBuffers = 2;
        try (final BoundedData data = createBoundedData()) {
            writeBuffers(data, numberOfBuffers);

            final VerifyNotificationResultSubpartitionView subpartitionView =
                    new VerifyNotificationResultSubpartitionView();
            final BoundedData.Reader reader = data.createReader(subpartitionView);
            final Buffer buffer1 = reader.nextBuffer();
            final Buffer buffer2 = reader.nextBuffer();
            assertThat(buffer1).isNotNull();
            assertThat(buffer2).isNotNull();

            assertThat(subpartitionView.isAvailable).isFalse();
            buffer1.recycleBuffer();
            // the view is notified while recycling buffer if reader has not tagged finished
            assertThat(subpartitionView.isAvailable).isTrue();

            subpartitionView.resetAvailable();
            assertThat(subpartitionView.isAvailable).isFalse();

            // the next buffer is null to make reader tag finished
            assertThat(reader.nextBuffer()).isNull();

            buffer2.recycleBuffer();
            // the view is not notified while recycling buffer if reader already finished
            assertThat(subpartitionView.isAvailable).isFalse();
        }
    }