@Override
    public void stop() {
        LOGGER.info("tubemq sink stopping");
        started = false;
        if (sinkThreadPool != null) {
            sinkThreadPool.shutdown();
        }
        for (Future<?> future : threadFutures) {
            // interrupt threads
            future.cancel(true);
        }
        threadFutures.clear();
        destroyConnection();
        super.stop();
    }
Destroy all producers and clear up caches.
inlong-tubemq/tubemq-connectors/tubemq-connector-flume/src/main/java/org/apache/flume/sink/tubemq/TubemqSink.java