private void createConnection() throws FlumeException {
        // if already connected, just skip
        if (sessionFactory != null) {
            return;
        }

        try {
            TubeClientConfig conf = initTubeConfig();
            sessionFactory = new TubeMultiSessionFactory(conf);
            logger.info("create tube connection successfully");
        } catch (TubeClientException e) {
            logger.error("create connnection error in tubesink, "
                    + "maybe tube master set error, please re-check. ex1 {}", e.getMessage());
            throw new FlumeException("connect to Tube error1, "
                    + "maybe zkstr/zkroot set error, please re-check");
        } catch (Throwable e) {
            logger.error("create connnection error in tubesink, "
                    + "maybe tube master set error/shutdown in progress, please re-check. ex2 {}", e);
            throw new FlumeException("connect to meta error2, "
                    + "maybe tube master set error/shutdown in progress, please re-check");
        }

        if (producerMap == null) {
            producerMap = new HashMap<String, MessageProducer>();
        }
    }
If this function is called successively without calling {@see #destroyConnection()}, only the
first call has any effect.

@throws FlumeException if an RPC client connection could not be opened
inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/sink/TubeSink.java