@Override
    public void releaseDataView() {
        memoryDataManagerOperation.onConsumerReleased(subpartitionId, consumerId);
    }
@Test
    void testRelease() {
        CompletableFuture<HsConsumerId> consumerReleasedFuture = new CompletableFuture<>();
        TestingMemoryDataManagerOperation memoryDataManagerOperation =
                TestingMemoryDataManagerOperation.builder()
                        .setOnConsumerReleasedBiConsumer(
                                (subpartitionId, consumerId) -> {
                                    consumerReleasedFuture.complete(consumerId);
                                })
                        .build();
        HsConsumerId consumerId = HsConsumerId.newId(null);
        HsSubpartitionConsumerMemoryDataManager subpartitionConsumerMemoryDataManager =
                createSubpartitionConsumerMemoryDataManager(consumerId, memoryDataManagerOperation);
        subpartitionConsumerMemoryDataManager.releaseDataView();
        assertThat(consumerReleasedFuture).isCompletedWithValue(consumerId);
    }