void close() {
        // Wait the flushing buffers to be completed before closed
        try {
            flushCompletableFuture.get();
        } catch (Exception e) {
            LOG.error("Failed to flush the buffers.", e);
            ExceptionUtils.rethrow(e);
        }
        flushBuffers();
    }
@Test
    void testClose() {
        TieredStoragePartitionId partitionId =
                TieredStorageIdMappingUtils.convertId(new ResultPartitionID());
        int subpartitionId = 0;
        int segmentId = 0;

        AtomicInteger numReceivedBuffers = new AtomicInteger(0);
        TestingPartitionFileWriter partitionFileWriter =
                new TestingPartitionFileWriter.Builder()
                        .setWriteFunction(
                                (ignoredPartitionId, bufferContexts) -> {
                                    numReceivedBuffers.addAndGet(
                                            bufferContexts
                                                    .get(subpartitionId)
                                                    .getSegmentBufferContexts()
                                                    .get(segmentId)
                                                    .getBufferAndIndexes()
                                                    .size());
                                    return FutureUtils.completedVoidFuture();
                                })
                        .build();
        SubpartitionRemoteCacheManager cacheManager =
                new SubpartitionRemoteCacheManager(
                        partitionId,
                        subpartitionId,
                        new TestingTieredStorageMemoryManager.Builder().build(),
                        partitionFileWriter);
        cacheManager.addBuffer(BufferBuilderTestUtils.buildSomeBuffer());
        cacheManager.close();
        assertThat(numReceivedBuffers).hasValue(1);
    }