@Override
    public void processElement(StreamRecord<CommittableMessage<CommT>> element) throws Exception {
        committableCollector.addMessage(element.getValue());
    }
@Test
    void testWaitForCommittablesOfLatestCheckpointBeforeCommitting() throws Exception {
        final MockCommitter committer = new MockCommitter();
        final OneInputStreamOperatorTestHarness<CommittableMessage<Integer>, Void> testHarness =
                createTestHarness(committer);
        testHarness.open();

        testHarness.processElement(new StreamRecord<>(new CommittableSummary<>(1, 1, 1L, 2, 0, 0)));

        testHarness.processElement(new StreamRecord<>(new CommittableWithLineage<>(1, 1L, 1)));

        testHarness.notifyOfCompletedCheckpoint(1);

        assertThat(testHarness.getOutput()).isEmpty();
        // Not committed because incomplete
        assertThat(committer.committed).isEmpty();

        testHarness.processElement(new StreamRecord<>(new CommittableWithLineage<>(2, 1L, 1)));

        testHarness.notifyOfCompletedCheckpoint(2);

        assertThat(testHarness.getOutput()).isEmpty();
        assertThat(committer.committed).containsExactly(1, 2);
        testHarness.close();
    }