public void append(ByteBuffer record, DataType dataType) throws InterruptedException {
        if (dataType.isEvent()) {
            writeEvent(record, dataType);
        } else {
            writeRecord(record, dataType);
        }
    }
@Test
    void testAppendEventFinishCurrentBuffer() throws Exception {
        bufferSize = RECORD_SIZE * 3;
        AtomicInteger finishedBuffers = new AtomicInteger(0);
        HsMemoryDataManagerOperation memoryDataManagerOperation =
                TestingMemoryDataManagerOperation.builder()
                        .setRequestBufferFromPoolSupplier(() -> createBufferBuilder(bufferSize))
                        .setOnBufferFinishedRunnable(finishedBuffers::incrementAndGet)
                        .build();
        HsSubpartitionMemoryDataManager subpartitionMemoryDataManager =
                createSubpartitionMemoryDataManager(memoryDataManagerOperation);
        subpartitionMemoryDataManager.append(createRecord(0), DataType.DATA_BUFFER);
        subpartitionMemoryDataManager.append(createRecord(1), DataType.DATA_BUFFER);
        assertThat(finishedBuffers).hasValue(0);
        subpartitionMemoryDataManager.append(createRecord(2), DataType.EVENT_BUFFER);
        assertThat(finishedBuffers).hasValue(2);
    }