@Override
    public ExtractNode createExtractNode(StreamNode streamNodeInfo) {
        KafkaSource kafkaSource = (KafkaSource) streamNodeInfo;
        List<FieldInfo> fieldInfos = parseStreamFieldInfos(kafkaSource.getFieldList(), kafkaSource.getSourceName());
        Map<String, String> properties = parseProperties(kafkaSource.getProperties());

        Format format = parsingFormat(
                kafkaSource.getSerializationType(),
                kafkaSource.getWrapType(),
                kafkaSource.getDataSeparator(),
                kafkaSource.getKvSeparator(),
                kafkaSource.getDataEscapeChar(),
                kafkaSource.getIgnoreParseError());

        KafkaScanStartupMode startupMode = parseStartupMode(kafkaSource.getAutoOffsetReset());
        String topic = kafkaSource.getTopic();
        String bootstrapServers = kafkaSource.getBootstrapServers();

        final String primaryKey = kafkaSource.getPrimaryKey();
        String groupId = kafkaSource.getGroupId();
        String partitionOffset = kafkaSource.getPartitionOffsets();
        String scanTimestampMillis = kafkaSource.getTimestampMillis();
        return new KafkaExtractNode(kafkaSource.getSourceName(),
                kafkaSource.getSourceName(),
                fieldInfos,
                null,
                properties,
                topic,
                bootstrapServers,
                format,
                startupMode,
                primaryKey,
                groupId,
                partitionOffset,
                scanTimestampMillis);
    }
The Provider for creating Kafka extract or load nodes.
inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/node/provider/KafkaProvider.java