@Override
    public void releaseAllResources() {
        releaseInternal(null);
    }
@Test
    void testReleaseAllResources() throws Exception {
        int numSegments = 5;
        Queue<MemorySegment> segments = createsMemorySegments(numSegments);

        try {
            CountingAvailabilityListener listener = new CountingAvailabilityListener();
            SortMergeSubpartitionReader subpartitionReader =
                    createSortMergeSubpartitionReader(listener);

            subpartitionReader.readBuffers(segments, segments::add);
            assertThat(listener.numNotifications).isEqualTo(1);
            assertThat(subpartitionReader.unsynchronizedGetNumberOfQueuedBuffers()).isEqualTo(4);

            subpartitionReader.releaseAllResources();
            assertThat(subpartitionReader.getReleaseFuture()).isDone();
            assertThat(subpartitionReader.unsynchronizedGetNumberOfQueuedBuffers()).isZero();
            assertThat(subpartitionReader.getAvailabilityAndBacklog(false).isAvailable()).isTrue();
            assertThat(subpartitionReader.isReleased()).isTrue();

            assertThat(listener.numNotifications).isEqualTo(1);
            assertThat(subpartitionReader.getFailureCause()).isNull();
        } finally {
            assertThat(segments).hasSize(numSegments);
        }
    }