@Override
        public void run() {
            logger.info("Sink task {} started.", Thread.currentThread().getName());
            while (canSend) {
                boolean decrementFlag = false;
                Event event = null;
                EventStat es = null;
                String topic = null;
                try {
                    if (TubeSink.this.overflow) {
                        TubeSink.this.overflow = false;
                        Thread.sleep(10);
                    }
                    if (!resendQueue.isEmpty()) {
                        // Send the data in the retry queue first
                        es = resendQueue.poll();
                        if (es != null) {
                            event = es.getEvent();
                            if (event.getHeaders().containsKey(TOPIC)) {
                                topic = event.getHeaders().get(TOPIC);
                            }
                        }
                    } else {
                        event = eventQueue.take();
                        es = new EventStat(event);
                        if (event.getHeaders().containsKey(TOPIC)) {
                            topic = event.getHeaders().get(TOPIC);
                        }
                    }

                    if (event == null) {
                        // ignore event is null, when multiple-thread SinkTask running
                        // this null value comes from resendQueue
                        continue;
                    }

                    if (topic == null || topic.equals("")) {
                        logger.warn("no topic specified in event header, just skip this event");
                        continue;
                    }

                    Long expireTime = illegalTopicMap.get(topic);
                    if (expireTime != null) {
                        long currentTime = System.currentTimeMillis();
                        if (expireTime > currentTime) {
                            continue;
                        } else {

                            illegalTopicMap.remove(topic);
                        }
                    }

                    final EventStat eventStat = es;
                    boolean sendResult = sendMessage(event, topic, eventStat);
                    if (!sendResult) {
                        continue;
                    }

                    decrementFlag = true;

                } catch (InterruptedException e) {
                    logger.info("Thread {} has been interrupted!", Thread.currentThread().getName());
                    return;
                } catch (Throwable throwable) {
                    if (throwable instanceof TubeClientException) {
                        String message = throwable.getMessage();
                        if (message != null && (message.contains("No available queue for topic")
                                || message.contains("The brokers of topic are all forbidden"))) {
                            illegalTopicMap.put(topic, System.currentTimeMillis() + 60 * 1000);
                            logger.info("IllegalTopicMap.put " + topic);
                            continue;
                        } else {
                            try {
                                Thread.sleep(100);
                            } catch (InterruptedException e) {
                                // ignore..
                            }
                        }
                    }
                    resendEvent(es, decrementFlag);
                }
            }
        }
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