@Override
    public void setChannelStateWriter(ChannelStateWriter channelStateWriter) {
        checkState(this.channelStateWriter == null, "Already initialized");
        this.channelStateWriter = checkNotNull(channelStateWriter);
    }
@TestTemplate
    void testConsumeTimeoutableCheckpointBarrierQuickly() throws Exception {
        PipelinedSubpartition subpartition = createSubpartition();
        subpartition.setChannelStateWriter(ChannelStateWriter.NO_OP);
        assertSubpartitionChannelStateFuturesAndQueuedBuffers(subpartition, null, true, 0, false);

        // test without data buffer
        testConsumeQuicklyWithNDataBuffers(0, subpartition, 5L);

        // test with data buffer
        testConsumeQuicklyWithNDataBuffers(1, subpartition, 6L);
        testConsumeQuicklyWithNDataBuffers(2, subpartition, 7L);
    }