@Override
    public void createSinkResource(SinkInfo sinkInfo) {
        LOGGER.info("begin to create kafka topic for sinkId={}", sinkInfo.getId());

        KafkaSinkDTO kafkaInfo = KafkaSinkDTO.getFromJson(sinkInfo.getExtParams());
        String topicName = kafkaInfo.getTopicName();
        Integer partitionNum = kafkaInfo.getPartitionNum();
        Preconditions.expectNotBlank(topicName, ErrorCodeEnum.INVALID_PARAMETER, "topic name cannot be empty");
        Preconditions.expectNotNull(partitionNum, ErrorCodeEnum.INVALID_PARAMETER, "partition cannot be empty");

        try (Admin admin = getKafkaAdmin(kafkaInfo.getBootstrapServers())) {
            boolean topicExists = isTopicExists(admin, topicName, partitionNum);
            if (!topicExists) {
                CreateTopicsResult result = admin.createTopics(Collections.singleton(
                        new NewTopic(topicName, Optional.of(partitionNum), Optional.empty())));
                result.values().get(topicName).get();
            }
            sinkService.updateStatus(sinkInfo.getId(), SinkStatus.CONFIG_SUCCESSFUL.getCode(),
                    "create kafka topic success");
            LOGGER.info("success to create kafka topic [{}] for sinkInfo={}", topicName, sinkInfo);
        } catch (Throwable e) {
            LOGGER.error("create kafka topic error, ", e);
            sinkService.updateStatus(sinkInfo.getId(), SinkStatus.CONFIG_FAILED.getCode(), e.getMessage());
            throw new WorkflowException("create kafka topic failed, reason: " + e.getMessage());
        }
        this.assignCluster(sinkInfo);
    }
Kafka resource operator for creating Kafka topic
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/kafka/KafkaResourceOperator.java