@Override
    public void execute(Context context) throws Exception {
        LOG.info("Execute ScanFetchTask for split: {}", split);
        PostgresSourceFetchTaskContext ctx = (PostgresSourceFetchTaskContext) context;
        LOG.info("lk_test Execute ScanFetchTask ctx.getOffsetContext: {}", ctx.getOffsetContext());

        taskRunning = true;

        SnapshotSplitReadTask snapshotSplitReadTask =
                new SnapshotSplitReadTask(
                        ctx.getConnection(),
                        ctx.getDbzConnectorConfig(),
                        ctx.getDatabaseSchema(),
                        ctx.getOffsetContext(),
                        ctx.getDispatcher(),
                        ctx.getSnapshotChangeEventSourceMetrics(),
                        split);

        SnapshotSplitChangeEventSourceContext changeEventSourceContext =
                new SnapshotSplitChangeEventSourceContext();
        SnapshotResult<PostgresOffsetContext> snapshotResult =
                snapshotSplitReadTask.execute(changeEventSourceContext, ctx.getOffsetContext());

        if (!snapshotResult.isCompletedOrSkipped()) {
            LOG.error("snapshotResult.isCompletedOrSkipped throw.");
            taskRunning = false;
            throw new IllegalStateException(
                    String.format("Read snapshot for postgres split %s fail", split));
        }

        executeBackfillTask(ctx, changeEventSourceContext);
    }
A {@link FetchTask} implementation for Postgres to read snapshot split.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/source/fetch/PostgresScanFetchTask.java