private MySqlBinlogSplit createBinlogSplit() {
        Map<TableId, TableChange> tableSchemas = discoverCapturedTableSchemas();
        return new MySqlBinlogSplit(
                BINLOG_SPLIT_ID,
                sourceConfig.getStartupOptions().binlogOffset,
                BinlogOffset.ofNonStopping(),
                new ArrayList<>(),
                tableSchemas,
                0);
    }
A {@link MySqlSplitAssigner} which only read binlog from current binlog position.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/assigners/MySqlBinlogSplitAssigner.java