@Override
    public boolean send(PackProfile profile) {
        String producerTopic = null;
        try {
            // get idConfig
            IdTopicConfig idConfig = ConfigManager.getInstance().getSinkIdTopicConfig(
                    profile.getInlongGroupId(), profile.getInlongStreamId());
            if (idConfig == null) {
                // add default topics first
                if (CommonConfigHolder.getInstance().isEnableUnConfigTopicAccept()) {
                    producerTopic = CommonConfigHolder.getInstance().getRandDefTopics();
                    if (StringUtils.isEmpty(producerTopic)) {
                        sinkContext.fileMetricIncWithDetailStats(
                                StatConstants.EVENT_SINK_DEFAULT_TOPIC_MISSING, profile.getUid());
                        sinkContext.addSendResultMetric(profile, clusterName, profile.getUid(), false, 0);
                        sinkContext.getMqZoneSink().releaseAcquiredSizePermit(profile);
                        profile.fail(DataProxyErrCode.GROUPID_OR_STREAMID_NOT_CONFIGURE, "");
                        return false;
                    }
                    sinkContext.fileMetricIncWithDetailStats(
                            StatConstants.EVENT_SINK_DEFAULT_TOPIC_USED, profile.getUid());
                } else {
                    sinkContext.fileMetricIncWithDetailStats(
                            StatConstants.EVENT_SINK_CONFIG_TOPIC_MISSING, profile.getUid());
                    sinkContext.addSendResultMetric(profile, clusterName, profile.getUid(), false, 0);
                    sinkContext.getMqZoneSink().releaseAcquiredSizePermit(profile);
                    profile.fail(DataProxyErrCode.GROUPID_OR_STREAMID_NOT_CONFIGURE, "");
                    return false;
                }
            } else {
                producerTopic = idConfig.getPulsarTopicName(tenant, namespace);
            }
            // get producer
            Producer<byte[]> producer = this.producerMap.get(producerTopic);
            if (producer == null) {
                try {
                    logger.info("try to new a object for topic " + producerTopic);
                    SecureRandom secureRandom = new SecureRandom(
                            (producerTopic + System.currentTimeMillis()).getBytes());
                    String producerName = producerTopic + "-" + secureRandom.nextLong();
                    producer = baseBuilder.clone().topic(producerTopic)
                            .producerName(producerName)
                            .create();
                    logger.info("create new producer success:{}", producer.getProducerName());
                    Producer<byte[]> oldProducer = this.producerMap.putIfAbsent(producerTopic, producer);
                    if (oldProducer != null) {
                        producer.close();
                        logger.info("close producer success:{}", producer.getProducerName());
                        producer = oldProducer;
                    }
                } catch (Throwable ex) {
                    logger.error("create new producer failed", ex);
                }
                // create producer failed
                if (producer == null) {
                    sinkContext.fileMetricIncWithDetailStats(
                            StatConstants.EVENT_SINK_PRODUCER_CREATE_FAILURE, producerTopic);
                    sinkContext.processSendFail(profile, clusterName, producerTopic, 0,
                            DataProxyErrCode.PRODUCER_IS_NULL, "");
                    return false;
                }
                sinkContext.fileMetricIncWithDetailStats(
                        StatConstants.EVENT_SINK_PRODUCER_CREATE_SUCCESS, producerTopic);
            }
            // send
            if (profile instanceof SimplePackProfile) {
                this.sendSimplePackProfile((SimplePackProfile) profile, idConfig, producer, producerTopic);
            } else {
                this.sendBatchPackProfile((BatchPackProfile) profile, idConfig, producer, producerTopic);
            }
            return true;
        } catch (Exception ex) {
            sinkContext.fileMetricIncWithDetailStats(StatConstants.EVENT_SINK_SEND_EXCEPTION, producerTopic);
            sinkContext.processSendFail(profile, clusterName, profile.getUid(), 0,
                    DataProxyErrCode.SEND_REQUEST_TO_MQ_FAILURE, ex.getMessage());
            if (logCounter.shouldPrint()) {
                logger.error("Send Message to Pulsar failure", ex);
            }
            return false;
        }
    }
send
@param profile   the profile to send
@return   whether sent the profile
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/pulsar/PulsarHandler.java