@GuardedBy("consumerLock")
    // this method only called from subpartitionMemoryDataManager with write lock.
    public boolean addBuffer(HsBufferContext bufferContext) {
        tryIncreaseBacklog(bufferContext.getBuffer());
        unConsumedBuffers.add(bufferContext);
        trimHeadingReleasedBuffers();
        return unConsumedBuffers.size() <= 1;
    }
@Test
    void testAddBuffer() {
        TestingMemoryDataManagerOperation memoryDataManagerOperation =
                TestingMemoryDataManagerOperation.builder().build();
        HsSubpartitionConsumerMemoryDataManager subpartitionConsumerMemoryDataManager =
                createSubpartitionConsumerMemoryDataManager(memoryDataManagerOperation);
        ArrayDeque<HsBufferContext> initialBuffers = new ArrayDeque<>();
        initialBuffers.add(createBufferContext(0, false));
        initialBuffers.add(createBufferContext(1, false));
        subpartitionConsumerMemoryDataManager.addInitialBuffers(initialBuffers);
        subpartitionConsumerMemoryDataManager.addBuffer(createBufferContext(2, true));

        assertThat(subpartitionConsumerMemoryDataManager.consumeBuffer(0, Collections.emptyList()))
                .hasValueSatisfying(
                        bufferAndBacklog -> {
                            assertThat(bufferAndBacklog.getSequenceNumber()).isEqualTo(0);
                            assertThat(bufferAndBacklog.buffer().getDataType())
                                    .isEqualTo(Buffer.DataType.DATA_BUFFER);
                        });
        assertThat(subpartitionConsumerMemoryDataManager.consumeBuffer(1, Collections.emptyList()))
                .hasValueSatisfying(
                        bufferAndBacklog -> {
                            assertThat(bufferAndBacklog.getSequenceNumber()).isEqualTo(1);
                            assertThat(bufferAndBacklog.buffer().getDataType())
                                    .isEqualTo(Buffer.DataType.DATA_BUFFER);
                        });
        assertThat(subpartitionConsumerMemoryDataManager.consumeBuffer(2, Collections.emptyList()))
                .hasValueSatisfying(
                        bufferAndBacklog -> {
                            assertThat(bufferAndBacklog.getSequenceNumber()).isEqualTo(2);
                            assertThat(bufferAndBacklog.buffer().getDataType())
                                    .isEqualTo(Buffer.DataType.EVENT_BUFFER);
                        });
    }