@Override
    public Collection<CommittableWithLineage<CommT>> commit(
            boolean fullyReceived, Committer<CommT> committer)
            throws IOException, InterruptedException {
        Collection<CommitRequestImpl<CommT>> requests = getPendingRequests(fullyReceived);
        requests.forEach(CommitRequestImpl::setSelected);
        committer.commit(new ArrayList<>(requests));
        requests.forEach(CommitRequestImpl::setCommittedIfNoError);
        Collection<CommittableWithLineage<CommT>> committed = drainFinished();
        metricGroup.setCurrentPendingCommittablesGauge(() -> getPendingRequests(false).size());
        return committed;
    }
@Test
    void testCommit() throws IOException, InterruptedException {
        final CheckpointCommittableManagerImpl<Integer> checkpointCommittables =
                new CheckpointCommittableManagerImpl<>(1, 1, 1L, METRIC_GROUP);
        checkpointCommittables.upsertSummary(new CommittableSummary<>(1, 1, 1L, 1, 0, 0));
        checkpointCommittables.upsertSummary(new CommittableSummary<>(2, 1, 1L, 2, 0, 0));
        checkpointCommittables.addCommittable(new CommittableWithLineage<>(3, 1L, 1));
        checkpointCommittables.addCommittable(new CommittableWithLineage<>(4, 1L, 2));

        final Committer<Integer> committer = new NoOpCommitter();
        // Only commit fully received committables
        Collection<CommittableWithLineage<Integer>> commitRequests =
                checkpointCommittables.commit(true, committer);
        assertThat(commitRequests)
                .hasSize(1)
                .satisfiesExactly(c -> assertThat(c.getCommittable()).isEqualTo(3));

        // Commit all committables
        commitRequests = checkpointCommittables.commit(false, committer);
        assertThat(commitRequests)
                .hasSize(1)
                .satisfiesExactly(c -> assertThat(c.getCommittable()).isEqualTo(4));
    }