private MySqlOffsetContext loadStartingOffsetState(
            OffsetContext.Loader loader, MySqlSplit mySqlSplit) {
        BinlogOffset offset =
                mySqlSplit.isSnapshotSplit()
                        ? BinlogOffset.ofEarliest()
                        : initializeEffectiveOffset(
                                mySqlSplit.asBinlogSplit().getStartingOffset(), connection);

        LOG.info("Starting offset is initialized to {}", offset);

        MySqlOffsetContext mySqlOffsetContext =
                (MySqlOffsetContext) loader.load(offset.getOffset());

        if (!isBinlogAvailable(mySqlOffsetContext)) {
            throw new IllegalStateException(
                    "The connector is trying to read binlog starting at "
                            + mySqlOffsetContext.getSourceInfo()
                            + ", but this is no longer "
                            + "available on the server. Reconfigure the connector to use a snapshot when needed.");
        }
        return mySqlOffsetContext;
    }
Loads the connector's persistent offset (if present) via the given loader.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/task/context/StatefulTaskContext.java