@Override
        public void run(Timeout timeout) throws Exception {
            Timeout timeoutMsg = reqTimeouts.remove(messageId);
            if (timeoutMsg != null) {
                timeoutMsg.cancel();
            }
            TcpCallFuture future = reqObjects.remove(messageId);
            if (future == null) {
                return;
            }
            long curTime = System.currentTimeMillis();
            ConcurrentHashMap<Integer, Boolean> messageIdMap;
            // find and process in using clients
            TcpNettyClient nettyTcpClient = usingClientMaps.get(future.getClientAddr());
            if (nettyTcpClient != null
                    && nettyTcpClient.getChanTermId() == future.getChanTerm()) {
                messageIdMap = channelMsgIdMap.get(future.getChanStr());
                if (messageIdMap != null) {
                    messageIdMap.remove(messageId);
                }
                try {
                    nettyTcpClient.getChannel().eventLoop().execute(
                            () -> future.onMessageAck(new ProcessResult(ErrorCode.SEND_WAIT_TIMEOUT)));
                } catch (Throwable ex) {
                    if (callbackExceptCnt.shouldPrint()) {
                        logger.info("ClientMgr({}) msg timeout, callback exception!",
                                senderId, ex);
                    }
                } finally {
                    nettyTcpClient.decInFlightMsgCnt(future.getChanTerm());
                    releaseAsyncCachedPermits(future);
                    baseSender.getMetricHolder().addCallbackFailMetric(
                            ErrorCode.SEND_WAIT_TIMEOUT.getErrCode(),
                            future.getGroupId(), future.getStreamId(),
                            future.getMsgCnt(), (System.currentTimeMillis() - curTime));
                }
                return;
            }
            // find and process in deleting clients
            nettyTcpClient = deletingClientMaps.get(future.getClientAddr());
            if (nettyTcpClient != null
                    && nettyTcpClient.getChanTermId() == future.getChanTerm()) {
                messageIdMap = channelMsgIdMap.get(future.getChanStr());
                if (messageIdMap != null) {
                    messageIdMap.remove(messageId);
                }
                try {
                    nettyTcpClient.getChannel().eventLoop().execute(
                            () -> future.onMessageAck(new ProcessResult(ErrorCode.SEND_WAIT_TIMEOUT)));
                } catch (Throwable ex) {
                    if (callbackExceptCnt.shouldPrint()) {
                        logger.info("ClientMgr({}) msg timeout, callback2 exception!",
                                senderId, ex);
                    }
                } finally {
                    nettyTcpClient.decInFlightMsgCnt(future.getChanTerm());
                    releaseAsyncCachedPermits(future);
                    baseSender.getMetricHolder().addCallbackFailMetric(ErrorCode.SEND_WAIT_TIMEOUT.getErrCode(),
                            future.getGroupId(), future.getStreamId(), future.getMsgCnt(),
                            (System.currentTimeMillis() - curTime));
                }
            }
        }
Time out task call back handle
inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/network/tcp/TcpClientMgr.java