@Override
    public Status process() {
        if (!started) {
            return Status.BACKOFF;
        }
        Channel channel = getChannel();
        Transaction tx = channel.getTransaction();
        tx.begin();
        Status status = Status.READY;
        try {
            Event event = channel.take();
            if (event != null) {
                if (!eventQueue.offer(event, eventOfferTimeout, TimeUnit.MILLISECONDS)) {
                    LOGGER.info("[{}] Channel --> Queue(has no enough space,current code point) --> tubemq, Check "
                            + "if tubemq server or network is ok.(if this situation last long time it will cause"
                            + " memoryChannel full and fileChannel write.)", getName());
                    counter.incrementRollbackCount();
                    tx.rollback();
                } else {
                    tx.commit();
                }
            } else {
                // if event is null, that means queue is empty, backoff it.
                status = Status.BACKOFF;
                tx.commit();
            }
        } catch (Throwable t) {
            LOGGER.error("Process event failed!" + this.getName(), t);
            try {
                counter.incrementRollbackCount();
                tx.rollback();
            } catch (Throwable e) {
                LOGGER.error("tubemq sink transaction rollback exception", e);
            }
        } finally {
            tx.close();
        }
        return status;
    }
Destroy all producers and clear up caches.
inlong-tubemq/tubemq-connectors/tubemq-connector-flume/src/main/java/org/apache/flume/sink/tubemq/TubemqSink.java