@Override
    public void releaseAllResources() throws IOException {
        synchronized (lock) {
            for (ResultSubpartitionView view : allViews.values()) {
                view.releaseAllResources();
            }
            allViews.clear();

            for (ResultSubpartitionView view : unregisteredAvailableViews) {
                view.releaseAllResources();
            }
            unregisteredAvailableViews.clear();

            for (Tuple2<ResultSubpartition.BufferAndBacklog, Integer> tuple2 : cachedBuffers) {
                tuple2.f0.buffer().recycleBuffer();
            }
            cachedBuffers.clear();

            isReleased = true;
        }
    }
@Test
    void testReleaseAllResources() throws IOException {
        assertThat(view.isReleased()).isFalse();
        assertThat(view0.isReleased()).isFalse();
        assertThat(view1.isReleased()).isFalse();
        assertThat(buffers0).allMatch(x -> !x.isRecycled());
        assertThat(buffers1).allMatch(x -> !x.isRecycled());

        // Verifies that cached buffers are also recycled.
        view0.notifyDataAvailable();

        view.releaseAllResources();

        assertThat(view.isReleased()).isTrue();
        assertThat(view0.isReleased()).isTrue();
        assertThat(view1.isReleased()).isTrue();
        assertThat(buffers0).allMatch(Buffer::isRecycled);
        assertThat(buffers1).allMatch(Buffer::isRecycled);
    }