void append(Buffer buffer, int subpartitionId, boolean flush) {
        subpartitionCacheManagers[subpartitionId].append(buffer);
        increaseNumCachedBytesAndCheckFlush(buffer.readableBytes(), flush);
    }
@Test
    void testAppend() {
        int numAddBuffers = 100;
        int maxBufferSizeBytes = 100;
        int subpartitionId = 0;
        Random random = new Random();

        TestingTieredStorageMemoryManager memoryManager =
                new TestingTieredStorageMemoryManager.Builder().build();

        AtomicInteger numReceivedBuffers = new AtomicInteger(0);
        AtomicInteger numReceivedBytes = new AtomicInteger(0);
        TestingPartitionFileWriter partitionFileWriter =
                new TestingPartitionFileWriter.Builder()
                        .setWriteFunction(
                                (partitionId, subpartitionBufferContexts) -> {
                                    Tuple2<Integer, Integer> numBuffersAndBytes =
                                            getNumReceivedBuffersAndBytes(
                                                    subpartitionBufferContexts);
                                    numReceivedBuffers.getAndAdd(numBuffersAndBytes.f0);
                                    numReceivedBytes.getAndAdd(numBuffersAndBytes.f1);
                                    return FutureUtils.completedVoidFuture();
                                })
                        .build();
        DiskCacheManager diskCacheManager =
                new DiskCacheManager(
                        TieredStorageIdMappingUtils.convertId(new ResultPartitionID()),
                        1,
                        1024,
                        memoryManager,
                        partitionFileWriter);

        // Append the buffers to the disk cache manager
        int numExpectBytes = 0;
        for (int i = 0; i < numAddBuffers; i++) {
            int bufferSizeBytes = random.nextInt(maxBufferSizeBytes) + 1;
            numExpectBytes += bufferSizeBytes;
            diskCacheManager.append(
                    BufferBuilderTestUtils.buildSomeBuffer(bufferSizeBytes), subpartitionId, true);
        }

        assertThat(diskCacheManager.getBufferIndex(subpartitionId)).isEqualTo(numAddBuffers);
        diskCacheManager.close();

        assertThat(numReceivedBuffers).hasValue(numAddBuffers);
        assertThat(numReceivedBytes).hasValue(numExpectBytes);
    }