@Override
        public void run() {
            logger.info("ClientMgr({}) start MaintThread!", senderId);
            long curTime;
            long checkRound = 0L;
            boolean printFailNodes;
            ProcessResult procResult = new ProcessResult();
            Set<String> failNodes = new HashSet<>();
            while (!bShutDown) {
                printFailNodes = ((Math.abs(checkRound++) % 90L) == 0L);
                try {
                    curTime = System.currentTimeMillis();
                    // clean deleting nodes
                    if (deletingClientMaps != null && !deletingClientMaps.isEmpty()) {
                        if (lastUpdateTime > 0
                                && curTime - lastUpdateTime > tcpConfig.getConCloseWaitPeriodMs()) {
                            for (TcpNettyClient client : deletingClientMaps.values()) {
                                if (client == null) {
                                    continue;
                                }
                                client.close(false);
                            }
                            deletingClientMaps.clear();
                        }
                    }
                    // check and keepalive using nodes
                    for (TcpNettyClient nettyTcpClient : usingClientMaps.values()) {
                        if (nettyTcpClient == null) {
                            continue;
                        }
                        if (nettyTcpClient.isActive()) {
                            if (nettyTcpClient.isIdleClient(curTime)) {
                                nettyTcpClient.sendHeartBeatMsg(procResult);
                            }
                        } else {
                            if ((nettyTcpClient.getClientUsingCnt() <= 0)
                                    && (nettyTcpClient.getMsgInflightCnt() <= 0)
                                    || (curTime - nettyTcpClient.getChanInvalidTime() > tcpConfig
                                            .getConCloseWaitPeriodMs())) {
                                nettyTcpClient.reconnect(false, channelTermGen.incrementAndGet());
                            }
                        }
                        if (printFailNodes) {
                            if (nettyTcpClient.isConFailNodes()) {
                                failNodes.add(nettyTcpClient.getClientAddr());
                            }
                        }
                    }
                } catch (Throwable ex) {
                    if (exptCounter.shouldPrint()) {
                        logger.warn("ClientMgr({}) MaintThread throw exception", senderId, ex);
                    }
                }
                if (printFailNodes && !failNodes.isEmpty()) {
                    logger.warn("ClientMgr({}) found continue connect fail nodes: {}",
                            senderId, failNodes);
                    failNodes.clear();
                }
                if (bShutDown) {
                    break;
                }
                ProxyUtils.sleepSomeTime(2000L);
            }
            logger.info("ClientMgr({}) exit MaintThread!", senderId);
        }
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