@Override
    public void start() {
        LOGGER.info("tubemq sink starting...");

        // create connection
        try {
            createConnection();
        } catch (FlumeException e) {
            // close connection
            destroyConnection();
            LOGGER.error("Unable to create tubemq client" + ". Exception follows.", e);
        }
        started = true;
        // submit worker threads
        for (int i = 0; i < threadNum; i++) {
            threadFutures.add(sinkThreadPool.submit(new SinkTask()));
        }
        super.start();
    }
Destroy all producers and clear up caches.
inlong-tubemq/tubemq-connectors/tubemq-connector-flume/src/main/java/org/apache/flume/sink/tubemq/TubemqSink.java