@Override
    public void addSplits(List<PulsarPartitionSplit> splits) {
        if (started) {
            // We only accept splits after this reader is started and registered to the pipeline.
            // This would ignore the splits from the state.
            super.addSplits(splits);
        } else {
            // Abort the pending transaction in this split.
            for (PulsarPartitionSplit split : splits) {
                LOG.info("Ignore the split {} saved in checkpoint.", split);

                TxnID transactionId = split.getUncommittedTransactionId();
                if (transactionId != null && coordinatorClient != null) {
                    try {
                        coordinatorClient.abort(transactionId);
                    } catch (Exception e) {
                        LOG.debug(
                                "Error in aborting transaction {} from the checkpoint",
                                transactionId,
                                e);
                    }
                }
            }
        }
    }
The map to store the start time of each checkpoint.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/source/reader/PulsarUnorderedSourceReader.java