public boolean send(ProfileEvent profileEvent, Transaction tx) throws IOException {
        // send
        Map<String, String> headers = profileEvent.getHeaders();
        String topic = headers.get(Constants.TOPIC);
        // get producer
        Producer<byte[]> producer = this.producerMap.get(topic);
        if (producer == null) {
            try {
                LOG.debug("try to new a producer for topic " + topic);
                producer = baseBuilder.clone().topic(topic)
                        .producerName(workerName + "-" + cacheClusterName + "-" + topic + System.nanoTime())
                        .create();
                LOG.debug("create a new producer success:{}", producer.getProducerName());
                Producer<byte[]> oldProducer = this.producerMap.putIfAbsent(topic, producer);
                if (oldProducer != null) {
                    producer.close();
                    LOG.debug("close producer success:{}", producer.getProducerName());
                    producer = oldProducer;
                }
            } catch (Throwable ex) {
                LOG.error("create new producer failed", ex);
            }
        }
        // create producer failed
        if (producer == null) {
            tx.rollback();
            tx.close();
            sinkContext.addSendResultMetric(profileEvent, topic, false, System.currentTimeMillis());
            LOG.error("failed to create producer, send failed");
            throw new IllegalStateException();
        }

        // sendAsync
        byte[] sendBytes = this.handler.parse(sinkContext, profileEvent);
        // check
        if (sendBytes == null) {
            tx.commit();
            profileEvent.ack();
            tx.close();
            return true;
        }
        long sendTime = System.currentTimeMillis();
        CompletableFuture<MessageId> future = producer.newMessage()
                .properties(headers)
                .value(sendBytes)
                .sendAsync();
        // callback
        future.whenCompleteAsync((msgId, ex) -> {
            if (ex != null) {
                LOG.error("Send fail:{}", ex.getMessage());
                LOG.error(ex.getMessage(), ex);
                tx.rollback();
                tx.close();
                sinkContext.addSendResultMetric(profileEvent, topic, false, sendTime);
            } else {
                tx.commit();
                tx.close();
                sinkContext.addSendResultMetric(profileEvent, topic, true, sendTime);
                profileEvent.ack();
            }
        });
        return true;
    }
send

@param  profileEvent
@param  tx
@return              boolean
@throws IOException
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarProducerCluster.java