private Map<TableId, TableChange> discoverCapturedTableSchemas() {
        final List<TableId> capturedTableIds;
        try {
            capturedTableIds = listTables(jdbc, tableFilters);
        } catch (SQLException e) {
            throw new FlinkRuntimeException("Failed to discover captured tables", e);
        }
        if (capturedTableIds.isEmpty()) {
            throw new IllegalArgumentException(
                    String.format(
                            "Can't find any matched tables,"
                                    + " please check your configured database-name: %s and table-name: %s",
                            sourceConfig.getDbzConfiguration().getString(MySqlSourceOptions.DATABASE_NAME.key()),
                            sourceConfig.getDbzConfiguration().getString(MySqlSourceOptions.TABLE_NAME.key())));
        }

        // fetch table schemas
        MySqlSchema mySqlSchema = new MySqlSchema(sourceConfig, false);
        Map<TableId, TableChange> tableSchemas = new HashMap<>();
        for (TableId tableId : capturedTableIds) {
            TableChange tableSchema = mySqlSchema.getTableSchema(jdbc, tableId);
            tableSchemas.put(tableId, tableSchema);
        }
        return tableSchemas;
    }
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