public boolean send(ProfileEvent profileEvent, Transaction tx) throws IOException {
        String topic = sinkContext.getTopic(profileEvent.getUid());
        ProducerRecord<String, byte[]> record = handler.parse(sinkContext, profileEvent);
        long sendTime = System.currentTimeMillis();
        // check
        if (record == null || StringUtils.isEmpty(topic)) {
            tx.commit();
            profileEvent.ack();
            tx.close();
            sinkContext.addSendResultMetric(profileEvent, topic, false, sendTime);
            return true;
        }
        try {
            producer.send(record,
                    (metadata, ex) -> {
                        if (ex == null) {
                            tx.commit();
                            sinkContext.addSendResultMetric(profileEvent, topic, true, sendTime);
                            profileEvent.ack();
                        } else {

                            if (ex instanceof RecordTooLargeException) {
                                // for the message bigger than configuredMaxPayloadSize, just discard it;
                                // otherwise, retry and wait for the server side changes the limitation
                                if (record.value().length > configuredMaxPayloadSize) {
                                    tx.commit();
                                    profileEvent.ack();
                                } else {
                                    tx.rollback();
                                }
                            } else if (ex instanceof UnknownTopicOrPartitionException
                                    || !(ex instanceof RetriableException)) {
                                // for non-retriable exception, just discard it
                                tx.commit();
                                profileEvent.ack();
                            } else {
                                tx.rollback();
                            }
                            LOG.error(String.format("send failed, topic is %s", topic), ex);
                            sinkContext.addSendResultMetric(profileEvent, topic, false, sendTime);
                        }
                        tx.close();
                    });
            return true;
        } catch (Exception e) {
            tx.rollback();
            tx.close();
            LOG.error(e.getMessage(), e);
            sinkContext.addSendResultMetric(profileEvent, topic, false, sendTime);
            return false;
        }
    }
Send data

@param profileEvent data to send
@return boolean
@throws IOException
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/kafka/KafkaProducerCluster.java