@Override
    public boolean send(PackProfile profile) {
        String topic = null;
        try {
            // get idConfig
            IdTopicConfig idConfig = ConfigManager.getInstance().getSinkIdTopicConfig(
                    profile.getInlongGroupId(), profile.getInlongStreamId());
            if (idConfig == null) {
                // add default topics first
                if (CommonConfigHolder.getInstance().isEnableUnConfigTopicAccept()) {
                    topic = CommonConfigHolder.getInstance().getRandDefTopics();
                    if (StringUtils.isEmpty(topic)) {
                        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 {
                topic = idConfig.getTopicName();
            }
            // create producer failed
            if (producer == null) {
                sinkContext.fileMetricIncWithDetailStats(StatConstants.EVENT_SINK_PRODUCER_NULL, topic);
                sinkContext.processSendFail(profile, clusterName, topic, 0, DataProxyErrCode.PRODUCER_IS_NULL, "");
                return false;
            }
            // send
            if (profile instanceof SimplePackProfile) {
                this.sendSimplePackProfile((SimplePackProfile) profile, idConfig, topic);
            } else {
                this.sendBatchPackProfile((BatchPackProfile) profile, idConfig, topic);
            }
            return true;
        } catch (Exception ex) {
            sinkContext.fileMetricIncWithDetailStats(StatConstants.EVENT_SINK_SEND_EXCEPTION, topic);
            sinkContext.processSendFail(profile, clusterName, profile.getUid(), 0,
                    DataProxyErrCode.SEND_REQUEST_TO_MQ_FAILURE, ex.getMessage());
            if (logCounter.shouldPrint()) {
                logger.error("Send Message to Kafka failure", ex);
            }
            return false;
        }
    }
send  message to mq
@param profile the profile need to send
@return whether sent the profile
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/kafka/KafkaHandler.java