public HsDataView registerNewConsumer(
            int subpartitionId,
            HsConsumerId consumerId,
            HsSubpartitionConsumerInternalOperations operation)
            throws IOException {
        synchronized (lock) {
            checkState(!isReleased, "HsFileDataManager is already released.");
            lazyInitialize();

            HsSubpartitionFileReader subpartitionReader =
                    fileReaderFactory.createFileReader(
                            subpartitionId,
                            consumerId,
                            dataFileChannel,
                            operation,
                            dataIndex,
                            hybridShuffleConfiguration.getMaxBuffersReadAhead(),
                            this::releaseSubpartitionReader,
                            headerBuf);

            allReaders.add(subpartitionReader);

            mayTriggerReading();
            return subpartitionReader;
        }
    }
@Test
    void testScheduleReadersOrdered() throws Exception {
        TestingHsSubpartitionFileReader reader1 = new TestingHsSubpartitionFileReader();
        TestingHsSubpartitionFileReader reader2 = new TestingHsSubpartitionFileReader();

        CompletableFuture<Void> readBuffersFinished1 = new CompletableFuture<>();
        CompletableFuture<Void> readBuffersFinished2 = new CompletableFuture<>();
        reader1.setReadBuffersConsumer(
                (requestedBuffers, readBuffers) -> {
                    assertThat(readBuffersFinished2).isNotDone();
                    readBuffersFinished1.complete(null);
                });
        reader2.setReadBuffersConsumer(
                (requestedBuffers, readBuffers) -> {
                    assertThat(readBuffersFinished1).isDone();
                    readBuffersFinished2.complete(null);
                });

        reader1.setPriority(1);
        reader2.setPriority(2);

        factory.allReaders.add(reader1);
        factory.allReaders.add(reader2);

        fileDataManager.registerNewConsumer(0, DEFAULT, subpartitionViewOperation);
        fileDataManager.registerNewConsumer(1, DEFAULT, subpartitionViewOperation);

        // trigger run.
        ioExecutor.trigger();

        assertThat(readBuffersFinished2).isCompleted();
    }