private void send(Event rowEvent, Transaction tx) throws ProducerException, InterruptedException {
        ProfileEvent event = (ProfileEvent) rowEvent;
        ClsIdConfig idConfig = context.getIdConfig(event.getUid());
        if (idConfig == null) {
            event.ack();
            LOG.error("There is no cls id config for uid {}, discard it", event.getUid());
            context.addSendResultMetric(event, context.getTaskName(), false, System.currentTimeMillis());
            return;
        }
        event.getHeaders().put(ClsSinkContext.KEY_TOPIC_ID, idConfig.getTopicId());
        context.addSendMetric(event, idConfig.getTopicId());
        AsyncProducerClient client = context.getClient(idConfig.getSecretId());
        // transform
        TransformProcessor<String, Map<String, Object>> processor =
                context.getTransformProcessor(event.getUid());
        List<LogItem> record = null;
        if (processor == null) {
            record = handler.parse(context, event);
        } else {
            record = handler.parse(context, event, processor);
        }
        ClsCallback callback = new ClsCallback(tx, context, event);
        client.putLogs(idConfig.getTopicId(), record, callback);
    }
Send event to Cls

@param rowEvent Row event.
@param tx Transaction
@throws ProducerException
@throws InterruptedException
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/cls/ClsChannelWorker.java