private void createConnection(CreatePulsarClientCallBack callBack) throws FlumeException {
        if (pulsarClient != null) {
            return;
        }
        try {
            pulsarClient = initPulsarClient(pulsarServerUrl);
            callBack.handleCreateClientSuccess(pulsarServerUrl);
        } catch (PulsarClientException e) {
            callBack.handleCreateClientException(pulsarServerUrl);
            logger.error("create connnection error in metasink, "
                    + "maybe pulsar master set error, please re-check.url{}, ex1 {}",
                    pulsarServerUrl,
                    e.getMessage());
        } catch (Throwable e) {
            callBack.handleCreateClientException(pulsarServerUrl);
            logger.error("create connnection error in metasink, "
                    + "maybe pulsar master set error/shutdown in progress, please "
                    + "re-check. url{}, ex2 {}",
                    pulsarServerUrl,
                    e.getMessage());
        }
    }
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/pulsar/PulsarClientService.java