@Override
    public void fail(Throwable failureCause) {
        synchronized (lock) {
            if (isFailed) {
                return;
            }
            isFailed = true;
            BufferIndexOrError bufferIndexOrError;
            // empty from tail, in-case subpartition view consumes concurrently and gets the wrong
            // order
            while ((bufferIndexOrError = loadedBuffers.pollLast()) != null) {
                if (bufferIndexOrError.getBuffer().isPresent()) {
                    bufferIndexOrError.getBuffer().get().recycleBuffer();
                    tryDecreaseBacklog(bufferIndexOrError.getBuffer().get());
                }
            }

            loadedBuffers.add(BufferIndexOrError.newError(failureCause));
            operations.notifyDataAvailable();
        }
    }
@Test
    void testFail() throws Exception {
        AtomicInteger numOfNotify = new AtomicInteger(0);
        subpartitionOperation.setNotifyDataAvailableRunnable(numOfNotify::incrementAndGet);
        HsSubpartitionFileReaderImpl subpartitionFileReader = createSubpartitionFileReader();
        Deque<BufferIndexOrError> loadedBuffers = subpartitionFileReader.getLoadedBuffers();
        writeDataToFile(targetChannel, 0, 2);

        subpartitionFileReader.prepareForScheduling();
        Queue<MemorySegment> memorySegments = createsMemorySegments(2);
        // trigger reading, add buffer to queue.
        AtomicInteger numReleased = new AtomicInteger(0);
        subpartitionFileReader.readBuffers(
                memorySegments, (buffer) -> numReleased.incrementAndGet());

        assertThat(memorySegments).isEmpty();
        assertThat(loadedBuffers).hasSize(2);
        assertThat(numOfNotify).hasValue(1);

        subpartitionFileReader.fail(new RuntimeException("expected exception."));
        // all buffers in file reader queue should recycle during fail.
        assertThat(numReleased).hasValue(2);
        BufferIndexOrError error = loadedBuffers.poll();
        assertThat(loadedBuffers).isEmpty();
        assertThat(error).isNotNull();
        assertThat(error.getThrowable())
                .hasValueSatisfying(
                        throwable ->
                                assertThat(throwable)
                                        .isInstanceOf(RuntimeException.class)
                                        .hasMessage("expected exception."));

        // subpartitionReader fail should notify downstream.
        assertThat(numOfNotify).hasValue(2);
    }