@Override
    public void notifyCheckpointComplete(long checkpointId) throws Exception {
        this.minHeap.add(checkpointId);
        if (this.minHeap.size() <= this.lsnCommitCheckpointsDelay) {
            LOG.info("Pending checkpoints '{}'.", this.minHeap);
            return;
        }
        final long checkpointIdToCommit = this.minHeap.poll();
        LOG.info(
                "Pending checkpoints '{}', to be committed checkpoint id '{}'.",
                this.minHeap,
                checkpointIdToCommit);

        // After all snapshot splits are finished, update stream split's metadata and reset start
        // offset, which maybe smaller than before.
        // In case that new start-offset of stream split has been recycled, don't commit offset
        // during new table added phase.
        if (isCommitOffset()) {
            super.notifyCheckpointComplete(checkpointIdToCommit);
        }
    }
@Test
    void testNotifyCheckpointWindowSizeOne() throws Exception {
        final PostgresSourceReader reader = createReader(1);
        final List<Long> completedCheckpointIds = new ArrayList<>();
        MockPostgresDialect.setNotifyCheckpointCompleteCallback(
                id -> completedCheckpointIds.add(id));
        reader.notifyCheckpointComplete(11L);
        assertThat(completedCheckpointIds).isEmpty();
        reader.notifyCheckpointComplete(12L);
        assertThat(completedCheckpointIds).containsExactly(11L);
        reader.notifyCheckpointComplete(13L);
        assertThat(completedCheckpointIds).containsExactly(11L, 12L);
    }