public boolean sendMessage(String topic, Event event,
            SendMessageCallBack sendMessageCallBack, EventStat es) {
        Producer producer = null;
        try {
            producer = getProducer(topic);
        } catch (Exception e) {
            if (logPrinterA.shouldPrint()) {
                logger.error("Get producer failed!", e);
            }
        }

        if (producer == null) {
            logger.error("Get producer is null!");
            MetricsManager.getInstance().addSendFailed(1);
            return false;
        }

        Map<String, String> proMap = new HashMap<>();
        proMap.put("auditIp", localIp);
        String streamId = "";
        String groupId = "";
        if (event.getHeaders().containsKey(AttributeConstants.INLONG_STREAM_ID)) {
            streamId = event.getHeaders().get(AttributeConstants.INLONG_STREAM_ID);
            proMap.put(AttributeConstants.INLONG_STREAM_ID, streamId);
        }
        if (event.getHeaders().containsKey(AttributeConstants.INLONG_GROUP_ID)) {
            groupId = event.getHeaders().get(AttributeConstants.INLONG_GROUP_ID);
            proMap.put(AttributeConstants.INLONG_GROUP_ID, groupId);
        }

        logger.debug("producer send msg!");
        producer.newMessage().properties(proMap).value(event.getBody())
                .sendAsync().thenAccept((msgId) -> {
                    sendMessageCallBack.handleMessageSendSuccess((MessageIdImpl) msgId, es);

                }).exceptionally((e) -> {
                    sendMessageCallBack.handleMessageSendException(es, e);
                    return null;
                });
        return true;
    }
send message

@param topic
@param event
@param sendMessageCallBack
@param es
@return
inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/sink/pulsar/PulsarClientService.java