private void createTopic(SinkInfo sinkInfo) {
        PulsarSinkDTO pulsarSinkDTO = PulsarSinkDTO.getFromJson(sinkInfo.getExtParams());
        PulsarDataNodeDTO pulsarDataNodeInfo = getPulsarDataNodeInfo(sinkInfo);
        try {

            PulsarClusterInfo pulsarClusterInfo = PulsarClusterInfo.builder().adminUrl(pulsarDataNodeInfo.getAdminUrl())
                    .token(pulsarDataNodeInfo.getToken()).build();
            // create pulsar tenant
            pulsarOperator.createTenant(pulsarClusterInfo, pulsarSinkDTO.getPulsarTenant());
            // use default config to create namespace
            InlongPulsarInfo pulsarInfo = new InlongPulsarInfo();
            pulsarOperator.createNamespace(pulsarClusterInfo, pulsarInfo, pulsarSinkDTO.getPulsarTenant(),
                    pulsarSinkDTO.getNamespace());
            String queueModel = pulsarSinkDTO.getPartitionNum() > 0 ? InlongConstants.PULSAR_QUEUE_TYPE_PARALLEL
                    : InlongConstants.PULSAR_QUEUE_TYPE_SERIAL;
            PulsarTopicInfo topicInfo = PulsarTopicInfo.builder().pulsarTenant(pulsarSinkDTO.getPulsarTenant())
                    .namespace(pulsarSinkDTO.getNamespace())
                    .topicName(pulsarSinkDTO.getTopic())
                    .numPartitions(pulsarSinkDTO.getPartitionNum())
                    .queueModule(queueModel)
                    .build();
            // create topic
            pulsarOperator.createTopic(pulsarClusterInfo, topicInfo);
            final String info = "success to create Pulsar resource";
            sinkService.updateStatus(sinkInfo.getId(), SinkStatus.CONFIG_SUCCESSFUL.getCode(), info);
            LOG.info(info + " for sinkInfo={}", sinkInfo);
        } catch (Exception e) {
            LOG.error("init pulsar admin error", e);
            throw new BusinessException();
        }
    }
Pulsar resource operate for creating pulsar resource
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/pulsar/PulsarResourceOperator.java