From ee783080f341580cff84005cd90c60b28f474cc5 Mon Sep 17 00:00:00 2001
From: Liu Jicong <liujicong@qq.com>
Date: Fri, 28 Jan 2022 19:22:48 +0800
Subject: [PATCH] fix mem leak

---
 include/common/common.h        | 15 ++++++++++++++-
 source/client/src/tmq.c        |  4 ++++
 source/dnode/vnode/src/tq/tq.c |  2 +-
 3 files changed, 19 insertions(+), 2 deletions(-)

diff --git a/include/common/common.h b/include/common/common.h
index 092a666e73..16a691c338 100644
--- a/include/common/common.h
+++ b/include/common/common.h
@@ -174,7 +174,7 @@ static FORCE_INLINE void* tDecodeSMqConsumeRsp(void* buf, SMqConsumeRsp* pRsp) {
   return buf;
 }
 
-static FORCE_INLINE void destroySSDataBlock(SSDataBlock* pBlock) {
+static FORCE_INLINE void tDeleteSSDataBlock(SSDataBlock* pBlock) {
   if (pBlock == NULL) {
     return;
   }
@@ -192,6 +192,19 @@ static FORCE_INLINE void destroySSDataBlock(SSDataBlock* pBlock) {
 }
 
 
+static FORCE_INLINE void tDeleteSMqConsumeRsp(SMqConsumeRsp* pRsp) {
+  if (pRsp->schemas) {
+    if (pRsp->schemas->nCols) {
+      tfree(pRsp->schemas->pSchema);
+    }
+    free(pRsp->schemas);
+  }
+  for (int i = 0; i < taosArrayGetSize(pRsp->pBlockData); i++) {
+    SSDataBlock* pDataBlock = (SSDataBlock*)taosArrayGet(pRsp->pBlockData, i);
+    tDeleteSSDataBlock(pDataBlock);
+  }
+}
+
 //======================================================================================================================
 // the following structure shared by parser and executor
 typedef struct SColumn {
diff --git a/source/client/src/tmq.c b/source/client/src/tmq.c
index ccdd1e64cb..26177f466e 100644
--- a/source/client/src/tmq.c
+++ b/source/client/src/tmq.c
@@ -456,6 +456,7 @@ static char *formatTimestamp(char *buf, int64_t val, int precision) {
 int32_t tmq_poll_cb_inner(void* param, const SDataBuf* pMsg, int32_t code) {
   if (code == -1) {
     printf("msg discard\n");
+    free(param);
     return 0;
   }
   char pBuf[128];
@@ -465,6 +466,7 @@ int32_t tmq_poll_cb_inner(void* param, const SDataBuf* pMsg, int32_t code) {
   tDecodeSMqConsumeRsp(pMsg->pData, &rsp);
   if (rsp.numOfTopics == 0) {
     /*printf("no data\n");*/
+    free(param);
     return 0;
   }
   int32_t colNum = rsp.schemas->nCols;
@@ -501,6 +503,8 @@ int32_t tmq_poll_cb_inner(void* param, const SDataBuf* pMsg, int32_t code) {
       printf("\n");
     }
   }
+  tDeleteSMqConsumeRsp(&rsp);
+  free(param);
   /*printf("\n-----msg end------\n");*/
   return 0;
 }
diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c
index 49d15e6148..7299710586 100644
--- a/source/dnode/vnode/src/tq/tq.c
+++ b/source/dnode/vnode/src/tq/tq.c
@@ -788,7 +788,7 @@ int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg) {
   if (rsp.pBlockData) {
     for (int i = 0; i < taosArrayGetSize(rsp.pBlockData); i++) {
       SSDataBlock* pBlock = taosArrayGet(rsp.pBlockData, i);
-      destroySSDataBlock(pBlock);
+      tDeleteSSDataBlock(pBlock);
     }
     free(rsp.pBlockData);
   }
-- 
GitLab