public Collection<? extends CheckpointCommittableManager<CommT>> getCheckpointCommittablesUpTo(
            long checkpointId) {
        // clean up fully committed previous checkpoints
        // this wouldn't work with concurrent unaligned checkpoints
        Collection<CheckpointCommittableManagerImpl<CommT>> checkpoints =
                checkpointCommittables.headMap(checkpointId, true).values();
        checkpoints.removeIf(CheckpointCommittableManagerImpl::isFinished);
        return checkpoints;
    }
@Test
    void testGetCheckpointCommittablesUpTo() {
        final CommittableCollector<Integer> committableCollector =
                new CommittableCollector<>(1, 1, METRIC_GROUP);
        CommittableSummary<Integer> first = new CommittableSummary<>(1, 1, 1L, 1, 0, 0);
        committableCollector.addMessage(first);
        CommittableSummary<Integer> second = new CommittableSummary<>(1, 1, 2L, 1, 0, 0);
        committableCollector.addMessage(second);
        committableCollector.addMessage(new CommittableSummary<>(1, 1, 3L, 1, 0, 0));

        assertThat(committableCollector.getCheckpointCommittablesUpTo(2)).hasSize(2);

        assertThat(committableCollector.getEndOfInputCommittable()).isNull();
    }