fix mem leak

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 {

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;
 }

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);
   }