@Override
    public ExtractNode createExtractNode(StreamNode streamNodeInfo) {
        MySQLBinlogSource binlogSource = (MySQLBinlogSource) streamNodeInfo;
        FieldTypeMappingStrategy fieldTypeMappingStrategy =
                fieldTypeStrategyFactory.getInstance(binlogSource.getSourceType());
        List<FieldInfo> fieldInfos = parseStreamFieldInfos(binlogSource.getFieldList(), binlogSource.getSourceName(),
                fieldTypeMappingStrategy);
        Map<String, String> properties = parseProperties(binlogSource.getProperties());

        final String database = binlogSource.getDatabaseWhiteList();
        final String primaryKey = binlogSource.getPrimaryKey();
        final String hostName = binlogSource.getHostname();
        final String username = binlogSource.getUser();
        final String password = binlogSource.getPassword();
        final Integer port = binlogSource.getPort();
        Integer serverId = null;
        if (binlogSource.getServerId() != null && binlogSource.getServerId() > 0) {
            serverId = binlogSource.getServerId();
        }
        String tables = binlogSource.getTableWhiteList();
        final List<String> tableNames = Splitter.on(",").splitToList(tables);
        final String serverTimeZone = binlogSource.getServerTimezone();

        // TODO Needs to be configurable for those parameters
        if (binlogSource.isAllMigration()) {
            // Unique properties when migrate all tables in database
            properties.put("migrate-all", "true");
        }
        if (binlogSource.isOnlyIncremental()) {
            properties.put("scan.startup.mode", "latest-offset");
        }

        return new MySqlExtractNode(binlogSource.getSourceName(),
                binlogSource.getSourceName(),
                fieldInfos,
                null,
                properties,
                primaryKey,
                tableNames,
                hostName,
                username,
                password,
                database,
                port,
                serverId,
                true,
                serverTimeZone);
    }
The Provider for creating MySQLBinlog extract nodes.
inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/node/provider/MySQLBinlogProvider.java