private boolean sendMessage(Event event, String topic, EventStat es)
                throws TubeClientException, InterruptedException {
            MessageProducer producer = getProducer(topic);
            if (producer == null) {
                illegalTopicMap.put(topic, System.currentTimeMillis() + 30 * 1000);
                logger.error("Get producer is null, topic:{}", topic);
                return false;
            }

            Message message = new Message(topic, event.getBody());
            message.setAttrKeyVal("auditIp", localIp);
            String streamId = "";
            String groupId = "";
            if (event.getHeaders().containsKey(org.apache.inlong.audit.consts.AttributeConstants.INLONG_STREAM_ID)) {
                streamId = event.getHeaders().get(org.apache.inlong.audit.consts.AttributeConstants.INLONG_STREAM_ID);
                message.setAttrKeyVal(org.apache.inlong.audit.consts.AttributeConstants.INLONG_STREAM_ID, streamId);
            }
            if (event.getHeaders().containsKey(org.apache.inlong.audit.consts.AttributeConstants.INLONG_GROUP_ID)) {
                groupId = event.getHeaders().get(org.apache.inlong.audit.consts.AttributeConstants.INLONG_GROUP_ID);
                message.setAttrKeyVal(org.apache.inlong.audit.consts.AttributeConstants.INLONG_GROUP_ID, groupId);
            }

            logger.debug("producer start to send msg...");
            producer.sendMessage(message, new MyCallback(es));

            illegalTopicMap.remove(topic);
            return true;
        }
Currently, all topics are published by the same producer. If needed, extend it to multi producers.

@param topic
@throws TubeClientException
inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/sink/TubeSink.java