public InLongHttpMsgSender genHttpSenderByClusterId(
            HttpMsgSenderConfig configure) throws ProxySdkException {
        validProxyConfigNotNull(configure);
        // get groupId's clusterIdKey
        ProcessResult procResult = new ProcessResult();
        ProxyConfigEntry proxyConfigEntry = qryProxyMetaConfigure(configure, procResult);;
        String clusterIdKey = ProxyUtils.buildClusterIdKey(
                configure.getDataRptProtocol(), configure.getRegionName(), proxyConfigEntry.getClusterId());
        // get local built sender
        InLongHttpMsgSender messageSender = (InLongHttpMsgSender) clusterIdSenderMap.get(clusterIdKey);
        if (messageSender != null) {
            return messageSender;
        }
        // generate sender
        senderCacheLock.writeLock().lock();
        try {
            // re-get the created sender based on the clusterId Key after locked
            messageSender = (InLongHttpMsgSender) clusterIdSenderMap.get(clusterIdKey);
            if (messageSender != null) {
                return messageSender;
            }
            // build a new sender based on clusterId Key
            messageSender = new InLongHttpMsgSender(configure, msgSenderFactory, clusterIdKey);
            if (!messageSender.start(procResult)) {
                messageSender.close();
                throw new ProxySdkException("Failed to start cluster sender: " + procResult);
            }
            clusterIdSenderMap.put(clusterIdKey, messageSender);
            logger.info("MsgSenderFactory({}) generated a new clusterId({}) sender({})",
                    this.factoryNo, clusterIdKey, messageSender.getSenderId());
            return messageSender;
        } catch (Throwable ex) {
            if (exptCounter.shouldPrint()) {
                logger.warn("MsgSenderFactory({}) build cluster sender({}) exception",
                        this.factoryNo, clusterIdKey, ex);
            }
            throw new ProxySdkException("Failed to build cluster sender: " + ex.getMessage());
        } finally {
            senderCacheLock.writeLock().unlock();
        }
    }
Base Message Sender Factory

Used to manage the instance relationship of the sender factory.
Since both singleton and multiple instances involve the same relationship,
they are abstracted and maintained separately here.
inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/BaseMsgSenderFactory.java