@Override
    public void updateProxyInfoList(boolean nodeChanged, ConcurrentHashMap<String, HostInfo> hostInfoMap) {
        if (hostInfoMap.isEmpty() || !this.started.get()) {
            return;
        }
        long curTime = System.currentTimeMillis();
        try {
            // shuffle candidate nodes
            List<HostInfo> candidateNodes = new ArrayList<>(hostInfoMap.size());
            candidateNodes.addAll(hostInfoMap.values());
            Collections.sort(candidateNodes);
            Collections.shuffle(candidateNodes);
            int curTotalCnt = candidateNodes.size();
            int needActiveCnt = Math.min(this.tcpConfig.getAliveConnections(), curTotalCnt);
            // build next step nodes
            TcpNettyClient client;
            int maxCycleCnt = 3;
            this.connFailNodeMap.clear();
            List<String> realHosts = new ArrayList<>();
            ConcurrentHashMap<String, TcpNettyClient> tmpClientMaps = new ConcurrentHashMap<>();
            do {
                int selectCnt = 0;
                for (HostInfo hostInfo : candidateNodes) {
                    if (realHosts.contains(hostInfo.getReferenceName())) {
                        continue;
                    }
                    try {
                        client = new TcpNettyClient(this.senderId,
                                this.bootstrap, hostInfo, this.tcpConfig);
                        if (!client.connect(false, channelTermGen.incrementAndGet())) {
                            this.connFailNodeMap.put(hostInfo.getReferenceName(), hostInfo);
                            client.close(false);
                            continue;
                        }
                        realHosts.add(hostInfo.getReferenceName());
                        tmpClientMaps.put(hostInfo.getReferenceName(), client);
                        if (++selectCnt >= needActiveCnt) {
                            break;
                        }
                    } catch (Throwable ex) {
                        if (updConExptCnt.shouldPrint()) {
                            logger.warn("ClientMgr({}) build client {} exception",
                                    senderId, hostInfo.getReferenceName(), ex);
                        }
                    }
                }
                if (!realHosts.isEmpty()) {
                    break;
                }
                ProxyUtils.sleepSomeTime(1000L);
            } while (--maxCycleCnt > 0);
            // update active nodes
            if (realHosts.isEmpty()) {
                if (nodeChanged) {
                    logger.error("ClientMgr({}) changed nodes, but all connect failure, nodes={}!",
                            this.senderId, candidateNodes);
                } else {
                    logger.error("ClientMgr({}) re-choose nodes, but all connect failure, nodes={}!",
                            this.senderId, candidateNodes);
                }
            } else {
                this.lastUpdateTime = System.currentTimeMillis();
                this.deletingClientMaps = this.usingClientMaps;
                this.usingClientMaps = tmpClientMaps;
                this.activeNodes = realHosts;
                if (nodeChanged) {
                    logger.info("ClientMgr({}) changed nodes, wast {}ms, nodeCnt=(r:{}-a:{}), actives={}, fail={}",
                            senderId, (System.currentTimeMillis() - curTime),
                            needActiveCnt, realHosts.size(), realHosts, connFailNodeMap.keySet());
                } else {
                    logger.info("ClientMgr({}) re-choose nodes, wast {}ms, nodeCnt=(r:{}-a:{}), actives={}, fail={}",
                            senderId, (System.currentTimeMillis() - curTime),
                            needActiveCnt, realHosts.size(), realHosts, connFailNodeMap.keySet());
                }
            }
        } catch (Throwable ex) {
            if (updConExptCnt.shouldPrint()) {
                logger.warn("ClientMgr({}) update nodes throw exception", senderId, ex);
            }
        }
    }
Build up the connection between the server and client.
inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/network/tcp/TcpClientMgr.java