public abstract void prepareSnapshotPreBarrier(long checkpointId) throws Exception;
@Test
    void testPrepareCheckpointPreBarrier() throws Exception {
        final AtomicInteger intRef = new AtomicInteger();

        final OneInputStreamOperator<String, String> one = new ValidatingOperator(intRef, 0);
        final OneInputStreamOperator<String, String> two = new ValidatingOperator(intRef, 1);
        final OneInputStreamOperator<String, String> three = new ValidatingOperator(intRef, 2);

        final OperatorChain<?, ?> chain = setupOperatorChain(one, two, three);
        chain.prepareSnapshotPreBarrier(ValidatingOperator.CHECKPOINT_ID);

        assertThat(intRef).hasValue(3);
    }