From dc8a90c864d9d3c65f1da37fe28550d86d58bf63 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Thu, 2 Nov 2023 08:11:51 +0800
Subject: [PATCH] fix(stream): fix memory leak.

---
 source/libs/stream/src/streamDispatch.c |  6 +++---
 source/libs/stream/src/streamMeta.c     | 26 ++++++++++++-------------
 2 files changed, 15 insertions(+), 17 deletions(-)

diff --git a/source/libs/stream/src/streamDispatch.c b/source/libs/stream/src/streamDispatch.c
index d9f73c1ba0..edfc66762d 100644
--- a/source/libs/stream/src/streamDispatch.c
+++ b/source/libs/stream/src/streamDispatch.c
@@ -41,9 +41,9 @@ static int32_t tInitStreamDispatchReq(SStreamDispatchReq* pReq, const SStreamTas
                                       int32_t numOfBlocks, int64_t dstTaskId, int32_t type);
 
 void initRpcMsg(SRpcMsg* pMsg, int32_t msgType, void* pCont, int32_t contLen) {
-    pMsg->msgType = msgType;
-    pMsg->pCont = pCont;
-    pMsg->contLen = contLen;
+  pMsg->msgType = msgType;
+  pMsg->pCont = pCont;
+  pMsg->contLen = contLen;
 }
 
 int32_t tEncodeStreamDispatchReq(SEncoder* pEncoder, const SStreamDispatchReq* pReq) {
diff --git a/source/libs/stream/src/streamMeta.c b/source/libs/stream/src/streamMeta.c
index e6d5f023d6..887e879934 100644
--- a/source/libs/stream/src/streamMeta.c
+++ b/source/libs/stream/src/streamMeta.c
@@ -844,6 +844,12 @@ static bool waitForEnoughDuration(SMetaHbInfo* pInfo) {
   return false;
 }
 
+static void clearHbMsg(SStreamHbMsg* pMsg, SArray* pIdList) {
+  taosArrayDestroy(pMsg->pTaskStatus);
+  taosArrayDestroy(pMsg->pUpdateNodes);
+  taosArrayDestroy(pIdList);
+}
+
 void metaHbToMnode(void* param, void* tmrId) {
   int64_t rid = *(int64_t*)param;
 
@@ -977,17 +983,13 @@ void metaHbToMnode(void* param, void* tmrId) {
     tEncodeSize(tEncodeStreamHbMsg, &hbMsg, tlen, code);
     if (code < 0) {
       stError("vgId:%d encode stream hb msg failed, code:%s", pMeta->vgId, tstrerror(code));
-      taosArrayDestroy(hbMsg.pTaskStatus);
-      taosReleaseRef(streamMetaId, rid);
-      return;
+      goto _end;
     }
 
     void* buf = rpcMallocCont(tlen);
     if (buf == NULL) {
       stError("vgId:%d encode stream hb msg failed, code:%s", pMeta->vgId, tstrerror(TSDB_CODE_OUT_OF_MEMORY));
-      taosArrayDestroy(hbMsg.pTaskStatus);
-      taosReleaseRef(streamMetaId, rid);
-      return;
+      goto _end;
     }
 
     SEncoder encoder;
@@ -995,15 +997,12 @@ void metaHbToMnode(void* param, void* tmrId) {
     if ((code = tEncodeStreamHbMsg(&encoder, &hbMsg)) < 0) {
       rpcFreeCont(buf);
       stError("vgId:%d encode stream hb msg failed, code:%s", pMeta->vgId, tstrerror(code));
-      taosArrayDestroy(hbMsg.pTaskStatus);
-      taosReleaseRef(streamMetaId, rid);
-      return;
+      goto _end;
     }
     tEncoderClear(&encoder);
 
-    SRpcMsg msg = {0};
+    SRpcMsg msg = {.info.noResp = 1,};
     initRpcMsg(&msg, TDMT_MND_STREAM_HEARTBEAT, buf, tlen);
-    msg.info.noResp = 1;
 
     pMeta->pHbInfo->hbCount += 1;
 
@@ -1014,9 +1013,8 @@ void metaHbToMnode(void* param, void* tmrId) {
     stDebug("vgId:%d no tasks and no mnd epset, not send stream hb to mnode", pMeta->vgId);
   }
 
-  taosArrayDestroy(hbMsg.pTaskStatus);
-  taosArrayDestroy(hbMsg.pUpdateNodes);
-
+  _end:
+  clearHbMsg(&hbMsg, pIdList);
   taosTmrReset(metaHbToMnode, META_HB_CHECK_INTERVAL, param, streamEnv.timer, &pMeta->pHbInfo->hbTmr);
   taosReleaseRef(streamMetaId, rid);
 }