@Override
    public LoadNode createLoadNode(StreamNode nodeInfo, Map<String, StreamField> constantFieldMap) {
        KafkaSink kafkaSink = (KafkaSink) nodeInfo;
        Map<String, String> properties = parseProperties(kafkaSink.getProperties());
        List<FieldInfo> fieldInfos = parseSinkFieldInfos(kafkaSink.getSinkFieldList(), kafkaSink.getSinkName());
        List<FieldRelation> fieldRelations = parseSinkFields(kafkaSink.getSinkFieldList(), constantFieldMap);

        Integer sinkParallelism = kafkaSink.getPartitionNum();
        Format format = parseFormat(kafkaSink.getSerializationType());

        return new KafkaLoadNode(
                kafkaSink.getSinkName(),
                kafkaSink.getSinkName(),
                fieldInfos,
                fieldRelations,
                Lists.newArrayList(),
                null,
                kafkaSink.getTopicName(),
                kafkaSink.getBootstrapServers(),
                format,
                sinkParallelism,
                properties,
                kafkaSink.getPrimaryKey());
    }
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