@Override
        public void afterHandleScn(OracleOffsetContext offsetContext) {
            super.afterHandleScn(offsetContext);
            // check do we need to stop for fetch redo log for snapshot split.
            if (isBoundedRead()) {
                final RedoLogOffset currentRedoLogOffset =
                        getCurrentRedoLogOffset(offsetContext.getOffset());
                // reach the high watermark, the redo log fetcher should be finished
                if (currentRedoLogOffset.isAtOrAfter(redoLogSplit.getEndingOffset())) {
                    // send binlog end event
                    try {
                        dispatcher.dispatchWatermarkEvent(
                                offsetContext.getPartition(),
                                redoLogSplit,
                                currentRedoLogOffset,
                                WatermarkKind.END);
                    } catch (InterruptedException e) {
                        LOG.error("Send signal event error.", e);
                        errorHandler.setProducerThrowable(
                                new DebeziumException("Error processing binlog signal event", e));
                    }
                    // tell fetcher the redo log task finished
                    ((OracleScanFetchTask.SnapshotStreamSplitChangeEventSourceContext) context)
                            .finished();
                }
            }
        }
A wrapped task to read all redo log for table and also supports read bounded (from lowWatermark
to highWatermark) binlog.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/reader/fetch/OracleStreamFetchTask.java