@Override
        public void run() {
            logger.info("Sink task {} started.", Thread.currentThread().getName());
            while (canSend) {
                boolean decrementFlag = false;
                Event event = null;
                EventStat eventStat = null;
                try {
                    if (PulsarSink.this.overflow) {
                        PulsarSink.this.overflow = false;
                        Thread.currentThread().sleep(10);
                    }
                    if (!resendQueue.isEmpty()) {
                        /*
                         * Send the data in the retry queue first
                         */
                        eventStat = resendQueue.poll();
                        if (eventStat != null) {
                            event = eventStat.getEvent();
                        }
                    } else {
                        if (currentInFlightCount.get() > BATCH_SIZE) {
                            /*
                             * Under the condition that the number of unresponsive messages is greater than 1w, the
                             * number of unresponsive messages sent to pulsar will be printed periodically
                             */
                            logCounter++;
                            if (logCounter == 1 || logCounter % 100000 == 0) {
                                logger.info(getName()
                                        + " currentInFlightCount={} resendQueue"
                                        + ".size={}",
                                        currentInFlightCount.get(), resendQueue.size());
                            }
                            if (logCounter > Long.MAX_VALUE - 10) {
                                logCounter = 0;
                            }
                        }
                        event = eventQueue.take();
                        eventStat = new EventStat(event);
                        sinkCounter.incrementEventDrainAttemptCount();
                    }
                    logger.debug("Event is {}, topic = {} ", event, topic);

                    if (event == null) {
                        continue;
                    }

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

                    final EventStat es = eventStat;
                    boolean sendResult = pulsarClientService.sendMessage(topic, event,
                            PulsarSink.this, es);
                    if (!sendResult) {
                        continue;
                    }
                    currentInFlightCount.incrementAndGet();
                    decrementFlag = true;
                } catch (InterruptedException e) {
                    logger.info("Thread {} has been interrupted!", Thread.currentThread().getName());
                    return;
                } catch (Throwable t) {
                    if (t instanceof PulsarClientException) {
                        String message = t.getMessage();
                        if (message != null && (message.contains("No available queue for topic")
                                || message.contains("The brokers of topic are all forbidden"))) {
                            logger.info("IllegalTopicMap.put " + topic);
                            continue;
                        } else {
                            try {
                                Thread.sleep(100);
                            } catch (InterruptedException e) {
                                // ignore..
                            }
                        }
                    }
                    resendEvent(eventStat, decrementFlag);
                }
            }
        }
Resend the data and store the data in the memory cache.
@param es
@param isDecrement
inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/sink/PulsarSink.java