private void destroyConnection() {
        for (String topic : producerMap.keySet()) {
            MessageProducer producer = producerMap.get(topic);
            try {
                producer.shutdown();
            } catch (Throwable e) {
                LOGGER.error("destroy producer error in tubemqSink, ex", e);
            }
        }
        producerMap.clear();

        if (sessionFactory != null) {
            try {
                sessionFactory.shutdown();
            } catch (Exception e) {
                LOGGER.error("destroy sessionFactory error in tubemqSink, MetaClientException", e);
            }
        }
        sessionFactory = null;
        LOGGER.debug("closed meta producer");
    }
Destroy all producers and clear up caches.
inlong-tubemq/tubemq-connectors/tubemq-connector-flume/src/main/java/org/apache/flume/sink/tubemq/TubemqSink.java