private void checkErrorAndRethrow() {
        Throwable cause = cursorCommitThrowable.get();
        if (cause != null) {
            throw new RuntimeException("An error occurred in acknowledge message.", cause);
        }
    }
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/PulsarOrderedSourceReader.java