public void append(ByteBuffer record, int targetChannel, Buffer.DataType dataType)
            throws IOException {
        try {
            getSubpartitionMemoryDataManager(targetChannel).append(record, dataType);
        } catch (InterruptedException e) {
            throw new IOException(e);
        }
    }
@Test
    void testAppendMarkBufferFinished() throws Exception {
        AtomicInteger finishedBuffers = new AtomicInteger(0);
        HsSpillingStrategy spillingStrategy =
                TestingSpillingStrategy.builder()
                        .setOnBufferFinishedFunction(
                                (numTotalUnSpillBuffers, currentPoolSize) -> {
                                    finishedBuffers.incrementAndGet();
                                    return Optional.of(Decision.NO_ACTION);
                                })
                        .build();
        bufferSize = Integer.BYTES * 3;
        HsMemoryDataManager memoryDataManager = createMemoryDataManager(spillingStrategy);

        memoryDataManager.append(createRecord(0), 0, Buffer.DataType.DATA_BUFFER);
        memoryDataManager.append(createRecord(1), 0, Buffer.DataType.DATA_BUFFER);
        assertThat(finishedBuffers).hasValue(0);

        memoryDataManager.append(createRecord(2), 0, Buffer.DataType.DATA_BUFFER);
        assertThat(finishedBuffers).hasValue(1);

        memoryDataManager.append(createRecord(3), 0, Buffer.DataType.DATA_BUFFER);
        memoryDataManager.append(createRecord(4), 0, Buffer.DataType.EVENT_BUFFER);
        assertThat(finishedBuffers).hasValue(3);
    }