private void sendEvent(MessageProducer producer, EventStat es) throws Exception {
            // send message with callback
            Message message = new Message(es.getTopic(), es.getEvent().getBody());
            producer.sendMessage(message, new MessageSentCallback() {

                @Override
                public void onMessageSent(MessageSentResult result) {
                    if (!result.isSuccess()) {
                        resendEvent(es);
                    }
                }

                @Override
                public void onException(Throwable e) {
                    LOGGER.error("exception caught", e);
                    if (e instanceof OverflowException) {
                        overflow = true;
                    }
                    resendEvent(es);
                }
            });
        }
fetch message, wait if queue is empty

@return EventStat
@throws Exception
inlong-tubemq/tubemq-connectors/tubemq-connector-flume/src/main/java/org/apache/flume/sink/tubemq/TubemqSink.java