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 testRunReadBuffersThrowException() throws Exception {
        TestingHsSubpartitionFileReader reader = new TestingHsSubpartitionFileReader();
        CompletableFuture<Throwable> cause = new CompletableFuture<>();
        reader.setFailConsumer((cause::complete));
        reader.setReadBuffersConsumer(
                (requestedBuffers, readBuffers) -> {
                    throw new IOException("expected exception.");
                });
        factory.allReaders.add(reader);

        fileDataManager.registerNewConsumer(0, DEFAULT, subpartitionViewOperation);

        ioExecutor.trigger();

        assertThat(cause).isCompleted();
        assertThat(cause.get())
                .isInstanceOf(IOException.class)
                .hasMessageContaining("expected exception.");
    }