memory leak if response is not received

src/rpc/src/rpcMain.c

@@ -601,11 +601,10 @@ static void rpcReleaseConn(SRpcConn *pConn) {
     taosHashRemove(pRpc->hash, hashstr, size);
     rpcFreeMsg(pConn->pRspMsg); // it may have a response msg saved, but not request msg
     pConn->pRspMsg = NULL;
-  
-    if (pConn->pReqMsg) rpcFreeCont(pConn->pReqMsg);
   } 
 
   // memset could not be used, since lockeBy can not be reset
+  if (pConn->pReqMsg) rpcFreeCont(pConn->pReqMsg);
   pConn->inType = 0;
   pConn->outType = 0;
   pConn->inTranId = 0;
@@ -959,6 +958,7 @@ static void rpcProcessBrokenLink(SRpcConn *pConn) {
   if (pConn->outType) {
     SRpcReqContext *pContext = pConn->pContext;
     pContext->code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
+    pConn->pReq = NULL;
     taosTmrStart(rpcProcessConnError, 0, pContext, pRpc->tmrCtrl);
   }
 
@@ -1061,6 +1061,7 @@ static void rpcProcessIncomingMsg(SRpcConn *pConn, SRpcHead *pHead) {
     SRpcReqContext *pContext = pConn->pContext;
     rpcMsg.handle = pContext;
     pConn->pContext = NULL;
+    pConn->pReq = NULL;
 
     // for UDP, port may be changed by server, the port in ipSet shall be used for cache
     if (pHead->code != TSDB_CODE_RPC_TOO_SLOW) {
@@ -1297,6 +1298,7 @@ static void rpcProcessRetryTimer(void *param, void *tmrId) {
       tDebug("%s, failed to send msg:%s to %s:%hu", pConn->info, taosMsg[pConn->outType], pConn->peerFqdn, pConn->peerPort);
       if (pConn->pContext) {
         pConn->pContext->code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
+        pConn->pReq = NULL;
         taosTmrStart(rpcProcessConnError, 0, pConn->pContext, pRpc->tmrCtrl);
         rpcReleaseConn(pConn);
       }