public void consumed(HsConsumerId consumerId) {
        checkState(!released, "Buffer is already released.");
        checkState(consumed.add(consumerId), "Consume buffer repeatedly is unexpected.");
        // increase ref count when buffer is consumed, will be decreased when downstream finish
        // consuming.
        buffer.retainBuffer();
    }
@Test
    void testBufferConsumedRepeatedly() {
        final HsConsumerId consumerId = HsConsumerId.DEFAULT;
        bufferContext.consumed(consumerId);
        assertThatThrownBy(() -> bufferContext.consumed(consumerId))
                .isInstanceOf(IllegalStateException.class)
                .hasMessageContaining("Consume buffer repeatedly is unexpected.");
    }