private RedoLogSplitReadTask createBackfillRedoLogReadTask(
            StreamSplit backfillRedoLogSplit, OracleSourceFetchTaskContext context) {
        OracleConnectorConfig oracleConnectorConfig =
                context.getSourceConfig().getDbzConnectorConfig();
        final OffsetContext.Loader<OracleOffsetContext> loader =
                new LogMinerOracleOffsetContextLoader(oracleConnectorConfig);
        final OracleOffsetContext oracleOffsetContext =
                loader.load(backfillRedoLogSplit.getStartingOffset().getOffset());
        // we should only capture events for the current table,
        // otherwise, we may can't find corresponding schema
        Configuration dezConf =
                context.getSourceConfig()
                        .getDbzConfiguration()
                        .edit()
                        .with("table.include.list", split.getTableId().toString())
                        // Disable heartbeat event in snapshot split fetcher
                        .with(Heartbeat.HEARTBEAT_INTERVAL, 0)
                        .build();
        // task to read redo log and backfill for current split
        return new RedoLogSplitReadTask(
                new OracleConnectorConfig(dezConf),
                createOracleConnection(context.getSourceConfig().getDbzConfiguration()),
                context.getDispatcher(),
                context.getErrorHandler(),
                context.getDatabaseSchema(),
                context.getSourceConfig().getOriginDbzConnectorConfig(),
                context.getStreamingChangeEventSourceMetrics(),
                backfillRedoLogSplit);
    }
The task to work for fetching data of Oracle table snapshot split.
 Copy from com.ververica:flink-connector-oracle-cdc:2.3.0
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/reader/fetch/OracleScanFetchTask.java