private void createKafkaTopic(InlongKafkaInfo kafkaInfo, String topicName) throws Exception {
        // create Kafka topic
        ClusterInfo clusterInfo = clusterService.getOne(kafkaInfo.getInlongClusterTag(), null, ClusterType.KAFKA);
        kafkaOperator.createTopic(kafkaInfo, (KafkaClusterInfo) clusterInfo, topicName);

        boolean exist = kafkaOperator.topicIsExists((KafkaClusterInfo) clusterInfo, topicName);
        if (!exist) {
            String bootStrapServers = clusterInfo.getUrl();
            log.error("topic={} not exists in {}", topicName, bootStrapServers);
            throw new WorkflowListenerException("topic=" + topicName + " not exists in " + bootStrapServers);
        }

        // Kafka consumers do not need to register in advance

        // save the consumer group info for the Kafka topic
        String consumeGroup = String.format(KAFKA_CONSUMER_GROUP, kafkaInfo.getInlongClusterTag(), topicName);
        Integer id = consumeService.saveBySystem(kafkaInfo, topicName, consumeGroup);
        log.info("success to save inlong consume [{}] for consumerGroup={}, groupId={}, topic={}",
                id, consumeGroup, kafkaInfo.getInlongGroupId(), topicName);
    }
Create Kafka Topic and Subscription, and save the consumer group info.
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/kafka/KafkaQueueResourceOperator.java