@Override
        public void run() {
            int topicAndCondCnt = topicSendRounds.size();
            long sentCount = 0;
            int roundIndex = 0;
            while (msgCount < 0 || sentCount < msgCount) {
                roundIndex = (int) (sentCount++ % topicAndCondCnt);
                try {
                    Tuple2<String, String> target = topicSendRounds.get(roundIndex);
                    Message message = MixedUtils.buildMessage(
                            target.getF0(), target.getF1(), sentData, sentCount);
                    // use sync or async process
                    if (syncProduction) {
                        MessageSentResult procResult =
                                producer.sendMessage(message);
                        TOTAL_COUNTER.incrementAndGet();
                        if (procResult.isSuccess()) {
                            SENT_SUCC_COUNTER.incrementAndGet();
                        } else {
                            SENT_FAIL_COUNTER.incrementAndGet();
                        }
                    } else {
                        producer.sendMessage(message, new DefaultSendCallback());
                    }
                } catch (Throwable e1) {
                    TOTAL_COUNTER.incrementAndGet();
                    SENT_EXCEPT_COUNTER.incrementAndGet();
                    logger.error("sendMessage exception: ", e1);
                }
                // Limit sending flow control to avoid frequent errors
                // caused by too many inflight messages being sent
                if (!withoutDelay) {
                    MixedUtils.coolSending(sentCount);
                }
            }
            // finished, close client
            try {
                producer.shutdown();
            } catch (Throwable e) {
                logger.error("producer shutdown error: ", e);
            }
        }
Initializes the TubeMQ producer client(s) with the specified requirements.
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/cli/CliProducer.java