public void notifyInflightMsgClosed() {
        long curTime;
        Timeout timeoutTask;
        TcpNettyClient nettyTcpClient;
        for (Integer messageId : this.reqTimeouts.keySet()) {
            if (messageId == null) {
                continue;
            }
            timeoutTask = this.reqTimeouts.remove(messageId);
            if (timeoutTask != null) {
                timeoutTask.cancel();
            }
            TcpCallFuture callFuture = this.reqObjects.remove(messageId);
            if (callFuture == null) {
                continue;
            }
            curTime = System.currentTimeMillis();
            // find and process in using clients
            nettyTcpClient = usingClientMaps.get(callFuture.getClientAddr());
            if (nettyTcpClient != null
                    && nettyTcpClient.getChanTermId() == callFuture.getChanTerm()) {
                try {
                    nettyTcpClient.getChannel().eventLoop().execute(
                            () -> callFuture.onMessageAck(new ProcessResult(ErrorCode.SDK_CLOSED)));
                } catch (Throwable ex) {
                    if (callbackExceptCnt.shouldPrint()) {
                        logger.info("ClientMgr({}) closed, callback exception!",
                                senderId, ex);
                    }
                } finally {
                    nettyTcpClient.decInFlightMsgCnt(callFuture.getChanTerm());
                    this.releaseAsyncCachedPermits(callFuture);
                    baseSender.getMetricHolder().addCallbackFailMetric(ErrorCode.SDK_CLOSED.getErrCode(),
                            callFuture.getGroupId(), callFuture.getStreamId(), callFuture.getMsgCnt(),
                            (System.currentTimeMillis() - curTime));
                }
                return;
            }
            // find and process in deleting clients
            nettyTcpClient = deletingClientMaps.get(callFuture.getClientAddr());
            if (nettyTcpClient != null
                    && nettyTcpClient.getChanTermId() == callFuture.getChanTerm()) {
                try {
                    nettyTcpClient.getChannel().eventLoop().execute(
                            () -> callFuture.onMessageAck(new ProcessResult(ErrorCode.SDK_CLOSED)));
                } catch (Throwable ex) {
                    if (callbackExceptCnt.shouldPrint()) {
                        logger.info("ClientMgr({}) closed, callback2 exception!",
                                senderId, ex);
                    }
                } finally {
                    nettyTcpClient.decInFlightMsgCnt(callFuture.getChanTerm());
                    this.releaseAsyncCachedPermits(callFuture);
                    baseSender.getMetricHolder().addCallbackFailMetric(ErrorCode.SDK_CLOSED.getErrCode(),
                            callFuture.getGroupId(), callFuture.getStreamId(), callFuture.getMsgCnt(),
                            (System.currentTimeMillis() - curTime));
                }
            }
        }
    }
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