private void executeBackfillTask(
            PostgresSourceFetchTaskContext ctx,
            SnapshotSplitChangeEventSourceContext changeEventSourceContext)
            throws InterruptedException {
        final StreamSplit backfillSplit =
                new StreamSplit(
                        split.splitId(),
                        changeEventSourceContext.getLowWatermark(),
                        changeEventSourceContext.getHighWatermark(),
                        new ArrayList<>(),
                        split.getTableSchemas(),
                        0);

        // optimization that skip the WAL read when the low watermark >= high watermark
        final boolean backfillRequired =
                backfillSplit.getEndingOffset().isAtOrAfter(backfillSplit.getStartingOffset());
        // lk_test TODO maybe has bug
        LOG.info("in executeBackfillTask get getEndingOffset:{}, getStartingOffset:{}, backfillRequired:{}",
                backfillSplit.getEndingOffset(), backfillSplit.getStartingOffset(), backfillRequired);
        if (backfillRequired) {
            LOG.info(
                    "Skip the backfill {} for split {}: low watermark >= high watermark",
                    backfillSplit,
                    split);
            ctx.getDispatcher()
                    .dispatchWatermarkEvent(
                            ctx.getOffsetContext().getPartition(),
                            backfillSplit,
                            backfillSplit.getEndingOffset(),
                            WatermarkKind.END);

            taskRunning = false;
            return;
        }

        final PostgresOffsetContext.Loader loader =
                new PostgresOffsetContext.Loader(ctx.getDbzConnectorConfig());
        final PostgresOffsetContext postgresOffsetContext =
                loader.load(backfillSplit.getStartingOffset().getOffset());

        // we should only capture events for the current table,
        // otherwise, we may not find corresponding schema
        Configuration dbzConf =
                ctx.getDbzConnectorConfig()
                        .getConfig()
                        .edit()
                        .with("table.include.list", split.getTableId().toString())
                        // Disable heartbeat event in snapshot split fetcher
                        .with(Heartbeat.HEARTBEAT_INTERVAL, 0)
                        .build();

        final PostgresStreamFetchTask.StreamSplitReadTask backfillReadTask =
                new PostgresStreamFetchTask.StreamSplitReadTask(
                        new PostgresConnectorConfig(dbzConf),
                        ctx.getSnapShotter(),
                        ctx.getConnection(),
                        ctx.getDispatcher(),
                        ctx.getErrorHandler(),
                        ctx.getTaskContext().getClock(),
                        ctx.getDatabaseSchema(),
                        ctx.getTaskContext(),
                        ctx.getReplicationConnection(),
                        backfillSplit);
        LOG.info("Execute backfillReadTask for split {}", split);
        backfillReadTask.execute(new PostgresChangeEventSourceContext(), postgresOffsetContext);
    }
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