@Override
    public void notifyCheckpointComplete(long checkpointId) throws Exception {
        if (!running) {
            log.info("notifyCheckpointComplete() called on closed source");
            return;
        }

        PulsarFetcher<T> fetcher = this.pulsarFetcher;

        if (fetcher == null) {
            log.info("notifyCheckpointComplete() called on uninitialized source");
            return;
        }

        log.debug("Source {} received confirmation for unknown checkpoint id {}",
                taskIndex, checkpointId);

        try {
            if (!pendingOffsetsToCommit.containsKey(checkpointId)) {
                log.warn("Source {} received confirmation for unknown checkpoint id {}",
                        taskIndex, checkpointId);
                return;
            }

            Map<TopicRange, MessageId> offset = pendingOffsetsToCommit.get(checkpointId);

            // remove older checkpoints in map
            Iterator<Long> iterator = pendingOffsetsToCommit.keySet().iterator();
            while (iterator.hasNext()) {
                Long key = iterator.next();
                iterator.remove();
                if (Objects.equals(key, checkpointId)) {
                    break;
                }
            }

            if (offset == null || offset.size() == 0) {
                log.debug("Source {} has empty checkpoint state", taskIndex);
                return;
            }
            fetcher.commitOffsetToState(offset);
        } catch (Exception e) {
            if (running) {
                throw e;
            }
        }
    }
Try to restore the old save point.

@param stateStore state store
@return state data
@throws Exception Type incompatibility, serialization failure
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/internal/FlinkPulsarSourceWithoutAdmin.java