private PulsarDataNodeDTO getPulsarDataNodeInfo(SinkInfo sinkInfo) {

        String dataNodeName = sinkInfo.getDataNodeName();
        Preconditions.expectNotBlank(dataNodeName, ErrorCodeEnum.INVALID_PARAMETER,
                "Pulsar admin url not specified and data node is empty");
        PulsarDataNodeInfo dataNodeInfo = (PulsarDataNodeInfo) dataNodeHelper.getDataNodeInfo(
                dataNodeName, sinkInfo.getSinkType());
        PulsarDataNodeDTO pulsarDataNodeDTO = new PulsarDataNodeDTO();
        CommonBeanUtils.copyProperties(dataNodeInfo, pulsarDataNodeDTO, true);
        return pulsarDataNodeDTO;
    }
Pulsar resource operate for creating pulsar resource
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/pulsar/PulsarResourceOperator.java