private PostgresOffsetContext loadStartingOffsetState(
            PostgresOffsetContext.Loader loader, SourceSplitBase sourceSplitBase) {
        Offset offset =
                sourceSplitBase.isSnapshotSplit()
                        ? new PostgresOffsetFactory()
                                .createInitialOffset() // get an offset for starting snapshot
                        : sourceSplitBase.asStreamSplit().getStartingOffset();

        PostgresOffsetContext offsetContext =
                loader.load(
                        Objects.requireNonNull(offset, "offset is null for the sourceSplitBase")
                                .getOffset());
        return offsetContext;
    }
The context of {@link PostgresScanFetchTask} and {@link PostgresStreamFetchTask}.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/source/fetch/PostgresSourceFetchTaskContext.java