public void commitCurrentOffset() {
        if (streamSplitReadTask != null && streamSplitReadTask.offsetContext != null) {
            PostgresOffsetContext postgresOffsetContext = streamSplitReadTask.offsetContext;

            // only extracting and storing the lsn of the last commit
            Long commitLsn =
                    (Long) postgresOffsetContext
                            .getOffset()
                            .get(PostgresOffsetContext.LAST_COMMIT_LSN_KEY);
            if (commitLsn != null
                    && (lastCommitLsn == null
                            || Lsn.valueOf(commitLsn).compareTo(Lsn.valueOf(lastCommitLsn)) > 0)) {
                lastCommitLsn = commitLsn;

                Map<String, Object> offsets = new HashMap<>();
                offsets.put(PostgresOffsetContext.LAST_COMMIT_LSN_KEY, lastCommitLsn);
                LOG.debug(
                        "Committing offset {} for {}",
                        Lsn.valueOf(lastCommitLsn),
                        streamSplitReadTask.streamSplit);
                streamSplitReadTask.commitOffset(offsets);
            }
        }
    }
A {@link FetchTask} implementation for Postgres to read streaming changes.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/source/fetch/PostgresStreamFetchTask.java