feat:[TD-32166] check input params for tmq

source/client/src/clientTmq.c

@@ -554,6 +554,9 @@ char** tmq_list_to_c_array(const tmq_list_t* list) {
 }
 
 static int32_t tmqCommitDone(SMqCommitCbParamSet* pParamSet) {
+  if (pParamSet == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int64_t refId = pParamSet->refId;
   int32_t code = 0;
   tmq_t* tmq = taosAcquireRef(tmqMgmt.rsetId, refId);
@@ -575,6 +578,9 @@ static int32_t tmqCommitDone(SMqCommitCbParamSet* pParamSet) {
 }
 
 static int32_t commitRspCountDown(SMqCommitCbParamSet* pParamSet, int64_t consumerId, const char* pTopic, int32_t vgId) {
+  if (pParamSet == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t waitingRspNum = atomic_sub_fetch_32(&pParamSet->waitingRspNum, 1);
   if (waitingRspNum == 0) {
     tqDebugC("consumer:0x%" PRIx64 " topic:%s vgId:%d all commit-rsp received, commit completed", consumerId, pTopic,
@@ -603,6 +609,9 @@ static int32_t tmqCommitCb(void* param, SDataBuf* pBuf, int32_t code) {
 
 static int32_t doSendCommitMsg(tmq_t* tmq, int32_t vgId, SEpSet* epSet, STqOffsetVal* offset, const char* pTopicName,
                                SMqCommitCbParamSet* pParamSet) {
+  if (tmq == NULL || epSet == NULL || offset == NULL || pTopicName == NULL || pParamSet == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   SMqVgOffset pOffset = {0};
 
   pOffset.consumerId = tmq->consumerId;
@@ -673,6 +682,9 @@ static int32_t doSendCommitMsg(tmq_t* tmq, int32_t vgId, SEpSet* epSet, STqOffse
 }
 
 static int32_t getTopicByName(tmq_t* tmq, const char* pTopicName, SMqClientTopic** topic) {
+  if (tmq == NULL || pTopicName == NULL || topic == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t numOfTopics = taosArrayGetSize(tmq->clientTopics);
   for (int32_t i = 0; i < numOfTopics; ++i) {
     SMqClientTopic* pTopic = taosArrayGet(tmq->clientTopics, i);
@@ -689,6 +701,9 @@ static int32_t getTopicByName(tmq_t* tmq, const char* pTopicName, SMqClientTopic
 
 static int32_t prepareCommitCbParamSet(tmq_t* tmq, tmq_commit_cb* pCommitFp, void* userParam, int32_t rspNum,
                                        SMqCommitCbParamSet** ppParamSet) {
+  if (tmq == NULL || ppParamSet == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   SMqCommitCbParamSet* pParamSet = taosMemoryCalloc(1, sizeof(SMqCommitCbParamSet));
   if (pParamSet == NULL) {
     return terrno;
@@ -704,6 +719,9 @@ static int32_t prepareCommitCbParamSet(tmq_t* tmq, tmq_commit_cb* pCommitFp, voi
 }
 
 static int32_t getClientVg(tmq_t* tmq, char* pTopicName, int32_t vgId, SMqClientVg** pVg) {
+  if (tmq == NULL || pTopicName == NULL || pVg == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   SMqClientTopic* pTopic = NULL;
   int32_t         code = getTopicByName(tmq, pTopicName, &pTopic);
   if (code != 0) {
@@ -724,6 +742,9 @@ static int32_t getClientVg(tmq_t* tmq, char* pTopicName, int32_t vgId, SMqClient
 }
 
 static int32_t innerCommit(tmq_t* tmq, char* pTopicName, STqOffsetVal* offsetVal, SMqClientVg* pVg, SMqCommitCbParamSet* pParamSet){
+  if (tmq == NULL || pTopicName == NULL || offsetVal == NULL || pVg == NULL || pParamSet == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = 0;
   if (offsetVal->type <= 0) {
     code = TSDB_CODE_TMQ_INVALID_MSG;
@@ -754,6 +775,9 @@ static int32_t innerCommit(tmq_t* tmq, char* pTopicName, STqOffsetVal* offsetVal
 
 static int32_t asyncCommitOffset(tmq_t* tmq, char* pTopicName, int32_t vgId, STqOffsetVal* offsetVal,
                                  tmq_commit_cb* pCommitFp, void* userParam) {
+  if (tmq == NULL || pTopicName == NULL || offsetVal == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   tqInfoC("consumer:0x%" PRIx64 " do manual commit offset for %s, vgId:%d", tmq->consumerId, pTopicName, vgId);
   SMqCommitCbParamSet* pParamSet = NULL;
   int32_t code = prepareCommitCbParamSet(tmq, pCommitFp, userParam, 0, &pParamSet);
@@ -807,6 +831,9 @@ end:
 }
 
 static int32_t innerCommitAll(tmq_t* tmq, SMqCommitCbParamSet* pParamSet){
+  if (tmq == NULL || pParamSet == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = 0;
   taosRLockLatch(&tmq->lock);
   int32_t numOfTopics = taosArrayGetSize(tmq->clientTopics);
@@ -842,6 +869,9 @@ END:
 }
 
 static void asyncCommitAllOffsets(tmq_t* tmq, tmq_commit_cb* pCommitFp, void* userParam) {
+  if (tmq == NULL) {
+    return;
+  }
   int32_t code = 0;
   SMqCommitCbParamSet* pParamSet = NULL;
   // init waitingRspNum as DEFAULT_COMMIT_CNT to prevent concurrency issue
@@ -1071,12 +1101,15 @@ END:
 }
 
 static void defaultCommitCbFn(tmq_t* pTmq, int32_t code, void* param) {
-  if (code != 0) {
+  if (code != 0 && pTmq != NULL) {
     tqErrorC("consumer:0x%" PRIx64 ", failed to commit offset, code:%s", pTmq->consumerId, tstrerror(code));
   }
 }
 
 static void tmqFreeRspWrapper(SMqRspWrapper* rspWrapper) {
+  if (rspWrapper == NULL) {
+    return;
+  }
   if (rspWrapper->tmqRspType == TMQ_MSG_TYPE__EP_RSP) {
     tDeleteSMqAskEpRsp(&rspWrapper->epRsp);
   } else if (rspWrapper->tmqRspType == TMQ_MSG_TYPE__POLL_DATA_RSP) {
@@ -1091,12 +1124,18 @@ static void tmqFreeRspWrapper(SMqRspWrapper* rspWrapper) {
 }
 
 static void freeClientVg(void* param) {
+  if (param == NULL) {
+    return;
+  }
   SMqClientVg* pVg = param;
   tOffsetDestroy(&pVg->offsetInfo.endOffset);
   tOffsetDestroy(&pVg->offsetInfo.beginOffset);
   tOffsetDestroy(&pVg->offsetInfo.committedOffset);
 }
 static void freeClientTopic(void* param) {
+  if (param == NULL) {
+    return;
+  }
   SMqClientTopic* pTopic = param;
   taosMemoryFreeClear(pTopic->schema.pSchema);
   taosArrayDestroyEx(pTopic->vgs, freeClientVg);
@@ -1104,6 +1143,9 @@ static void freeClientTopic(void* param) {
 
 static void initClientTopicFromRsp(SMqClientTopic* pTopic, SMqSubTopicEp* pTopicEp, SHashObj* pVgOffsetHashMap,
                                    tmq_t* tmq) {
+  if (pTopic == NULL || pTopicEp == NULL || pVgOffsetHashMap == NULL || tmq == NULL) {
+    return;
+  }
   pTopic->schema = pTopicEp->schema;
   pTopicEp->schema.nCols = 0;
   pTopicEp->schema.pSchema = NULL;
@@ -1167,6 +1209,9 @@ static void initClientTopicFromRsp(SMqClientTopic* pTopic, SMqSubTopicEp* pTopic
 }
 
 static void buildNewTopicList(tmq_t* tmq, SArray* newTopics, const SMqAskEpRsp* pRsp){
+        if (tmq == NULL || newTopics == NULL || pRsp == NULL) {
+        return;
+        }
   SHashObj* pVgOffsetHashMap = taosHashInit(64, MurmurHash3_32, false, HASH_NO_LOCK);
   if (pVgOffsetHashMap == NULL) {
     tqErrorC("consumer:0x%" PRIx64 " taos hash init null, code:%d", tmq->consumerId, terrno);
@@ -1221,6 +1266,9 @@ static void buildNewTopicList(tmq_t* tmq, SArray* newTopics, const SMqAskEpRsp*
 }
 
 static void doUpdateLocalEp(tmq_t* tmq, int32_t epoch, const SMqAskEpRsp* pRsp) {
+        if (tmq == NULL || pRsp == NULL) {
+        return;
+        }
   int32_t topicNumGet = taosArrayGetSize(pRsp->topics);
   // vnode transform (epoch == tmq->epoch && topicNumGet != 0)
   // ask ep rsp (epoch == tmq->epoch && topicNumGet == 0)
@@ -1337,6 +1385,9 @@ static int32_t askEpCb(void* param, SDataBuf* pMsg, int32_t code) {
 }
 
 static int32_t askEp(tmq_t* pTmq, void* param, bool sync, bool updateEpSet) {
+  if (pTmq == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   SMqAskEpReq req = {0};
   req.consumerId = pTmq->consumerId;
   req.epoch = updateEpSet ? -1 : pTmq->epoch;
@@ -1395,6 +1446,9 @@ static int32_t askEp(tmq_t* pTmq, void* param, bool sync, bool updateEpSet) {
 }
 
 void tmqHandleAllDelayedTask(tmq_t* pTmq) {
+        if (pTmq == NULL) {
+        return;
+        }
   STaosQall* qall = NULL;
   int32_t    code = 0;
 
@@ -1443,6 +1497,7 @@ void tmqHandleAllDelayedTask(tmq_t* pTmq) {
 }
 
 void tmqClearUnhandleMsg(tmq_t* tmq) {
+  if (tmq == NULL) return;
   SMqRspWrapper* rspWrapper = NULL;
   while (taosGetQitem(tmq->qall, (void**)&rspWrapper) != 0) {
     tmqFreeRspWrapper(rspWrapper);
@@ -1507,6 +1562,7 @@ int32_t tmq_subscription(tmq_t* tmq, tmq_list_t** topics) {
 }
 
 void tmqFreeImpl(void* handle) {
+        if (handle == NULL) return;
   tmq_t*  tmq = (tmq_t*)handle;
   int64_t id = tmq->consumerId;
 
@@ -1715,6 +1771,7 @@ _failed:
 }
 
 static int32_t syncAskEp(tmq_t* pTmq) {
+  if (pTmq == NULL) return TSDB_CODE_INVALID_PARA;
   SAskEpInfo* pInfo = taosMemoryMalloc(sizeof(SAskEpInfo));
   if (pInfo == NULL) return terrno;
   if (tsem2_init(&pInfo->sem, 0, 0) != 0) {
@@ -1897,6 +1954,9 @@ void tmq_conf_set_auto_commit_cb(tmq_conf_t* conf, tmq_commit_cb* cb, void* para
 }
 
 static void getVgInfo(tmq_t* tmq, char* topicName, int32_t vgId, SMqClientVg** pVg) {
+  if (tmq == NULL || topicName == NULL || pVg == NULL) {
+    return;
+  }
   int32_t topicNumCur = taosArrayGetSize(tmq->clientTopics);
   for (int i = 0; i < topicNumCur; i++) {
     SMqClientTopic* pTopicCur = taosArrayGet(tmq->clientTopics, i);
@@ -1914,6 +1974,9 @@ static void getVgInfo(tmq_t* tmq, char* topicName, int32_t vgId, SMqClientVg** p
 }
 
 static SMqClientTopic* getTopicInfo(tmq_t* tmq, char* topicName) {
+  if (tmq == NULL || topicName == NULL) {
+    return NULL;
+  }
   int32_t topicNumCur = taosArrayGetSize(tmq->clientTopics);
   for (int i = 0; i < topicNumCur; i++) {
     SMqClientTopic* pTopicCur = taosArrayGet(tmq->clientTopics, i);
@@ -2026,6 +2089,9 @@ EXIT:
 }
 
 void tmqBuildConsumeReqImpl(SMqPollReq* pReq, tmq_t* tmq, int64_t timeout, SMqClientTopic* pTopic, SMqClientVg* pVg) {
+  if (pReq == NULL || tmq == NULL || pTopic == NULL || pVg == NULL) {
+    return;
+  }
   (void)snprintf(pReq->subKey, TSDB_SUBSCRIBE_KEY_LEN, "%s%s%s", tmq->groupId, TMQ_SEPARATOR, pTopic->topicName);
   pReq->withTbName = tmq->withTbName;
   pReq->consumerId = tmq->consumerId;
@@ -2072,7 +2138,7 @@ void changeByteEndian(char* pData) {
 }
 
 static void tmqGetRawDataRowsPrecisionFromRes(void* pRetrieve, void** rawData, int64_t* rows, int32_t* precision) {
-  if (pRetrieve == NULL) {
+  if (pRetrieve == NULL || rawData == NULL || rows == NULL) {
     return;
   }
   if (*(int64_t*)pRetrieve == 0) {
@@ -2092,6 +2158,9 @@ static void tmqGetRawDataRowsPrecisionFromRes(void* pRetrieve, void** rawData, i
 
 static void tmqBuildRspFromWrapperInner(SMqPollRspWrapper* pWrapper, SMqClientVg* pVg, int64_t* numOfRows,
                                         SMqRspObj* pRspObj) {
+  if (pWrapper == NULL || pVg == NULL || numOfRows == NULL || pRspObj == NULL) {
+    return;
+  }
   pRspObj->resIter = -1;
   pRspObj->resInfo.totalRows = 0;
   pRspObj->resInfo.precision = TSDB_TIME_PRECISION_MILLI;
@@ -2130,6 +2199,9 @@ static void tmqBuildRspFromWrapperInner(SMqPollRspWrapper* pWrapper, SMqClientVg
 }
 
 static int32_t doTmqPollImpl(tmq_t* pTmq, SMqClientTopic* pTopic, SMqClientVg* pVg, int64_t timeout) {
+  if (pTmq == NULL || pTopic == NULL || pVg == NULL) {
+    return TSDB_CODE_INVALID_MSG;
+  }
   SMqPollReq      req = {0};
   char*           msg = NULL;
   SMqPollCbParam* pParam = NULL;
@@ -2199,6 +2271,9 @@ static int32_t doTmqPollImpl(tmq_t* pTmq, SMqClientTopic* pTopic, SMqClientVg* p
 
 // broadcast the poll request to all related vnodes
 static int32_t tmqPollImpl(tmq_t* tmq, int64_t timeout) {
+  if (tmq == NULL) {
+    return TSDB_CODE_INVALID_MSG;
+  }
   int32_t code = 0;
 
   taosWLockLatch(&tmq->lock);
@@ -2258,6 +2333,9 @@ end:
 
 static void updateVgInfo(SMqClientVg* pVg, STqOffsetVal* reqOffset, STqOffsetVal* rspOffset, int64_t sver, int64_t ever,
                          int64_t consumerId, bool hasData) {
+  if (pVg == NULL || reqOffset == NULL || rspOffset == NULL) {
+    return;
+  }
   if (!pVg->seekUpdated) {
     tqDebugC("consumer:0x%" PRIx64 " local offset is update, since seekupdate not set", consumerId);
     if (hasData) {
@@ -2283,6 +2361,9 @@ static SMqRspObj* buildRsp(SMqPollRspWrapper* pollRspWrapper){
     SMqBatchMetaRsp batchMetaRsp;
   } MEMSIZE;
 
+  if (pollRspWrapper == NULL) {
+    return NULL;
+  }
   SMqRspObj* pRspObj = taosMemoryCalloc(1, sizeof(SMqRspObj));
   if (pRspObj == NULL) {
     tqErrorC("buildRsp:failed to allocate memory");
@@ -2297,6 +2378,9 @@ static SMqRspObj* buildRsp(SMqPollRspWrapper* pollRspWrapper){
 }
 
 static void processMqRspError(tmq_t* tmq, SMqRspWrapper* pRspWrapper){
+  if (tmq == NULL || pRspWrapper == NULL) {
+    return;
+  }
   SMqPollRspWrapper* pollRspWrapper = &pRspWrapper->pollRsp;
 
   if (pRspWrapper->code == TSDB_CODE_VND_INVALID_VGROUP_ID) {  // for vnode transform
@@ -2322,6 +2406,9 @@ static void processMqRspError(tmq_t* tmq, SMqRspWrapper* pRspWrapper){
   taosWUnLockLatch(&tmq->lock);
 }
 static SMqRspObj* processMqRsp(tmq_t* tmq, SMqRspWrapper* pRspWrapper){
+  if (tmq == NULL || pRspWrapper == NULL) {
+    return NULL;
+  }
   SMqRspObj* pRspObj = NULL;
 
   if (pRspWrapper->tmqRspType == TMQ_MSG_TYPE__EP_RSP) {
@@ -2401,6 +2488,9 @@ static SMqRspObj* processMqRsp(tmq_t* tmq, SMqRspWrapper* pRspWrapper){
 }
 
 static void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout) {
+  if (tmq == NULL) {
+    return NULL;
+  }
   tqDebugC("consumer:0x%" PRIx64 " start to handle the rsp, total:%d", tmq->consumerId, taosQallItemSize(tmq->qall));
 
   void* returnVal = NULL;
@@ -2478,6 +2568,7 @@ TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t timeout) {
 }
 
 static void displayConsumeStatistics(tmq_t* pTmq) {
+  if (pTmq == NULL) return;
   taosRLockLatch(&pTmq->lock);
   int32_t numOfTopics = taosArrayGetSize(pTmq->clientTopics);
   tqInfoC("consumer:0x%" PRIx64 " closing poll:%" PRId64 " rows:%" PRId64 " topics:%d, final epoch:%d",
@@ -2680,7 +2771,11 @@ void tmq_commit_async(tmq_t* tmq, const TAOS_RES* pRes, tmq_commit_cb* cb, void*
   }
 }
 
-static void commitCallBackFn(tmq_t* UNUSED_PARAM(tmq), int32_t code, void* param) {
+static void commitCallBackFn(tmq_t* tmq, int32_t code, void* param) {
+  if (param == NULL) {
+    tqErrorC("invalid param in commit cb");
+    return;
+  }
   SSyncCommitInfo* pInfo = (SSyncCommitInfo*)param;
   pInfo->code = code;
   if (tsem2_post(&pInfo->sem) != 0){
@@ -2730,6 +2825,10 @@ int32_t tmq_commit_sync(tmq_t* tmq, const TAOS_RES* pRes) {
 
 // wal range will be ok after calling tmq_get_topic_assignment or poll interface
 static int32_t checkWalRange(SVgOffsetInfo* offset, int64_t value) {
+  if (offset == NULL) {
+    tqErrorC("invalid offset, null");
+    return TSDB_CODE_INVALID_PARA;
+  }
   if (offset->walVerBegin == -1 || offset->walVerEnd == -1) {
     tqErrorC("Assignment or poll interface need to be called first");
     return TSDB_CODE_TMQ_NEED_INITIALIZED;
@@ -2848,6 +2947,9 @@ end:
 }
 
 int32_t tmqGetNextResInfo(TAOS_RES* res, bool convertUcs4, SReqResultInfo** pResInfo) {
+        if (res == NULL || pResInfo == NULL) {
+        return TSDB_CODE_INVALID_PARA;
+        }
   SMqRspObj*  pRspObj = (SMqRspObj*)res;
   SMqDataRsp* data = &pRspObj->dataRsp;
 
@@ -2885,7 +2987,7 @@ int32_t tmqGetNextResInfo(TAOS_RES* res, bool convertUcs4, SReqResultInfo** pRes
 }
 
 static int32_t tmqGetWalInfoCb(void* param, SDataBuf* pMsg, int32_t code) {
-  if (param == NULL) {
+  if (param == NULL || pMsg == NULL) {
     return code;
   }
   SMqVgWalInfoParam* pParam = param;
@@ -2958,6 +3060,9 @@ static bool isInSnapshotMode(int8_t type, bool useSnapshot) {
 }
 
 static int32_t tmCommittedCb(void* param, SDataBuf* pMsg, int32_t code) {
+  if (param == NULL) {
+    return code;
+  }
   SMqCommittedParam* pParam = param;
 
   if (code != 0) {
@@ -2987,6 +3092,9 @@ end:
 }
 
 int64_t getCommittedFromServer(tmq_t* tmq, char* tname, int32_t vgId, SEpSet* epSet) {
+  if (tmq == NULL || tname == NULL || epSet == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t     code = 0;
   SMqVgOffset pOffset = {0};
 

source/dnode/mnode/impl/src/mndConsumer.c

@@ -52,6 +52,9 @@ int32_t mndInitConsumer(SMnode *pMnode) {
       .deleteFp = (SdbDeleteFp)mndConsumerActionDelete,
   };
 
+  if (pMnode == NULL){
+    return TSDB_CODE_INVALID_PARA;
+  }
   mndSetMsgHandle(pMnode, TDMT_MND_TMQ_SUBSCRIBE, mndProcessSubscribeReq);
   mndSetMsgHandle(pMnode, TDMT_MND_TMQ_HB, mndProcessMqHbReq);
   mndSetMsgHandle(pMnode, TDMT_MND_TMQ_ASK_EP, mndProcessAskEpReq);
@@ -66,6 +69,9 @@ int32_t mndInitConsumer(SMnode *pMnode) {
 void mndCleanupConsumer(SMnode *pMnode) {}
 
 int32_t mndSendConsumerMsg(SMnode *pMnode, int64_t consumerId, uint16_t msgType, SRpcHandleInfo *info) {
+  if (pMnode == NULL || info == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = 0;
   void   *msg  = rpcMallocCont(sizeof(int64_t));
   MND_TMQ_NULL_CHECK(msg);
@@ -88,6 +94,9 @@ END:
 }
 
 static int32_t validateTopics(STrans* pTrans, SCMSubscribeReq *subscribe, SMnode *pMnode, const char *pUser) {
+  if (pTrans == NULL || subscribe == NULL || pMnode == NULL || pUser == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   SMqTopicObj *pTopic = NULL;
   int32_t      code = 0;
 
@@ -130,6 +139,9 @@ END:
 }
 
 static int32_t mndProcessConsumerClearMsg(SRpcMsg *pMsg) {
+  if (pMsg == NULL || pMsg->pCont == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t              code = 0;
   SMnode              *pMnode = pMsg->info.node;
   SMqConsumerClearMsg *pClearMsg = pMsg->pCont;
@@ -155,6 +167,9 @@ END:
 }
 
 static int32_t checkPrivilege(SMnode *pMnode, SMqConsumerObj *pConsumer, SMqHbRsp *rsp, char *user) {
+  if (pMnode == NULL || pConsumer == NULL || rsp == NULL || user == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = 0;
   rsp->topicPrivileges = taosArrayInit(taosArrayGetSize(pConsumer->currentTopics), sizeof(STopicPrivilege));
   MND_TMQ_NULL_CHECK(rsp->topicPrivileges);
@@ -181,6 +196,9 @@ END:
 }
 
 static void storeOffsetRows(SMnode *pMnode, SMqHbReq *req, SMqConsumerObj *pConsumer){
+  if (pMnode == NULL || req == NULL || pConsumer == NULL){
+    return;
+  }
   for (int i = 0; i < taosArrayGetSize(req->topics); i++) {
     TopicOffsetRows *data = taosArrayGet(req->topics, i);
     if (data == NULL){
@@ -210,6 +228,9 @@ static void storeOffsetRows(SMnode *pMnode, SMqHbReq *req, SMqConsumerObj *pCons
 }
 
 static int32_t buildMqHbRsp(SRpcMsg *pMsg, SMqHbRsp *rsp){
+  if (pMsg == NULL || rsp == NULL){
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t tlen = tSerializeSMqHbRsp(NULL, 0, rsp);
   if (tlen <= 0){
     return TSDB_CODE_TMQ_INVALID_MSG;
@@ -229,6 +250,9 @@ static int32_t buildMqHbRsp(SRpcMsg *pMsg, SMqHbRsp *rsp){
 }
 
 static int32_t mndProcessMqHbReq(SRpcMsg *pMsg) {
+  if (pMsg == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t         code = 0;
   SMnode         *pMnode = pMsg->info.node;
   SMqHbReq        req = {0};
@@ -256,6 +280,9 @@ END:
 }
 
 static int32_t addEpSetInfo(SMnode *pMnode, SMqConsumerObj *pConsumer, int32_t epoch, SMqAskEpRsp *rsp){
+  if (pMnode == NULL || pConsumer == NULL || rsp == NULL){
+    return TSDB_CODE_INVALID_PARA;
+  }
   taosRLockLatch(&pConsumer->lock);
 
   int32_t numOfTopics = taosArrayGetSize(pConsumer->currentTopics);
@@ -359,6 +386,9 @@ static int32_t addEpSetInfo(SMnode *pMnode, SMqConsumerObj *pConsumer, int32_t e
 }
 
 static int32_t buildAskEpRsp(SRpcMsg *pMsg, SMqAskEpRsp *rsp, int32_t serverEpoch, int64_t consumerId){
+  if (pMsg == NULL || rsp == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = 0;
   // encode rsp
   int32_t tlen = sizeof(SMqRspHead) + tEncodeSMqAskEpRsp(NULL, rsp);
@@ -388,6 +418,9 @@ static int32_t buildAskEpRsp(SRpcMsg *pMsg, SMqAskEpRsp *rsp, int32_t serverEpoc
 }
 
 static int32_t mndProcessAskEpReq(SRpcMsg *pMsg) {
+  if (pMsg == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   SMnode     *pMnode = pMsg->info.node;
   SMqAskEpReq req = {0};
   SMqAskEpRsp rsp = {0};
@@ -431,6 +464,9 @@ END:
 }
 
 int32_t mndSetConsumerDropLogs(STrans *pTrans, SMqConsumerObj *pConsumer) {
+  if (pConsumer == NULL || pTrans == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t  code = 0;
   SSdbRaw *pCommitRaw = mndConsumerActionEncode(pConsumer);
   MND_TMQ_NULL_CHECK(pCommitRaw);
@@ -445,6 +481,9 @@ END:
 }
 
 int32_t mndSetConsumerCommitLogs(STrans *pTrans, SMqConsumerObj *pConsumer) {
+  if (pConsumer == NULL || pTrans == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t  code = 0;
   SSdbRaw *pCommitRaw = mndConsumerActionEncode(pConsumer);
   MND_TMQ_NULL_CHECK(pCommitRaw);
@@ -459,6 +498,9 @@ END:
 }
 
 static void freeItem(void *param) {
+  if (param == NULL) {
+    return;
+  }
   void *pItem = *(void **)param;
   if (pItem != NULL) {
     taosMemoryFree(pItem);
@@ -475,6 +517,9 @@ if (taosArrayPush(pConsumerNew->array, &newTopicCopy) == NULL){\
 }
 
 static int32_t getTopicAddDelete(SMqConsumerObj *pExistedConsumer, SMqConsumerObj *pConsumerNew){
+  if (pExistedConsumer == NULL || pConsumerNew == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = 0;
   pConsumerNew->rebNewTopics = taosArrayInit(0, sizeof(void *));
   MND_TMQ_NULL_CHECK(pConsumerNew->rebNewTopics);
@@ -528,6 +573,9 @@ END:
 }
 
 static int32_t checkAndSortTopic(SMnode *pMnode, SArray *pTopicList){
+  if (pTopicList == NULL || pMnode == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   taosArraySort(pTopicList, taosArrayCompareString);
   taosArrayRemoveDuplicate(pTopicList, taosArrayCompareString, freeItem);
 
@@ -542,6 +590,9 @@ static int32_t checkAndSortTopic(SMnode *pMnode, SArray *pTopicList){
 }
 
 static int32_t buildSubConsumer(SMnode *pMnode, SCMSubscribeReq *subscribe, SMqConsumerObj** ppConsumer){
+  if (pMnode == NULL || subscribe == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int64_t         consumerId = subscribe->consumerId;
   char           *cgroup     = subscribe->cgroup;
   SMqConsumerObj *pConsumerNew     = NULL;
@@ -581,6 +632,9 @@ END:
 }
 
 int32_t mndProcessSubscribeReq(SRpcMsg *pMsg) {
+  if (pMsg == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   SMnode *pMnode = pMsg->info.node;
   char   *msgStr = pMsg->pCont;
   int32_t code = 0;
@@ -619,6 +673,9 @@ END:
 }
 
 SSdbRaw *mndConsumerActionEncode(SMqConsumerObj *pConsumer) {
+  if (pConsumer == NULL) {
+    return NULL;
+  }
   int32_t code = 0;
   int32_t lino = 0;
   terrno = TSDB_CODE_OUT_OF_MEMORY;
@@ -659,6 +716,9 @@ CM_ENCODE_OVER:
 }
 
 SSdbRow *mndConsumerActionDecode(SSdbRaw *pRaw) {
+  if (pRaw == NULL) {
+    return NULL;
+  }
   int32_t         code = 0;
   int32_t         lino = 0;
   SSdbRow        *pRow = NULL;
@@ -717,6 +777,9 @@ CM_DECODE_OVER:
 }
 
 static int32_t mndConsumerActionInsert(SSdb *pSdb, SMqConsumerObj *pConsumer) {
+  if (pConsumer == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   mInfo("consumer:0x%" PRIx64 " sub insert, cgroup:%s status:%d(%s) epoch:%d", pConsumer->consumerId, pConsumer->cgroup,
         pConsumer->status, mndConsumerStatusName(pConsumer->status), pConsumer->epoch);
   pConsumer->subscribeTime = pConsumer->createTime;
@@ -724,6 +787,9 @@ static int32_t mndConsumerActionInsert(SSdb *pSdb, SMqConsumerObj *pConsumer) {
 }
 
 static int32_t mndConsumerActionDelete(SSdb *pSdb, SMqConsumerObj *pConsumer) {
+  if (pConsumer == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   mInfo("consumer:0x%" PRIx64 " perform delete action, status:(%d)%s", pConsumer->consumerId, pConsumer->status,
         mndConsumerStatusName(pConsumer->status));
   tClearSMqConsumerObj(pConsumer);
@@ -744,6 +810,9 @@ static int32_t mndConsumerActionDelete(SSdb *pSdb, SMqConsumerObj *pConsumer) {
 
 // remove from topic list
 static void removeFromTopicList(SArray *topicList, const char *pTopic, int64_t consumerId, char *type) {
+  if (topicList == NULL || pTopic == NULL) {
+    return;
+  }
   int32_t size = taosArrayGetSize(topicList);
   for (int32_t i = 0; i < size; i++) {
     char *p = taosArrayGetP(topicList, i);
@@ -759,6 +828,9 @@ static void removeFromTopicList(SArray *topicList, const char *pTopic, int64_t c
 }
 
 static bool existInCurrentTopicList(const SMqConsumerObj *pConsumer, const char *pTopic) {
+  if (pConsumer == NULL || pTopic == NULL) {
+    return false;
+  }
   bool    existing = false;
   int32_t size = taosArrayGetSize(pConsumer->currentTopics);
   for (int32_t i = 0; i < size; i++) {
@@ -773,6 +845,9 @@ static bool existInCurrentTopicList(const SMqConsumerObj *pConsumer, const char
 }
 
 static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pOldConsumer, SMqConsumerObj *pNewConsumer) {
+  if (pOldConsumer == NULL || pNewConsumer == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   mInfo("consumer:0x%" PRIx64 " perform update action, update type:%d, subscribe-time:%" PRId64 ", createTime:%" PRId64,
         pOldConsumer->consumerId, pNewConsumer->updateType, pOldConsumer->subscribeTime, pOldConsumer->createTime);
 
@@ -857,6 +932,9 @@ static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pOldConsumer,
 }
 
 int32_t mndAcquireConsumer(SMnode *pMnode, int64_t consumerId, SMqConsumerObj** pConsumer) {
+  if (pMnode == NULL || pConsumer == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   SSdb           *pSdb = pMnode->pSdb;
   *pConsumer = sdbAcquire(pSdb, SDB_CONSUMER, &consumerId);
   if (*pConsumer == NULL) {
@@ -866,11 +944,17 @@ int32_t mndAcquireConsumer(SMnode *pMnode, int64_t consumerId, SMqConsumerObj**
 }
 
 void mndReleaseConsumer(SMnode *pMnode, SMqConsumerObj *pConsumer) {
+  if (pMnode == NULL || pConsumer == NULL) {
+    return;
+  }
   SSdb *pSdb = pMnode->pSdb;
   sdbRelease(pSdb, pConsumer);
 }
 
 static int32_t mndRetrieveConsumer(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rowsCapacity) {
+  if (pReq == NULL || pShow == NULL || pBlock == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   SMnode         *pMnode = pReq->info.node;
   SSdb           *pSdb = pMnode->pSdb;
   int32_t         numOfRows = 0;
@@ -1021,6 +1105,7 @@ END:
 }
 
 static void mndCancelGetNextConsumer(SMnode *pMnode, void *pIter) {
+  if (pMnode == NULL || pIter == NULL) return;
   SSdb *pSdb = pMnode->pSdb;
   sdbCancelFetchByType(pSdb, pIter, SDB_CONSUMER);
 }

source/dnode/mnode/impl/src/mndSubscribe.c

@@ -43,6 +43,9 @@ static void     mndCancelGetNextSubscribe(SMnode *pMnode, void *pIter);
 static int32_t  mndCheckConsumer(SRpcMsg *pMsg, SHashObj *hash);
 
 static int32_t mndSetSubCommitLogs(STrans *pTrans, SMqSubscribeObj *pSub) {
+  if (pTrans == NULL || pSub == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t  code = 0;
   SSdbRaw *pCommitRaw = mndSubActionEncode(pSub);
   MND_TMQ_NULL_CHECK(pCommitRaw);
@@ -68,6 +71,9 @@ int32_t mndInitSubscribe(SMnode *pMnode) {
       .deleteFp = (SdbDeleteFp)mndSubActionDelete,
   };
 
+  if (pMnode == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   mndSetMsgHandle(pMnode, TDMT_VND_TMQ_SUBSCRIBE_RSP, mndTransProcessRsp);
   mndSetMsgHandle(pMnode, TDMT_VND_TMQ_DELETE_SUB_RSP, mndTransProcessRsp);
   mndSetMsgHandle(pMnode, TDMT_MND_TMQ_TIMER, mndProcessRebalanceReq);
@@ -81,6 +87,9 @@ int32_t mndInitSubscribe(SMnode *pMnode) {
 }
 
 static int32_t mndCreateSubscription(SMnode *pMnode, const SMqTopicObj *pTopic, const char *subKey, SMqSubscribeObj** pSub) {
+  if(pMnode == NULL || pTopic == NULL || subKey == NULL || pSub == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = 0;
   MND_TMQ_RETURN_CHECK(tNewSubscribeObj(subKey, pSub));
   (*pSub)->dbUid = pTopic->dbUid;
@@ -99,6 +108,9 @@ END:
 
 static int32_t mndBuildSubChangeReq(void **pBuf, int32_t *pLen, SMqSubscribeObj *pSub, const SMqRebOutputVg *pRebVg,
                                     SSubplan *pPlan) {
+  if (pSub == NULL || pRebVg == NULL || pBuf == NULL || pLen == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   SMqRebVgReq req = {0};
   int32_t     code = 0;
   SEncoder encoder = {0};
@@ -146,6 +158,9 @@ END:
 
 static int32_t mndPersistSubChangeVgReq(SMnode *pMnode, STrans *pTrans, SMqSubscribeObj *pSub,
                                         const SMqRebOutputVg *pRebVg, SSubplan *pPlan) {
+  if (pMnode == NULL || pTrans == NULL || pSub == NULL || pRebVg == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = 0;
   void   *buf  = NULL;
 
@@ -180,6 +195,9 @@ END:
 }
 
 static void mndSplitSubscribeKey(const char *key, char *topic, char *cgroup, bool fullName) {
+  if (key == NULL || topic == NULL || cgroup == NULL) {
+    return;
+  }
   int32_t i = 0;
   while (key[i] != TMQ_SEPARATOR_CHAR) {
     i++;
@@ -197,6 +215,9 @@ static void mndSplitSubscribeKey(const char *key, char *topic, char *cgroup, boo
 }
 
 static int32_t mndGetOrCreateRebSub(SHashObj *pHash, const char *key, SMqRebInfo **pReb) {
+  if (pHash == NULL || key == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = 0;
   SMqRebInfo* pRebInfo = taosHashGet(pHash, key, strlen(key) + 1);
   if (pRebInfo == NULL) {
@@ -222,6 +243,9 @@ END:
 }
 
 static int32_t pushVgDataToHash(SArray *vgs, SHashObj *pHash, int64_t consumerId, char *key) {
+  if (vgs == NULL || pHash == NULL || key == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t         code = 0;
   SMqVgEp       **pVgEp = (SMqVgEp **)taosArrayPop(vgs);
   MND_TMQ_NULL_CHECK(pVgEp);
@@ -233,6 +257,9 @@ END:
 }
 
 static int32_t processRemovedConsumers(SMqRebOutputObj *pOutput, SHashObj *pHash, const SMqRebInputObj *pInput) {
+  if (pHash == NULL || pOutput == NULL || pInput == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = 0;
   int32_t numOfRemoved = taosArrayGetSize(pInput->pRebInfo->removedConsumers);
   int32_t actualRemoved = 0;
@@ -266,6 +293,9 @@ END:
 }
 
 static int32_t processNewConsumers(SMqRebOutputObj *pOutput, const SMqRebInputObj *pInput) {
+  if (pOutput == NULL || pInput == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = 0;
   int32_t numOfNewConsumers = taosArrayGetSize(pInput->pRebInfo->newConsumers);
 
@@ -285,6 +315,9 @@ END:
 }
 
 static int32_t processUnassignedVgroups(SMqRebOutputObj *pOutput, SHashObj *pHash) {
+  if (pOutput == NULL || pHash == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = 0;
   int32_t numOfVgroups = taosArrayGetSize(pOutput->pSub->unassignedVgs);
   for (int32_t i = 0; i < numOfVgroups; i++) {
@@ -296,6 +329,9 @@ END:
 
 static int32_t processModifiedConsumers(SMqRebOutputObj *pOutput, SHashObj *pHash, int32_t minVgCnt,
                                      int32_t remainderVgCnt) {
+  if (pOutput == NULL || pHash == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = 0;
   int32_t cnt = 0;
   void   *pIter = NULL;
@@ -328,6 +364,9 @@ END:
 }
 
 static int32_t processRemoveAddVgs(SMnode *pMnode, SMqRebOutputObj *pOutput) {
+  if (pMnode == NULL || pOutput == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = 0;
   int32_t totalVgNum = 0;
   SVgObj *pVgroup = NULL;
@@ -403,6 +442,9 @@ END:
 }
 
 static int32_t processSubOffsetRows(SMnode *pMnode, const SMqRebInputObj *pInput, SMqRebOutputObj *pOutput) {
+  if (pMnode == NULL || pInput == NULL || pOutput == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   SMqSubscribeObj *pSub = NULL;
   int32_t          code = mndAcquireSubscribeByKey(pMnode, pInput->pRebInfo->key, &pSub);  // put all offset rows
   if( code != 0){
@@ -465,6 +507,7 @@ END:
 }
 
 static void printRebalanceLog(SMqRebOutputObj *pOutput) {
+  if (pOutput == NULL) return;
   mInfo("sub:%s mq rebalance calculation completed, re-balanced vg", pOutput->pSub->key);
   for (int32_t i = 0; i < taosArrayGetSize(pOutput->rebVgs); i++) {
     SMqRebOutputVg *pOutputRebVg = taosArrayGet(pOutput->rebVgs, i);
@@ -492,6 +535,9 @@ static void printRebalanceLog(SMqRebOutputObj *pOutput) {
 
 static void calcVgroupsCnt(const SMqRebInputObj *pInput, int32_t totalVgNum, const char *pSubKey, int32_t *minVgCnt,
                            int32_t *remainderVgCnt) {
+  if (pInput == NULL || pSubKey == NULL || minVgCnt == NULL || remainderVgCnt == NULL) {
+    return;
+  }
   int32_t numOfRemoved = taosArrayGetSize(pInput->pRebInfo->removedConsumers);
   int32_t numOfAdded = taosArrayGetSize(pInput->pRebInfo->newConsumers);
   int32_t numOfFinal = pInput->oldConsumerNum + numOfAdded - numOfRemoved;
@@ -509,6 +555,9 @@ static void calcVgroupsCnt(const SMqRebInputObj *pInput, int32_t totalVgNum, con
 }
 
 static int32_t assignVgroups(SMqRebOutputObj *pOutput, SHashObj *pHash, int32_t minVgCnt) {
+  if (pOutput == NULL || pHash == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   SMqRebOutputVg *pRebVg = NULL;
   void           *pAssignIter = NULL;
   void           *pIter = NULL;
@@ -580,6 +629,9 @@ END:
 }
 
 static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqRebOutputObj *pOutput) {
+  if (pMnode == NULL || pInput == NULL || pOutput == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t     totalVgNum = processRemoveAddVgs(pMnode, pOutput);
   if (totalVgNum < 0){
     return totalVgNum;
@@ -605,6 +657,9 @@ END:
 }
 
 static int32_t presistConsumerByType(STrans *pTrans, SArray *consumers, int8_t type, char *cgroup, char *topic) {
+  if (pTrans == NULL || consumers == NULL || cgroup == NULL || topic == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t         code = 0;
   SMqConsumerObj *pConsumerNew = NULL;
   int32_t         consumerNum = taosArrayGetSize(consumers);
@@ -623,6 +678,9 @@ END:
 }
 
 static int32_t mndPresistConsumer(STrans *pTrans, const SMqRebOutputObj *pOutput, char *cgroup, char *topic) {
+  if (pTrans == NULL || pOutput == NULL || cgroup == NULL || topic == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = 0;
   MND_TMQ_RETURN_CHECK(presistConsumerByType(pTrans, pOutput->modifyConsumers, CONSUMER_UPDATE_REB, cgroup, NULL));
   MND_TMQ_RETURN_CHECK(presistConsumerByType(pTrans, pOutput->newConsumers, CONSUMER_ADD_REB, cgroup, topic));
@@ -632,6 +690,9 @@ END:
 }
 
 static int32_t mndPersistRebResult(SMnode *pMnode, SRpcMsg *pMsg, const SMqRebOutputObj *pOutput) {
+  if (pMnode == NULL || pMsg == NULL || pOutput == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   struct SSubplan *pPlan = NULL;
   int32_t          code = 0;
   STrans          *pTrans = NULL;
@@ -682,6 +743,7 @@ END:
 }
 
 static void freeRebalanceItem(void *param) {
+  if (param == NULL) return;
   SMqRebInfo *pInfo = param;
   taosArrayDestroy(pInfo->newConsumers);
   taosArrayDestroy(pInfo->removedConsumers);
@@ -689,6 +751,9 @@ static void freeRebalanceItem(void *param) {
 
 // type = 0 remove  type = 1 add
 static int32_t buildRebInfo(SHashObj *rebSubHash, SArray *topicList, int8_t type, char *group, int64_t consumerId) {
+  if (rebSubHash == NULL || topicList == NULL || group == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = 0;
   int32_t topicNum = taosArrayGetSize(topicList);
   for (int32_t i = 0; i < topicNum; i++) {
@@ -709,6 +774,9 @@ END:
 }
 
 static void checkForVgroupSplit(SMnode *pMnode, SMqConsumerObj *pConsumer, SHashObj *rebSubHash) {
+  if (pMnode == NULL || pConsumer == NULL || rebSubHash == NULL) {
+    return;
+  }
   int32_t newTopicNum = taosArrayGetSize(pConsumer->currentTopics);
   for (int32_t i = 0; i < newTopicNum; i++) {
     char            *topic = taosArrayGetP(pConsumer->currentTopics, i);
@@ -754,6 +822,9 @@ static void checkForVgroupSplit(SMnode *pMnode, SMqConsumerObj *pConsumer, SHash
 }
 
 static int32_t mndCheckConsumer(SRpcMsg *pMsg, SHashObj *rebSubHash) {
+  if (pMsg == NULL || rebSubHash == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   SMnode         *pMnode = pMsg->info.node;
   SSdb           *pSdb = pMnode->pSdb;
   SMqConsumerObj *pConsumer = NULL;
@@ -818,6 +889,9 @@ void mndRebCntDec() {
 }
 
 static void clearRebOutput(SMqRebOutputObj *rebOutput) {
+  if (rebOutput == NULL) {
+    return;
+  }
   taosArrayDestroy(rebOutput->newConsumers);
   taosArrayDestroy(rebOutput->modifyConsumers);
   taosArrayDestroy(rebOutput->removedConsumers);
@@ -827,6 +901,9 @@ static void clearRebOutput(SMqRebOutputObj *rebOutput) {
 }
 
 static int32_t initRebOutput(SMqRebOutputObj *rebOutput) {
+  if (rebOutput == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = 0;
   rebOutput->newConsumers = taosArrayInit(0, sizeof(int64_t));
   MND_TMQ_NULL_CHECK(rebOutput->newConsumers);
@@ -845,6 +922,9 @@ END:
 
 // This function only works when there are dirty consumers
 static int32_t checkConsumer(SMnode *pMnode, SMqSubscribeObj *pSub) {
+  if (pMnode == NULL || pSub == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = 0;
   void   *pIter = NULL;
   while (1) {
@@ -871,6 +951,9 @@ END:
 }
 
 static int32_t buildRebOutput(SMnode *pMnode, SMqRebInputObj *rebInput, SMqRebOutputObj *rebOutput) {
+  if (pMnode == NULL || rebInput == NULL || rebOutput == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   const char      *key = rebInput->pRebInfo->key;
   SMqSubscribeObj *pSub = NULL;
   int32_t          code = mndAcquireSubscribeByKey(pMnode, key, &pSub);
@@ -922,6 +1005,9 @@ END:
 }
 
 static int32_t mndProcessRebalanceReq(SRpcMsg *pMsg) {
+  if (pMsg == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int     code = 0;
   void   *pIter = NULL;
   SMnode *pMnode = pMsg->info.node;
@@ -986,6 +1072,9 @@ END:
 }
 
 static int32_t sendDeleteSubToVnode(SMnode *pMnode, SMqSubscribeObj *pSub, STrans *pTrans) {
+  if (pMnode == NULL || pSub == NULL || pTrans == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   void   *pIter = NULL;
   SVgObj *pVgObj = NULL;
   int32_t code = 0;
@@ -1024,6 +1113,9 @@ END:
 }
 
 static int32_t mndCheckConsumerByGroup(SMnode *pMnode, STrans *pTrans, char *cgroup, char *topic) {
+  if (pMnode == NULL || pTrans == NULL || cgroup == NULL || topic == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   void           *pIter = NULL;
   SMqConsumerObj *pConsumer = NULL;
   int             code = 0;
@@ -1056,6 +1148,9 @@ END:
 }
 
 static int32_t mndProcessDropCgroupReq(SRpcMsg *pMsg) {
+  if (pMsg == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   SMnode         *pMnode = pMsg->info.node;
   SMDropCgroupReq dropReq = {0};
   STrans         *pTrans = NULL;
@@ -1109,6 +1204,9 @@ END:
 void mndCleanupSubscribe(SMnode *pMnode) {}
 
 static SSdbRaw *mndSubActionEncode(SMqSubscribeObj *pSub) {
+  if (pSub == NULL) {
+    return NULL;
+  }
   int32_t code = 0;
   int32_t lino = 0;
   terrno = TSDB_CODE_OUT_OF_MEMORY;
@@ -1149,6 +1247,9 @@ SUB_ENCODE_OVER:
 }
 
 static SSdbRow *mndSubActionDecode(SSdbRaw *pRaw) {
+  if (pRaw == NULL) {
+    return NULL;
+  }
   int32_t code = 0;
   int32_t lino = 0;
   terrno = TSDB_CODE_OUT_OF_MEMORY;
@@ -1218,17 +1319,18 @@ SUB_DECODE_OVER:
 }
 
 static int32_t mndSubActionInsert(SSdb *pSdb, SMqSubscribeObj *pSub) {
-  mTrace("subscribe:%s, perform insert action", pSub->key);
+  mTrace("subscribe:%s, perform insert action", pSub != NULL ? pSub->key : "null");
   return 0;
 }
 
 static int32_t mndSubActionDelete(SSdb *pSdb, SMqSubscribeObj *pSub) {
-  mTrace("subscribe:%s, perform delete action", pSub->key);
+  mTrace("subscribe:%s, perform delete action", pSub != NULL ? pSub->key : "null");
   tDeleteSubscribeObj(pSub);
   return 0;
 }
 
 static int32_t mndSubActionUpdate(SSdb *pSdb, SMqSubscribeObj *pOldSub, SMqSubscribeObj *pNewSub) {
+  if (pOldSub == NULL || pNewSub == NULL) return -1;
   mTrace("subscribe:%s, perform update action", pOldSub->key);
   taosWLockLatch(&pOldSub->lock);
 
@@ -1249,6 +1351,9 @@ static int32_t mndSubActionUpdate(SSdb *pSdb, SMqSubscribeObj *pOldSub, SMqSubsc
 }
 
 int32_t mndAcquireSubscribeByKey(SMnode *pMnode, const char *key, SMqSubscribeObj** pSub) {
+  if (pMnode == NULL || key == NULL || pSub == NULL){
+    return TSDB_CODE_INVALID_PARA;
+  }
   SSdb            *pSdb = pMnode->pSdb;
   *pSub = sdbAcquire(pSdb, SDB_SUBSCRIBE, key);
   if (*pSub == NULL) {
@@ -1258,6 +1363,7 @@ int32_t mndAcquireSubscribeByKey(SMnode *pMnode, const char *key, SMqSubscribeOb
 }
 
 int32_t mndGetGroupNumByTopic(SMnode *pMnode, const char *topicName) {
+  if (pMnode == NULL || topicName == NULL) return 0;
   int32_t num = 0;
   SSdb   *pSdb = pMnode->pSdb;
 
@@ -1283,11 +1389,13 @@ int32_t mndGetGroupNumByTopic(SMnode *pMnode, const char *topicName) {
 }
 
 void mndReleaseSubscribe(SMnode *pMnode, SMqSubscribeObj *pSub) {
+  if (pMnode == NULL || pSub == NULL) return;
   SSdb *pSdb = pMnode->pSdb;
   sdbRelease(pSdb, pSub);
 }
 
 int32_t mndSetDropSubCommitLogs(SMnode *pMnode, STrans *pTrans, SMqSubscribeObj *pSub) {
+  if (pMnode == NULL || pTrans == NULL || pSub == NULL) return TSDB_CODE_INVALID_PARA;
   int32_t  code = 0;
   SSdbRaw *pCommitRaw = mndSubActionEncode(pSub);
   MND_TMQ_NULL_CHECK(pCommitRaw);
@@ -1302,6 +1410,7 @@ END:
 }
 
 int32_t mndDropSubByTopic(SMnode *pMnode, STrans *pTrans, const char *topicName) {
+  if (pMnode == NULL || pTrans == NULL || topicName == NULL) return TSDB_CODE_INVALID_PARA;
   SSdb            *pSdb = pMnode->pSdb;
   int32_t          code = 0;
   void            *pIter = NULL;
@@ -1337,6 +1446,10 @@ END:
 
 static int32_t buildResult(SSDataBlock *pBlock, int32_t *numOfRows, int64_t consumerId, const char* user, const char* fqdn,
                            const char *topic, const char *cgroup, SArray *vgs, SArray *offsetRows) {
+  if (pBlock == NULL || numOfRows == NULL || topic == NULL ||
+      cgroup == NULL || vgs == NULL || offsetRows == NULL){
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = 0;
   int32_t sz = taosArrayGetSize(vgs);
   for (int32_t j = 0; j < sz; j++) {
@@ -1424,6 +1537,9 @@ END:
 }
 
 int32_t mndRetrieveSubscribe(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rowsCapacity) {
+  if (pReq == NULL || pShow == NULL || pBlock == NULL){
+    return TSDB_CODE_INVALID_PARA;
+  }
   SMnode          *pMnode = pReq->info.node;
   SSdb            *pSdb = pMnode->pSdb;
   int32_t          numOfRows = 0;
@@ -1489,6 +1605,9 @@ END:
 }
 
 void mndCancelGetNextSubscribe(SMnode *pMnode, void *pIter) {
+  if (pMnode == NULL) {
+    return;
+  }
   SSdb *pSdb = pMnode->pSdb;
   sdbCancelFetchByType(pSdb, pIter, SDB_SUBSCRIBE);
 }

source/dnode/mnode/impl/src/mndTopic.c

@@ -55,6 +55,9 @@ int32_t mndInitTopic(SMnode *pMnode) {
       .deleteFp = (SdbDeleteFp)mndTopicActionDelete,
   };
 
+  if (pMnode == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   mndSetMsgHandle(pMnode, TDMT_MND_TMQ_CREATE_TOPIC, mndProcessCreateTopicReq);
   mndSetMsgHandle(pMnode, TDMT_MND_TMQ_DROP_TOPIC, mndProcessDropTopicReq);
   mndSetMsgHandle(pMnode, TDMT_VND_TMQ_ADD_CHECKINFO_RSP, mndTransProcessRsp);
@@ -81,6 +84,9 @@ void mndTopicGetShowName(const char* fullTopic, char* topic) {
 }
 
 SSdbRaw *mndTopicActionEncode(SMqTopicObj *pTopic) {
+  if (pTopic == NULL) {
+    return NULL;
+  }
   int32_t code = 0;
   int32_t lino = 0;
   terrno = TSDB_CODE_OUT_OF_MEMORY;
@@ -172,6 +178,7 @@ TOPIC_ENCODE_OVER:
 }
 
 SSdbRow *mndTopicActionDecode(SSdbRaw *pRaw) {
+  if (pRaw == NULL) return NULL;
   int32_t code = 0;
   int32_t lino = 0;
   terrno = TSDB_CODE_OUT_OF_MEMORY;
@@ -193,7 +200,7 @@ SSdbRow *mndTopicActionDecode(SSdbRaw *pRaw) {
   pTopic = sdbGetRowObj(pRow);
   if (pTopic == NULL) goto TOPIC_DECODE_OVER;
 
-  int32_t len;
+  int32_t len = 0;
   int32_t dataPos = 0;
   SDB_GET_BINARY(pRaw, dataPos, pTopic->name, TSDB_TOPIC_FNAME_LEN, TOPIC_DECODE_OVER);
   SDB_GET_BINARY(pRaw, dataPos, pTopic->db, TSDB_DB_FNAME_LEN, TOPIC_DECODE_OVER);
@@ -292,11 +299,12 @@ TOPIC_DECODE_OVER:
 }
 
 static int32_t mndTopicActionInsert(SSdb *pSdb, SMqTopicObj *pTopic) {
-  mTrace("topic:%s perform insert action", pTopic->name);
+  mTrace("topic:%s perform insert action", pTopic != NULL ? pTopic->name : "null");
   return 0;
 }
 
 static int32_t mndTopicActionDelete(SSdb *pSdb, SMqTopicObj *pTopic) {
+  if (pTopic == NULL) return 0;
   mTrace("topic:%s perform delete action", pTopic->name);
   taosMemoryFreeClear(pTopic->sql);
   taosMemoryFreeClear(pTopic->ast);
@@ -307,6 +315,7 @@ static int32_t mndTopicActionDelete(SSdb *pSdb, SMqTopicObj *pTopic) {
 }
 
 static int32_t mndTopicActionUpdate(SSdb *pSdb, SMqTopicObj *pOldTopic, SMqTopicObj *pNewTopic) {
+  if (pOldTopic == NULL || pNewTopic == NULL) return 0;
   mTrace("topic:%s perform update action", pOldTopic->name);
   (void)atomic_exchange_64(&pOldTopic->updateTime, pNewTopic->updateTime);
   (void)atomic_exchange_32(&pOldTopic->version, pNewTopic->version);
@@ -315,6 +324,9 @@ static int32_t mndTopicActionUpdate(SSdb *pSdb, SMqTopicObj *pOldTopic, SMqTopic
 }
 
 int32_t mndAcquireTopic(SMnode *pMnode, const char *topicName, SMqTopicObj **pTopic) {
+  if (pMnode == NULL || topicName == NULL || pTopic == NULL){
+    return TSDB_CODE_INVALID_PARA;
+  }
   SSdb        *pSdb = pMnode->pSdb;
   *pTopic = sdbAcquire(pSdb, SDB_TOPIC, topicName);
   if (*pTopic == NULL) {
@@ -324,11 +336,13 @@ int32_t mndAcquireTopic(SMnode *pMnode, const char *topicName, SMqTopicObj **pTo
 }
 
 void mndReleaseTopic(SMnode *pMnode, SMqTopicObj *pTopic) {
+  if (pMnode == NULL) return;
   SSdb *pSdb = pMnode->pSdb;
   sdbRelease(pSdb, pTopic);
 }
 
 static int32_t mndCheckCreateTopicReq(SCMCreateTopicReq *pCreate) {
+  if (pCreate == NULL) return TSDB_CODE_INVALID_PARA;
   if (pCreate->sql == NULL) return TSDB_CODE_MND_INVALID_TOPIC;
 
   if (pCreate->subType == TOPIC_SUB_TYPE__COLUMN) {
@@ -343,6 +357,7 @@ static int32_t mndCheckCreateTopicReq(SCMCreateTopicReq *pCreate) {
 }
 
 static int32_t extractTopicTbInfo(SNode *pAst, SMqTopicObj *pTopic) {
+  if (pAst == NULL || pTopic == NULL) return TSDB_CODE_INVALID_PARA;
   SNodeList *pNodeList = NULL;
   int32_t   code = 0;
   MND_TMQ_RETURN_CHECK(nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList));
@@ -367,6 +382,7 @@ END:
 }
 
 static int32_t sendCheckInfoToVnode(STrans *pTrans, SMnode *pMnode, SMqTopicObj *topicObj){
+  if (pTrans == NULL || pMnode == NULL || topicObj == NULL) return TSDB_CODE_INVALID_PARA;
   STqCheckInfo info = {0};
   (void)memcpy(info.topic, topicObj->name, TSDB_TOPIC_FNAME_LEN);
   info.ntbUid = topicObj->ntbUid;
@@ -388,7 +404,7 @@ static int32_t sendCheckInfoToVnode(STrans *pTrans, SMnode *pMnode, SMqTopicObj
     }
 
     // encoder check alter info
-    int32_t len;
+    int32_t len = 0;
     tEncodeSize(tEncodeSTqCheckInfo, &info, len, code);
     if (code != 0) {
       code = TSDB_CODE_OUT_OF_MEMORY;
@@ -426,6 +442,7 @@ END:
 
 static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *pCreate, SDbObj *pDb,
                               const char *userName) {
+  if (pMnode == NULL || pReq == NULL || pCreate == NULL || pDb == NULL || userName == NULL) return TSDB_CODE_INVALID_PARA;
   mInfo("start to create topic:%s", pCreate->name);
   STrans *pTrans = NULL;
   int32_t code = 0;
@@ -519,6 +536,9 @@ END:
 }
 
 static int32_t mndProcessCreateTopicReq(SRpcMsg *pReq) {
+  if (pReq == NULL || pReq->contLen <= 0) {
+    return TSDB_CODE_INVALID_MSG;
+  }
   SMnode           *pMnode = pReq->info.node;
   int32_t           code = TDB_CODE_SUCCESS;
   SMqTopicObj      *pTopic = NULL;
@@ -596,6 +616,9 @@ END:
 }
 
 static int32_t mndDropTopic(SMnode *pMnode, STrans *pTrans, SRpcMsg *pReq, SMqTopicObj *pTopic) {
+  if (pMnode == NULL || pTrans == NULL || pReq == NULL || pTopic == NULL) {
+    return TSDB_CODE_INVALID_MSG;
+  }
   int32_t code = 0;
   SSdbRaw *pCommitRaw = NULL;
   MND_TMQ_RETURN_CHECK(mndUserRemoveTopic(pMnode, pTrans, pTopic->name));
@@ -614,6 +637,9 @@ END:
 }
 
 bool checkTopic(SArray *topics, char *topicName){
+  if (topics == NULL || topicName == NULL) {
+    return false;
+  }
   int32_t sz = taosArrayGetSize(topics);
   for (int32_t i = 0; i < sz; i++) {
     char *name = taosArrayGetP(topics, i);
@@ -625,6 +651,9 @@ bool checkTopic(SArray *topics, char *topicName){
 }
 
 static int32_t mndCheckConsumerByTopic(SMnode *pMnode, STrans *pTrans, char *topicName){
+  if (pMnode == NULL || pTrans == NULL || topicName == NULL) {
+    return TSDB_CODE_INVALID_MSG;
+  }
   int32_t         code = 0;
   SSdb           *pSdb    = pMnode->pSdb;
   void           *pIter = NULL;
@@ -653,6 +682,9 @@ END:
 }
 
 static int32_t mndDropCheckInfoByTopic(SMnode *pMnode, STrans *pTrans, SMqTopicObj *pTopic){
+  if (pMnode == NULL || pTrans == NULL || pTopic == NULL) {
+    return TSDB_CODE_INVALID_MSG;
+  }
   // broadcast to all vnode
   void   *pIter = NULL;
   SVgObj *pVgroup = NULL;
@@ -693,6 +725,9 @@ END:
 }
 
 static int32_t mndProcessDropTopicReq(SRpcMsg *pReq) {
+  if (pReq == NULL) {
+    return TSDB_CODE_INVALID_MSG;
+  }
   SMnode        *pMnode  = pReq->info.node;
   SMDropTopicReq dropReq = {0};
   int32_t        code = 0;
@@ -756,6 +791,9 @@ END:
 }
 
 int32_t mndGetNumOfTopics(SMnode *pMnode, char *dbName, int32_t *pNumOfTopics) {
+  if (pMnode == NULL || dbName == NULL || pNumOfTopics == NULL) {
+    return TSDB_CODE_INVALID_MSG;
+  }
   *pNumOfTopics = 0;
 
   SSdb   *pSdb = pMnode->pSdb;
@@ -786,6 +824,9 @@ int32_t mndGetNumOfTopics(SMnode *pMnode, char *dbName, int32_t *pNumOfTopics) {
 }
 
 static void schemaToJson(SSchema *schema, int32_t nCols, char *schemaJson){
+  if (schema == NULL || schemaJson == NULL) {
+    return;
+  }
   char*   string = NULL;
   int32_t code = 0;
   cJSON* columns = cJSON_CreateArray();
@@ -838,6 +879,9 @@ END:
 }
 
 static int32_t mndRetrieveTopic(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rowsCapacity) {
+  if (pReq == NULL || pShow == NULL || pBlock == NULL) {
+    return TSDB_CODE_INVALID_MSG;
+  }
   SMnode      *pMnode = pReq->info.node;
   SSdb        *pSdb = pMnode->pSdb;
   int32_t      numOfRows = 0;
@@ -945,11 +989,15 @@ END:
 }
 
 static void mndCancelGetNextTopic(SMnode *pMnode, void *pIter) {
+  if (pMnode == NULL) return;
   SSdb *pSdb = pMnode->pSdb;
   sdbCancelFetchByType(pSdb, pIter, SDB_TOPIC);
 }
 
 bool mndTopicExistsForDb(SMnode *pMnode, SDbObj *pDb) {
+  if (pMnode == NULL || pDb == NULL) {
+    return false;
+  }
   SSdb        *pSdb = pMnode->pSdb;
   void        *pIter = NULL;
   SMqTopicObj *pTopic = NULL;

source/dnode/vnode/src/inc/tq.h

@@ -127,11 +127,11 @@ void    tqPushEmptyDataRsp(STqHandle* pHandle, int32_t vgId);
 int32_t tqMetaOpen(STQ* pTq);
 void    tqMetaClose(STQ* pTq);
 int32_t tqMetaSaveHandle(STQ* pTq, const char* key, const STqHandle* pHandle);
-int32_t tqMetaSaveInfo(STQ* pTq, TTB* ttb, const void* key, int32_t kLen, const void* value, int32_t vLen);
+int32_t tqMetaSaveInfo(STQ* pTq, TTB* ttb, const void* key, uint32_t kLen, const void* value, uint32_t vLen);
-int32_t tqMetaDeleteInfo(STQ* pTq, TTB* ttb, const void* key, int32_t kLen);
+int32_t tqMetaDeleteInfo(STQ* pTq, TTB* ttb, const void* key, uint32_t kLen);
 int32_t tqMetaCreateHandle(STQ* pTq, SMqRebVgReq* req, STqHandle* handle);
-int32_t tqMetaDecodeCheckInfo(STqCheckInfo *info, void *pVal, int32_t vLen);
+int32_t tqMetaDecodeCheckInfo(STqCheckInfo *info, void *pVal, uint32_t vLen);
-int32_t tqMetaDecodeOffsetInfo(STqOffset *info, void *pVal, int32_t vLen);
+int32_t tqMetaDecodeOffsetInfo(STqOffset *info, void *pVal, uint32_t vLen);
 int32_t tqMetaSaveOffset(STQ* pTq, STqOffset* pOffset);
 int32_t tqMetaGetHandle(STQ* pTq, const char* key, STqHandle** pHandle);
 int32_t tqMetaGetOffset(STQ* pTq, const char* subkey, STqOffset** pOffset);

source/dnode/vnode/src/tq/tq.c

@@ -25,11 +25,12 @@
 // 2: wait to be inited or cleanup
 static int32_t tqInitialize(STQ* pTq);
 
-static FORCE_INLINE bool tqIsHandleExec(STqHandle* pHandle) { return TMQ_HANDLE_STATUS_EXEC == pHandle->status; }
+static FORCE_INLINE bool tqIsHandleExec(STqHandle* pHandle) { return pHandle != NULL ? TMQ_HANDLE_STATUS_EXEC == pHandle->status : true; }
-static FORCE_INLINE void tqSetHandleExec(STqHandle* pHandle) { pHandle->status = TMQ_HANDLE_STATUS_EXEC; }
+static FORCE_INLINE void tqSetHandleExec(STqHandle* pHandle) { if (pHandle != NULL) pHandle->status = TMQ_HANDLE_STATUS_EXEC; }
-static FORCE_INLINE void tqSetHandleIdle(STqHandle* pHandle) { pHandle->status = TMQ_HANDLE_STATUS_IDLE; }
+static FORCE_INLINE void tqSetHandleIdle(STqHandle* pHandle) { if (pHandle != NULL) pHandle->status = TMQ_HANDLE_STATUS_IDLE; }
 
 void tqDestroyTqHandle(void* data) {
+  if (data == NULL) return;
   STqHandle* pData = (STqHandle*)data;
   qDestroyTask(pData->execHandle.task);
 
@@ -59,11 +60,17 @@ void tqDestroyTqHandle(void* data) {
 }
 
 static bool tqOffsetEqual(const STqOffset* pLeft, const STqOffset* pRight) {
+  if (pLeft == NULL || pRight == NULL) {
+    return false;
+  }
   return pLeft->val.type == TMQ_OFFSET__LOG && pRight->val.type == TMQ_OFFSET__LOG &&
          pLeft->val.version == pRight->val.version;
 }
 
 int32_t tqOpen(const char* path, SVnode* pVnode) {
+  if (path == NULL || pVnode == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   STQ* pTq = taosMemoryCalloc(1, sizeof(STQ));
   if (pTq == NULL) {
     return terrno;
@@ -104,6 +111,9 @@ int32_t tqOpen(const char* path, SVnode* pVnode) {
 }
 
 int32_t tqInitialize(STQ* pTq) {
+  if (pTq == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t vgId = TD_VID(pTq->pVnode);
   int32_t code = streamMetaOpen(pTq->path, pTq, tqBuildStreamTask, tqExpandStreamTask, vgId, -1,
                                 tqStartTaskCompleteCallback, &pTq->pStreamMeta);
@@ -157,6 +167,9 @@ void tqNotifyClose(STQ* pTq) {
 }
 
 void tqPushEmptyDataRsp(STqHandle* pHandle, int32_t vgId) {
+  if (pHandle == NULL) {
+    return;
+  }
   int32_t    code = 0;
   SMqPollReq req = {0};
   code = tDeserializeSMqPollReq(pHandle->msg->pCont, pHandle->msg->contLen, &req);
@@ -186,6 +199,9 @@ void tqPushEmptyDataRsp(STqHandle* pHandle, int32_t vgId) {
 
 int32_t tqSendDataRsp(STqHandle* pHandle, const SRpcMsg* pMsg, const SMqPollReq* pReq, const SMqDataRsp* pRsp, int32_t type,
                       int32_t vgId) {
+  if (pHandle == NULL || pMsg == NULL || pReq == NULL || pRsp == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int64_t sver = 0, ever = 0;
   walReaderValidVersionRange(pHandle->execHandle.pTqReader->pWalReader, &sver, &ever);
 
@@ -201,6 +217,9 @@ int32_t tqSendDataRsp(STqHandle* pHandle, const SRpcMsg* pMsg, const SMqPollReq*
 }
 
 int32_t tqProcessOffsetCommitReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {
+  if (pTq == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   SMqVgOffset vgOffset = {0};
   int32_t     vgId = TD_VID(pTq->pVnode);
 
@@ -243,7 +262,7 @@ int32_t tqProcessOffsetCommitReq(STQ* pTq, int64_t sversion, char* msg, int32_t
   }
 
   if (tqMetaSaveInfo(pTq, pTq->pOffsetStore, pOffset->subKey, strlen(pOffset->subKey), msg,
-                     msgLen - sizeof(vgOffset.consumerId)) < 0) {
+                     msgLen >= sizeof(vgOffset.consumerId) ? msgLen - sizeof(vgOffset.consumerId) : 0) < 0) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return -1;
   }
@@ -255,6 +274,9 @@ end:
 }
 
 int32_t tqProcessSeekReq(STQ* pTq, SRpcMsg* pMsg) {
+  if (pTq == NULL || pMsg == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   SMqSeekReq req = {0};
   int32_t    vgId = TD_VID(pTq->pVnode);
   SRpcMsg    rsp = {.info = pMsg->info};
@@ -297,6 +319,9 @@ end:
 }
 
 int32_t tqCheckColModifiable(STQ* pTq, int64_t tbUid, int32_t colId) {
+  if (pTq == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   void* pIter = NULL;
 
   while (1) {
@@ -327,6 +352,9 @@ int32_t tqCheckColModifiable(STQ* pTq, int64_t tbUid, int32_t colId) {
 }
 
 int32_t tqProcessPollPush(STQ* pTq, SRpcMsg* pMsg) {
+  if (pTq == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t vgId = TD_VID(pTq->pVnode);
   taosWLockLatch(&pTq->lock);
   if (taosHashGetSize(pTq->pPushMgr) > 0) {
@@ -362,6 +390,9 @@ int32_t tqProcessPollPush(STQ* pTq, SRpcMsg* pMsg) {
 }
 
 int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
+  if (pTq == NULL || pMsg == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   SMqPollReq req = {0};
   int        code = tDeserializeSMqPollReq(pMsg->pCont, pMsg->contLen, &req);
   if (code < 0) {
@@ -439,6 +470,9 @@ END:
 }
 
 int32_t tqProcessVgCommittedInfoReq(STQ* pTq, SRpcMsg* pMsg) {
+  if (pTq == NULL || pMsg == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   void*   data = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
   int32_t len = pMsg->contLen - sizeof(SMsgHead);
 
@@ -485,6 +519,9 @@ int32_t tqProcessVgCommittedInfoReq(STQ* pTq, SRpcMsg* pMsg) {
 }
 
 int32_t tqProcessVgWalInfoReq(STQ* pTq, SRpcMsg* pMsg) {
+  if (pTq == NULL || pMsg == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t    code = 0;
   SMqPollReq req = {0};
   if (tDeserializeSMqPollReq(pMsg->pCont, pMsg->contLen, &req) < 0) {
@@ -570,6 +607,9 @@ END:
 }
 
 int32_t tqProcessDeleteSubReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {
+  if (pTq == NULL || msg == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   SMqVDeleteReq* pReq = (SMqVDeleteReq*)msg;
   int32_t        vgId = TD_VID(pTq->pVnode);
 
@@ -616,8 +656,11 @@ int32_t tqProcessDeleteSubReq(STQ* pTq, int64_t sversion, char* msg, int32_t msg
 }
 
 int32_t tqProcessAddCheckInfoReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {
+  if (pTq == NULL || msg == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   STqCheckInfo info = {0};
-  int32_t      code = tqMetaDecodeCheckInfo(&info, msg, msgLen);
+  int32_t      code = tqMetaDecodeCheckInfo(&info, msg, msgLen >= 0 ? msgLen : 0);
   if (code != 0) {
     return code;
   }
@@ -628,10 +671,13 @@ int32_t tqProcessAddCheckInfoReq(STQ* pTq, int64_t sversion, char* msg, int32_t
     return code;
   }
 
-  return tqMetaSaveInfo(pTq, pTq->pCheckStore, info.topic, strlen(info.topic), msg, msgLen);
+  return tqMetaSaveInfo(pTq, pTq->pCheckStore, info.topic, strlen(info.topic), msg, msgLen >= 0 ? msgLen : 0);
 }
 
 int32_t tqProcessDelCheckInfoReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {
+  if (pTq == NULL || msg == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   if (taosHashRemove(pTq->pCheckInfo, msg, strlen(msg)) < 0) {
     return TSDB_CODE_TSC_INTERNAL_ERROR;
   }
@@ -639,6 +685,9 @@ int32_t tqProcessDelCheckInfoReq(STQ* pTq, int64_t sversion, char* msg, int32_t
 }
 
 int32_t tqProcessSubscribeReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {
+  if (pTq == NULL || msg == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int         ret = 0;
   SMqRebVgReq req = {0};
   SDecoder    dc = {0};

source/dnode/vnode/src/tq/tqMeta.c

@@ -16,6 +16,9 @@
 #include "tq.h"
 
 int32_t tEncodeSTqHandle(SEncoder* pEncoder, const STqHandle* pHandle) {
+  if (pEncoder == NULL || pHandle == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = 0;
   int32_t lino;
 
@@ -54,6 +57,9 @@ _exit:
 }
 
 int32_t tDecodeSTqHandle(SDecoder* pDecoder, STqHandle* pHandle) {
+  if (pDecoder == NULL || pHandle == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = 0;
   int32_t lino;
 
@@ -91,7 +97,10 @@ _exit:
   return code;
 }
 
-int32_t tqMetaDecodeCheckInfo(STqCheckInfo* info, void* pVal, int32_t vLen) {
+int32_t tqMetaDecodeCheckInfo(STqCheckInfo* info, void* pVal, uint32_t vLen) {
+  if (info == NULL || pVal == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   SDecoder decoder = {0};
   tDecoderInit(&decoder, (uint8_t*)pVal, vLen);
   int32_t code = tDecodeSTqCheckInfo(&decoder, info);
@@ -104,7 +113,10 @@ int32_t tqMetaDecodeCheckInfo(STqCheckInfo* info, void* pVal, int32_t vLen) {
   return code;
 }
 
-int32_t tqMetaDecodeOffsetInfo(STqOffset* info, void* pVal, int32_t vLen) {
+int32_t tqMetaDecodeOffsetInfo(STqOffset* info, void* pVal, uint32_t vLen) {
+  if (info == NULL || pVal == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   SDecoder decoder = {0};
   tDecoderInit(&decoder, (uint8_t*)pVal, vLen);
   int32_t code = tDecodeSTqOffset(&decoder, info);
@@ -118,9 +130,12 @@ int32_t tqMetaDecodeOffsetInfo(STqOffset* info, void* pVal, int32_t vLen) {
 }
 
 int32_t tqMetaSaveOffset(STQ* pTq, STqOffset* pOffset) {
+  if (pTq == NULL || pOffset == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   void*    buf = NULL;
   int32_t  code = TDB_CODE_SUCCESS;
-  int32_t  vlen;
+  uint32_t  vlen;
   SEncoder encoder = {0};
   tEncodeSize(tEncodeSTqOffset, pOffset, vlen, code);
   if (code < 0) {
@@ -147,7 +162,10 @@ END:
   return code;
 }
 
-int32_t tqMetaSaveInfo(STQ* pTq, TTB* ttb, const void* key, int32_t kLen, const void* value, int32_t vLen) {
+int32_t tqMetaSaveInfo(STQ* pTq, TTB* ttb, const void* key, uint32_t kLen, const void* value, uint32_t vLen) {
+  if (pTq == NULL || ttb == NULL || key == NULL || value == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = TDB_CODE_SUCCESS;
   TXN*    txn = NULL;
 
@@ -164,7 +182,10 @@ END:
   return code;
 }
 
-int32_t tqMetaDeleteInfo(STQ* pTq, TTB* ttb, const void* key, int32_t kLen) {
+int32_t tqMetaDeleteInfo(STQ* pTq, TTB* ttb, const void* key, uint32_t kLen) {
+  if (pTq == NULL || ttb == NULL || key == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = TDB_CODE_SUCCESS;
   TXN*    txn = NULL;
 
@@ -182,6 +203,9 @@ END:
 }
 
 int32_t tqMetaGetOffset(STQ* pTq, const char* subkey, STqOffset** pOffset) {
+  if (pTq == NULL || subkey == NULL || pOffset == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   void* data = taosHashGet(pTq->pOffset, subkey, strlen(subkey));
   if (data == NULL) {
     int vLen = 0;
@@ -191,7 +215,7 @@ int32_t tqMetaGetOffset(STQ* pTq, const char* subkey, STqOffset** pOffset) {
     }
 
     STqOffset offset = {0};
-    if (tqMetaDecodeOffsetInfo(&offset, data, vLen) != TDB_CODE_SUCCESS) {
+    if (tqMetaDecodeOffsetInfo(&offset, data, vLen >= 0 ? vLen : 0) != TDB_CODE_SUCCESS) {
       tdbFree(data);
       return TSDB_CODE_OUT_OF_MEMORY;
     }
@@ -214,8 +238,11 @@ int32_t tqMetaGetOffset(STQ* pTq, const char* subkey, STqOffset** pOffset) {
 }
 
 int32_t tqMetaSaveHandle(STQ* pTq, const char* key, const STqHandle* pHandle) {
+  if (pTq == NULL || key == NULL || pHandle == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t  code = TDB_CODE_SUCCESS;
-  int32_t  vlen;
+  uint32_t  vlen;
   void*    buf = NULL;
   SEncoder encoder = {0};
   tEncodeSize(tEncodeSTqHandle, pHandle, vlen, code);
@@ -238,7 +265,7 @@ int32_t tqMetaSaveHandle(STQ* pTq, const char* key, const STqHandle* pHandle) {
     goto END;
   }
 
-  TQ_ERR_GO_TO_END(tqMetaSaveInfo(pTq, pTq->pExecStore, key, (int)strlen(key), buf, vlen));
+  TQ_ERR_GO_TO_END(tqMetaSaveInfo(pTq, pTq->pExecStore, key, strlen(key), buf, vlen));
 
 END:
   tEncoderClear(&encoder);
@@ -247,6 +274,9 @@ END:
 }
 
 static int tqMetaInitHandle(STQ* pTq, STqHandle* handle) {
+  if (pTq == NULL || handle == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = TDB_CODE_SUCCESS;
 
   SVnode* pVnode = pTq->pVnode;
@@ -318,7 +348,10 @@ END:
   return code;
 }
 
-static int32_t tqMetaRestoreHandle(STQ* pTq, void* pVal, int vLen, STqHandle* handle) {
+static int32_t tqMetaRestoreHandle(STQ* pTq, void* pVal, uint32_t vLen, STqHandle* handle) {
+  if (pTq == NULL || pVal == NULL || handle == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t  vgId = TD_VID(pTq->pVnode);
   SDecoder decoder = {0};
   int32_t  code = TDB_CODE_SUCCESS;
@@ -335,6 +368,9 @@ END:
 }
 
 int32_t tqMetaCreateHandle(STQ* pTq, SMqRebVgReq* req, STqHandle* handle) {
+  if (pTq == NULL || req == NULL || handle == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t vgId = TD_VID(pTq->pVnode);
 
   (void)memcpy(handle->subKey, req->subKey, TSDB_SUBSCRIBE_KEY_LEN);
@@ -375,6 +411,9 @@ int32_t tqMetaCreateHandle(STQ* pTq, SMqRebVgReq* req, STqHandle* handle) {
 }
 
 static int32_t tqMetaTransformInfo(TDB* pMetaDB, TTB* pOld, TTB* pNew) {
+  if (pMetaDB == NULL || pOld == NULL || pNew == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   TBC*  pCur = NULL;
   void* pKey = NULL;
   int   kLen = 0;
@@ -404,6 +443,9 @@ END:
 }
 
 int32_t tqMetaGetHandle(STQ* pTq, const char* key, STqHandle** pHandle) {
+  if (pTq == NULL || key == NULL || pHandle == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   void* data = taosHashGet(pTq->pHandle, key, strlen(key));
   if (data == NULL) {
     int vLen = 0;
@@ -412,7 +454,7 @@ int32_t tqMetaGetHandle(STQ* pTq, const char* key, STqHandle** pHandle) {
       return TSDB_CODE_MND_SUBSCRIBE_NOT_EXIST;
     }
     STqHandle handle = {0};
-    if (tqMetaRestoreHandle(pTq, data, vLen, &handle) != 0) {
+    if (tqMetaRestoreHandle(pTq, data, vLen >= 0 ? vLen : 0, &handle) != 0) {
       tdbFree(data);
       tqDestroyTqHandle(&handle);
       return TSDB_CODE_OUT_OF_MEMORY;
@@ -429,6 +471,9 @@ int32_t tqMetaGetHandle(STQ* pTq, const char* key, STqHandle** pHandle) {
 }
 
 int32_t tqMetaOpenTdb(STQ* pTq) {
+  if (pTq == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = TDB_CODE_SUCCESS;
   TQ_ERR_GO_TO_END(tdbOpen(pTq->path, 16 * 1024, 1, &pTq->pMetaDB, 0, 0, NULL));
   TQ_ERR_GO_TO_END(tdbTbOpen("tq.db", -1, -1, NULL, pTq->pMetaDB, &pTq->pExecStore, 0));
@@ -440,6 +485,9 @@ END:
 }
 
 static int32_t replaceTqPath(char** path) {
+  if (path == NULL || *path == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   char*   tpath = NULL;
   int32_t code = tqBuildFName(&tpath, *path, TQ_SUBSCRIBE_NAME);
   if (code != 0) {
@@ -451,6 +499,9 @@ static int32_t replaceTqPath(char** path) {
 }
 
 static int32_t tqMetaRestoreCheckInfo(STQ* pTq) {
+  if (pTq == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   TBC*         pCur = NULL;
   void*        pKey = NULL;
   int          kLen = 0;
@@ -463,7 +514,7 @@ static int32_t tqMetaRestoreCheckInfo(STQ* pTq) {
   TQ_ERR_GO_TO_END(tdbTbcMoveToFirst(pCur));
 
   while (tdbTbcNext(pCur, &pKey, &kLen, &pVal, &vLen) == 0) {
-    TQ_ERR_GO_TO_END(tqMetaDecodeCheckInfo(&info, pVal, vLen));
+    TQ_ERR_GO_TO_END(tqMetaDecodeCheckInfo(&info, pVal, vLen >= 0 ? vLen : 0));
     TQ_ERR_GO_TO_END(taosHashPut(pTq->pCheckInfo, info.topic, strlen(info.topic), &info, sizeof(STqCheckInfo)));
   }
   info.colIdList = NULL;
@@ -477,6 +528,9 @@ END:
 }
 
 int32_t tqMetaOpen(STQ* pTq) {
+  if (pTq == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   char*   maindb = NULL;
   char*   offsetNew = NULL;
   int32_t code = TDB_CODE_SUCCESS;
@@ -504,6 +558,9 @@ END:
 }
 
 int32_t tqMetaTransform(STQ* pTq) {
+  if (pTq == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = TDB_CODE_SUCCESS;
   TDB*    pMetaDB = NULL;
   TTB*    pExecStore = NULL;
@@ -543,6 +600,9 @@ END:
 }
 
 void tqMetaClose(STQ* pTq) {
+  if (pTq == NULL) {
+    return;
+  }
   int32_t ret = 0;
   if (pTq->pExecStore) {
     tdbTbClose(pTq->pExecStore);

source/dnode/vnode/src/tq/tqOffset.c

@@ -17,6 +17,9 @@
 #include "tq.h"
 
 int32_t tqBuildFName(char** data, const char* path, char* name) {
+  if (data == NULL || path == NULL || name == NULL) {
+    return TSDB_CODE_INVALID_MSG;
+  }
   int32_t len = strlen(path) + strlen(name) + 2;
   char*   fname = taosMemoryCalloc(1, len);
   if(fname == NULL) {
@@ -33,6 +36,9 @@ int32_t tqBuildFName(char** data, const char* path, char* name) {
 }
 
 int32_t tqOffsetRestoreFromFile(STQ* pTq, char* name) {
+  if (pTq == NULL || name == NULL) {
+    return TSDB_CODE_INVALID_MSG;
+  }
   int32_t   code = TDB_CODE_SUCCESS;
   void*     pMemBuf = NULL;
 
@@ -54,6 +60,10 @@ int32_t tqOffsetRestoreFromFile(STQ* pTq, char* name) {
     }
     total += INT_BYTES;
     size = htonl(size);
+    if (size <= 0) {
+      code = TSDB_CODE_INVALID_MSG;
+      goto END;
+    }
     pMemBuf = taosMemoryCalloc(1, size);
     if (pMemBuf == NULL) {
       code = terrno;

source/dnode/vnode/src/tq/tqPush.c

@@ -17,6 +17,9 @@
 #include "vnd.h"
 
 int32_t tqProcessSubmitReqForSubscribe(STQ* pTq) {
+  if (pTq == NULL) {
+    return TSDB_CODE_INVALID_MSG;
+  }
   if (taosHashGetSize(pTq->pPushMgr) <= 0) {
     return 0;
   }
@@ -64,6 +67,9 @@ int32_t tqPushMsg(STQ* pTq, tmsg_t msgType) {
 }
 
 int32_t tqRegisterPushHandle(STQ* pTq, void* handle, SRpcMsg* pMsg) {
+  if (pTq == NULL || handle == NULL || pMsg == NULL) {
+    return TSDB_CODE_INVALID_MSG;
+  }
   int32_t    vgId = TD_VID(pTq->pVnode);
   STqHandle* pHandle = (STqHandle*)handle;
 
@@ -101,6 +107,9 @@ int32_t tqRegisterPushHandle(STQ* pTq, void* handle, SRpcMsg* pMsg) {
 }
 
 void tqUnregisterPushHandle(STQ* pTq, void *handle) {
+  if (pTq == NULL || handle == NULL) {
+    return;
+  }
   STqHandle *pHandle = (STqHandle*)handle;
   int32_t    vgId = TD_VID(pTq->pVnode);
 

source/dnode/vnode/src/tq/tqRead.c

@@ -17,6 +17,9 @@
 #include "tq.h"
 
 bool isValValidForTable(STqHandle* pHandle, SWalCont* pHead) {
+  if (pHandle == NULL || pHead == NULL) {
+    return false;
+  }
   if (pHandle->execHandle.subType != TOPIC_SUB_TYPE__TABLE) {
     return true;
   }
@@ -198,6 +201,9 @@ end:
 }
 
 int32_t tqFetchLog(STQ* pTq, STqHandle* pHandle, int64_t* fetchOffset, uint64_t reqId) {
+  if (pTq == NULL || pHandle == NULL || fetchOffset == NULL) {
+    return -1;
+  }
   int32_t code = -1;
   int32_t vgId = TD_VID(pTq->pVnode);
   int64_t id = pHandle->pWalReader->readerId;
@@ -259,9 +265,17 @@ END:
   return code;
 }
 
-bool tqGetTablePrimaryKey(STqReader* pReader) { return pReader->hasPrimaryKey; }
+bool tqGetTablePrimaryKey(STqReader* pReader) {
+  if (pReader == NULL) {
+    return false;
+  }
+  return pReader->hasPrimaryKey;
+}
 
 void tqSetTablePrimaryKey(STqReader* pReader, int64_t uid) {
+  if (pReader == NULL) {
+    return;
+  }
   bool            ret = false;
   SSchemaWrapper* schema = metaGetTableSchema(pReader->pVnodeMeta, uid, -1, 1, NULL);
   if (schema && schema->nCols >= 2 && schema->pSchema[1].flags & COL_IS_KEY) {
@@ -272,6 +286,9 @@ void tqSetTablePrimaryKey(STqReader* pReader, int64_t uid) {
 }
 
 STqReader* tqReaderOpen(SVnode* pVnode) {
+  if (pVnode == NULL) {
+    return NULL;
+  }
   STqReader* pReader = taosMemoryCalloc(1, sizeof(STqReader));
   if (pReader == NULL) {
     return NULL;
@@ -323,6 +340,9 @@ void tqReaderClose(STqReader* pReader) {
 }
 
 int32_t tqReaderSeek(STqReader* pReader, int64_t ver, const char* id) {
+  if (pReader == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   if (walReaderSeekVer(pReader->pWalReader, ver) < 0) {
     return -1;
   }
@@ -406,6 +426,9 @@ int32_t extractMsgFromWal(SWalReader* pReader, void** pItem, int64_t maxVer, con
 }
 
 bool tqNextBlockInWal(STqReader* pReader, const char* id, int sourceExcluded) {
+  if (pReader == NULL) {
+    return false;
+  }
   SWalReader* pWalReader = pReader->pWalReader;
 
   int64_t st = taosGetTimestampMs();
@@ -462,6 +485,9 @@ bool tqNextBlockInWal(STqReader* pReader, const char* id, int sourceExcluded) {
 }
 
 int32_t tqReaderSetSubmitMsg(STqReader* pReader, void* msgStr, int32_t msgLen, int64_t ver) {
+if (pReader == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   pReader->msg.msgStr = msgStr;
   pReader->msg.msgLen = msgLen;
   pReader->msg.ver = ver;
@@ -481,14 +507,29 @@ int32_t tqReaderSetSubmitMsg(STqReader* pReader, void* msgStr, int32_t msgLen, i
   return 0;
 }
 
-SWalReader* tqGetWalReader(STqReader* pReader) { return pReader->pWalReader; }
+SWalReader* tqGetWalReader(STqReader* pReader) {
+  if (pReader == NULL) {
+    return NULL;
+  }
+  return pReader->pWalReader;
+}
 
-SSDataBlock* tqGetResultBlock(STqReader* pReader) { return pReader->pResBlock; }
+SSDataBlock* tqGetResultBlock(STqReader* pReader) {
+  if (pReader == NULL) {
+    return NULL;
+  }
+  return pReader->pResBlock;
+}
 
-int64_t tqGetResultBlockTime(STqReader* pReader) { return pReader->lastTs; }
+int64_t tqGetResultBlockTime(STqReader* pReader) {
+  if (pReader == NULL) {
+    return 0;
+  }
+  return pReader->lastTs;
+}
 
 bool tqNextBlockImpl(STqReader* pReader, const char* idstr) {
-  if (pReader->msg.msgStr == NULL) {
+  if (pReader == NULL || pReader->msg.msgStr == NULL) {
     return false;
   }
 
@@ -525,7 +566,7 @@ bool tqNextBlockImpl(STqReader* pReader, const char* idstr) {
 }
 
 bool tqNextDataBlockFilterOut(STqReader* pReader, SHashObj* filterOutUids) {
-  if (pReader->msg.msgStr == NULL) return false;
+  if (pReader == NULL || pReader->msg.msgStr == NULL) return false;
 
   int32_t blockSz = taosArrayGetSize(pReader->submit.aSubmitTbData);
   while (pReader->nextBlk < blockSz) {
@@ -548,6 +589,9 @@ bool tqNextDataBlockFilterOut(STqReader* pReader, SHashObj* filterOutUids) {
 }
 
 int32_t tqMaskBlock(SSchemaWrapper* pDst, SSDataBlock* pBlock, const SSchemaWrapper* pSrc, char* mask) {
+  if (pDst == NULL || pBlock == NULL || pSrc == NULL || mask == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = 0;
 
   int32_t cnt = 0;
@@ -577,6 +621,9 @@ int32_t tqMaskBlock(SSchemaWrapper* pDst, SSDataBlock* pBlock, const SSchemaWrap
 }
 
 static int32_t buildResSDataBlock(STqReader* pReader, SSchemaWrapper* pSchema, const SArray* pColIdList) {
+  if (pReader == NULL || pSchema == NULL || pColIdList == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   SSDataBlock* pBlock = pReader->pResBlock;
   if (blockDataGetNumOfCols(pBlock) > 0) {
       blockDataDestroy(pBlock);
@@ -659,6 +706,9 @@ static int32_t doSetVal(SColumnInfoData* pColumnInfoData, int32_t rowIndex, SCol
 }
 
 int32_t tqRetrieveDataBlock(STqReader* pReader, SSDataBlock** pRes, const char* id) {
+  if (pReader == NULL || pRes == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   tqTrace("tq reader retrieve data block %p, index:%d", pReader->msg.msgStr, pReader->nextBlk);
   int32_t        code = 0;
   int32_t        line = 0;
@@ -825,6 +875,10 @@ END:
 static int32_t processBuildNew(STqReader* pReader, SSubmitTbData* pSubmitTbData, SArray* blocks, SArray* schemas,
                                SSchemaWrapper* pSchemaWrapper, char* assigned, int32_t numOfRows, int32_t curRow,
                                int32_t* lastRow) {
+  if (pReader == NULL || pSubmitTbData == NULL || blocks == NULL || schemas == NULL || pSchemaWrapper == NULL ||
+      assigned == NULL || lastRow == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t         code = 0;
   SSchemaWrapper* pSW = NULL;
   SSDataBlock*    block = NULL;
@@ -860,6 +914,9 @@ END:
   return code;
 }
 static int32_t tqProcessColData(STqReader* pReader, SSubmitTbData* pSubmitTbData, SArray* blocks, SArray* schemas) {
+  if (pReader == NULL || pSubmitTbData == NULL || blocks == NULL || schemas == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = 0;
   int32_t curRow = 0;
   int32_t lastRow = 0;
@@ -919,6 +976,9 @@ END:
 }
 
 int32_t tqProcessRowData(STqReader* pReader, SSubmitTbData* pSubmitTbData, SArray* blocks, SArray* schemas) {
+  if (pReader == NULL || pSubmitTbData == NULL || blocks == NULL || schemas == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t   code = 0;
   STSchema* pTSchema = NULL;
 
@@ -976,6 +1036,9 @@ END:
 }
 
 int32_t tqRetrieveTaosxBlock(STqReader* pReader, SArray* blocks, SArray* schemas, SSubmitTbData** pSubmitTbDataRet, int64_t *createTime) {
+  if (pReader == NULL || blocks == NULL || schemas == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   tqTrace("tq reader retrieve data block %p, %d", pReader->msg.msgStr, pReader->nextBlk);
   SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk);
   if (pSubmitTbData == NULL) {
@@ -1007,9 +1070,17 @@ int32_t tqRetrieveTaosxBlock(STqReader* pReader, SArray* blocks, SArray* schemas
   }
 }
 
-void tqReaderSetColIdList(STqReader* pReader, SArray* pColIdList) { pReader->pColIdList = pColIdList; }
+void tqReaderSetColIdList(STqReader* pReader, SArray* pColIdList) {
+  if (pReader == NULL){
+    return;
+  }
+  pReader->pColIdList = pColIdList;
+}
 
 void tqReaderSetTbUidList(STqReader* pReader, const SArray* tbUidList, const char* id) {
+  if (pReader == NULL || tbUidList == NULL) {
+    return;
+  }
   if (pReader->tbIdHash) {
     taosHashClear(pReader->tbIdHash);
   } else {
@@ -1032,6 +1103,9 @@ void tqReaderSetTbUidList(STqReader* pReader, const SArray* tbUidList, const cha
 }
 
 void tqReaderAddTbUidList(STqReader* pReader, const SArray* pTableUidList) {
+  if (pReader == NULL || pTableUidList == NULL) {
+    return;
+  }
   if (pReader->tbIdHash == NULL) {
     pReader->tbIdHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_ENTRY_LOCK);
     if (pReader->tbIdHash == NULL) {
@@ -1051,12 +1125,23 @@ void tqReaderAddTbUidList(STqReader* pReader, const SArray* pTableUidList) {
 }
 
 bool tqReaderIsQueriedTable(STqReader* pReader, uint64_t uid) {
-  return taosHashGet(pReader->tbIdHash, &uid, sizeof(uint64_t));
+  if (pReader == NULL) {
+    return false;
+  }
+  return taosHashGet(pReader->tbIdHash, &uid, sizeof(uint64_t)) != NULL;
 }
 
-bool tqCurrentBlockConsumed(const STqReader* pReader) { return pReader->msg.msgStr == NULL; }
+bool tqCurrentBlockConsumed(const STqReader* pReader) {
+  if (pReader == NULL) {
+    return false;
+  }
+  return pReader->msg.msgStr == NULL;
+}
 
 void tqReaderRemoveTbUidList(STqReader* pReader, const SArray* tbUidList) {
+  if (pReader == NULL || tbUidList == NULL) {
+    return;
+  }
   for (int32_t i = 0; i < taosArrayGetSize(tbUidList); i++) {
     int64_t* pKey = (int64_t*)taosArrayGet(tbUidList, i);
     if (pKey && taosHashRemove(pReader->tbIdHash, pKey, sizeof(int64_t)) != 0) {
@@ -1066,6 +1151,9 @@ void tqReaderRemoveTbUidList(STqReader* pReader, const SArray* tbUidList) {
 }
 
 int32_t tqUpdateTbUidList(STQ* pTq, const SArray* tbUidList, bool isAdd) {
+  if (pTq == NULL || tbUidList == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   void*   pIter = NULL;
   int32_t vgId = TD_VID(pTq->pVnode);
 

source/dnode/vnode/src/tq/tqScan.c

@@ -16,6 +16,9 @@
 #include "tq.h"
 
 int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, SMqDataRsp* pRsp, int32_t numOfCols, int8_t precision) {
+  if (pBlock == NULL || pRsp == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   size_t dataEncodeBufSize = blockGetEncodeSize(pBlock);
   int32_t dataStrLen = sizeof(SRetrieveTableRspForTmq) + dataEncodeBufSize;
   void*   buf = taosMemoryCalloc(1, dataStrLen);
@@ -47,18 +50,10 @@ int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, SMqDataRsp* pRsp, int32_t
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t tqAddBlockSchemaToRsp(const STqExecHandle* pExec, SMqDataRsp* pRsp) {
-  SSchemaWrapper* pSW = tCloneSSchemaWrapper(pExec->pTqReader->pSchemaWrapper);
-  if (pSW == NULL) {
-    return terrno;
-  }
-  if (taosArrayPush(pRsp->blockSchema, &pSW) == NULL) {
-    return terrno;
-  }
-  return 0;
-}
-
 static int32_t tqAddTbNameToRsp(const STQ* pTq, int64_t uid, SMqDataRsp* pRsp, int32_t n) {
+  if (pRsp == NULL || pTq == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   SMetaReader mr = {0};
   metaReaderDoInit(&mr, pTq->pVnode->pMeta, META_READER_LOCK);
 
@@ -84,6 +79,9 @@ static int32_t tqAddTbNameToRsp(const STQ* pTq, int64_t uid, SMqDataRsp* pRsp, i
 }
 
 int32_t getDataBlock(qTaskInfo_t task, const STqHandle* pHandle, int32_t vgId, SSDataBlock** res) {
+  if (task == NULL || pHandle == NULL || res == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   uint64_t ts = 0;
   qStreamSetOpen(task);
 
@@ -99,6 +97,9 @@ int32_t getDataBlock(qTaskInfo_t task, const STqHandle* pHandle, int32_t vgId, S
 }
 
 int32_t tqScanData(STQ* pTq, STqHandle* pHandle, SMqDataRsp* pRsp, STqOffsetVal* pOffset, const SMqPollReq* pRequest) {
+  if (pTq == NULL || pHandle == NULL || pRsp == NULL || pOffset == NULL || pRequest == NULL){
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t vgId = TD_VID(pTq->pVnode);
   int32_t code = 0;
   int32_t line = 0;
@@ -189,6 +190,9 @@ END:
 }
 
 int32_t tqScanTaosx(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, SMqBatchMetaRsp* pBatchMetaRsp, STqOffsetVal* pOffset) {
+  if (pTq == NULL || pHandle == NULL || pRsp == NULL || pBatchMetaRsp == NULL || pOffset == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   const STqExecHandle* pExec = &pHandle->execHandle;
   qTaskInfo_t          task = pExec->task;
   int code = qStreamPrepareScan(task, pOffset, pHandle->execHandle.subType);
@@ -280,6 +284,9 @@ int32_t tqScanTaosx(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, SMqBat
 }
 
 static int32_t buildCreateTbInfo(SMqDataRsp* pRsp, SVCreateTbReq* pCreateTbReq){
+  if (pRsp == NULL || pCreateTbReq == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t code = 0;
   void*   createReq = NULL;
   if (pRsp->createTableNum == 0) {
@@ -329,6 +336,9 @@ END:
 }
 
 static void tqProcessSubData(STQ* pTq, STqHandle* pHandle, SMqDataRsp* pRsp, int32_t* totalRows, int8_t sourceExcluded){
+  if (pTq == NULL || pHandle == NULL || pRsp == NULL || totalRows == NULL) {
+    return;
+  }
   int32_t code = 0;
   STqExecHandle* pExec = &pHandle->execHandle;
   STqReader* pReader = pExec->pTqReader;
@@ -407,6 +417,9 @@ END:
 
 int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, SMqDataRsp* pRsp, int32_t* totalRows,
                        int8_t sourceExcluded) {
+  if (pTq == NULL || pHandle == NULL || pRsp == NULL || totalRows == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   STqExecHandle* pExec = &pHandle->execHandle;
   int32_t        code = 0;
   STqReader* pReader = pExec->pTqReader;

source/dnode/vnode/src/tq/tqSnapshot.c

@@ -27,6 +27,9 @@ struct STqSnapReader {
 };
 
 int32_t tqSnapReaderOpen(STQ* pTq, int64_t sver, int64_t ever, int8_t type, STqSnapReader** ppReader) {
+  if (pTq == NULL || ppReader == NULL) {
+    return TSDB_CODE_INVALID_MSG;
+  }
   int32_t        code = 0;
   STqSnapReader* pReader = NULL;
 
@@ -77,12 +80,18 @@ _err:
 }
 
 void tqSnapReaderClose(STqSnapReader** ppReader) {
+  if (ppReader == NULL || *ppReader == NULL) {
+    return;
+  }
   tdbTbcClose((*ppReader)->pCur);
   taosMemoryFree(*ppReader);
   *ppReader = NULL;
 }
 
 int32_t tqSnapRead(STqSnapReader* pReader, uint8_t** ppData) {
+  if (pReader == NULL || ppData == NULL) {
+    return TSDB_CODE_INVALID_MSG;
+  }
   int32_t code = 0;
   void*   pKey = NULL;
   void*   pVal = NULL;
@@ -126,6 +135,9 @@ struct STqSnapWriter {
 };
 
 int32_t tqSnapWriterOpen(STQ* pTq, int64_t sver, int64_t ever, STqSnapWriter** ppWriter) {
+  if (pTq == NULL || ppWriter == NULL) {
+    return TSDB_CODE_INVALID_MSG;
+  }
   int32_t        code = 0;
   STqSnapWriter* pWriter = NULL;
 
@@ -156,6 +168,9 @@ _err:
 }
 
 int32_t tqSnapWriterClose(STqSnapWriter** ppWriter, int8_t rollback) {
+  if (ppWriter == NULL || *ppWriter == NULL) {
+    return TSDB_CODE_INVALID_MSG;
+  }
   int32_t        code = 0;
   STqSnapWriter* pWriter = *ppWriter;
   STQ*           pTq = pWriter->pTq;
@@ -180,6 +195,9 @@ _err:
 }
 
 int32_t tqSnapHandleWrite(STqSnapWriter* pWriter, uint8_t* pData, uint32_t nData) {
+  if (pWriter == NULL || pData == NULL || nData < sizeof(SSnapDataHdr)) {
+    return TSDB_CODE_INVALID_MSG;
+  }
   int32_t   code = 0;
   STQ*      pTq = pWriter->pTq;
   SDecoder  decoder = {0};
@@ -190,7 +208,7 @@ int32_t tqSnapHandleWrite(STqSnapWriter* pWriter, uint8_t* pData, uint32_t nData
   code = tDecodeSTqHandle(pDecoder, &handle);
   if (code) goto end;
   taosWLockLatch(&pTq->lock);
-  code = tqMetaSaveInfo(pTq, pTq->pExecStore, handle.subKey, (int)strlen(handle.subKey), pData + sizeof(SSnapDataHdr),
+  code = tqMetaSaveInfo(pTq, pTq->pExecStore, handle.subKey, strlen(handle.subKey), pData + sizeof(SSnapDataHdr),
                         nData - sizeof(SSnapDataHdr));
   taosWUnLockLatch(&pTq->lock);
 
@@ -202,6 +220,9 @@ end:
 }
 
 int32_t tqSnapCheckInfoWrite(STqSnapWriter* pWriter, uint8_t* pData, uint32_t nData) {
+  if (pWriter == NULL || pData == NULL || nData < sizeof(SSnapDataHdr)) {
+    return TSDB_CODE_INVALID_MSG;
+  }
   int32_t      code = 0;
   STQ*         pTq = pWriter->pTq;
   STqCheckInfo info = {0};
@@ -223,6 +244,9 @@ _err:
 }
 
 int32_t tqSnapOffsetWrite(STqSnapWriter* pWriter, uint8_t* pData, uint32_t nData) {
+  if (pWriter == NULL || pData == NULL || nData < sizeof(SSnapDataHdr)) {
+    return TSDB_CODE_INVALID_MSG;
+  }
   int32_t code = 0;
   STQ*    pTq = pWriter->pTq;
 

source/dnode/vnode/src/tq/tqUtil.c

@@ -21,6 +21,9 @@ static int32_t tqSendBatchMetaPollRsp(STqHandle* pHandle, const SRpcMsg* pMsg, c
                                       const SMqBatchMetaRsp* pRsp, int32_t vgId);
 
 int32_t tqInitDataRsp(SMqDataRsp* pRsp, STqOffsetVal pOffset) {
+  if (pRsp == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   pRsp->blockData = taosArrayInit(0, sizeof(void*));
   pRsp->blockDataLen = taosArrayInit(0, sizeof(int32_t));
 
@@ -41,6 +44,9 @@ void tqUpdateNodeStage(STQ* pTq, bool isLeader) {
 }
 
 static int32_t tqInitTaosxRsp(SMqDataRsp* pRsp, STqOffsetVal pOffset) {
+  if (pRsp == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   tOffsetCopy(&pRsp->reqOffset, &pOffset);
   tOffsetCopy(&pRsp->rspOffset, &pOffset);
 
@@ -81,6 +87,9 @@ static int32_t tqInitTaosxRsp(SMqDataRsp* pRsp, STqOffsetVal pOffset) {
 
 static int32_t extractResetOffsetVal(STqOffsetVal* pOffsetVal, STQ* pTq, STqHandle* pHandle, const SMqPollReq* pRequest,
                                      SRpcMsg* pMsg, bool* pBlockReturned) {
+  if (pOffsetVal == NULL || pTq == NULL || pHandle == NULL || pRequest == NULL || pMsg == NULL || pBlockReturned == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   uint64_t   consumerId = pRequest->consumerId;
   STqOffset* pOffset = NULL;
   int32_t    code = tqMetaGetOffset(pTq, pRequest->subKey, &pOffset);
@@ -142,6 +151,9 @@ static int32_t extractResetOffsetVal(STqOffsetVal* pOffsetVal, STQ* pTq, STqHand
 
 static int32_t extractDataAndRspForNormalSubscribe(STQ* pTq, STqHandle* pHandle, const SMqPollReq* pRequest,
                                                    SRpcMsg* pMsg, STqOffsetVal* pOffset) {
+  if (pTq == NULL || pHandle == NULL || pRequest == NULL || pMsg == NULL || pOffset == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   uint64_t consumerId = pRequest->consumerId;
   int32_t  vgId = TD_VID(pTq->pVnode);
   terrno = 0;
@@ -212,6 +224,9 @@ static void tDeleteCommon(void* parm) {}
 
 static int32_t extractDataAndRspForDbStbSubscribe(STQ* pTq, STqHandle* pHandle, const SMqPollReq* pRequest,
                                                   SRpcMsg* pMsg, STqOffsetVal* offset) {
+  if (pTq == NULL || pHandle == NULL || pRequest == NULL || pMsg == NULL || offset == NULL) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t         vgId = TD_VID(pTq->pVnode);
   SMqDataRsp      taosxRsp = {0};
   SMqBatchMetaRsp btMetaRsp = {0};
@@ -410,6 +425,9 @@ END:
 }
 
 int32_t tqExtractDataForMq(STQ* pTq, STqHandle* pHandle, const SMqPollReq* pRequest, SRpcMsg* pMsg) {
+  if (pTq == NULL || pHandle == NULL || pRequest == NULL || pMsg == NULL) {
+    return TSDB_CODE_TMQ_INVALID_MSG;
+  }
   int32_t      code = 0;
   STqOffsetVal reqOffset = {0};
   tOffsetCopy(&reqOffset, &pRequest->reqOffset);
@@ -445,6 +463,9 @@ END:
 
 static void initMqRspHead(SMqRspHead* pMsgHead, int32_t type, int32_t epoch, int64_t consumerId, int64_t sver,
                           int64_t ever) {
+  if (pMsgHead == NULL) {
+    return;
+  }
   pMsgHead->consumerId = consumerId;
   pMsgHead->epoch = epoch;
   pMsgHead->mqMsgType = type;
@@ -454,6 +475,9 @@ static void initMqRspHead(SMqRspHead* pMsgHead, int32_t type, int32_t epoch, int
 
 int32_t tqSendBatchMetaPollRsp(STqHandle* pHandle, const SRpcMsg* pMsg, const SMqPollReq* pReq,
                                const SMqBatchMetaRsp* pRsp, int32_t vgId) {
+  if (pHandle == NULL || pMsg == NULL || pReq == NULL || pRsp == NULL) {
+    return TSDB_CODE_TMQ_INVALID_MSG;
+  }
   int32_t len = 0;
   int32_t code = 0;
   tEncodeSize(tEncodeMqBatchMetaRsp, pRsp, len, code);
@@ -491,6 +515,9 @@ int32_t tqSendBatchMetaPollRsp(STqHandle* pHandle, const SRpcMsg* pMsg, const SM
 
 int32_t tqSendMetaPollRsp(STqHandle* pHandle, const SRpcMsg* pMsg, const SMqPollReq* pReq, const SMqMetaRsp* pRsp,
                           int32_t vgId) {
+  if (pHandle == NULL || pMsg == NULL || pReq == NULL || pRsp == NULL) {
+    return TSDB_CODE_TMQ_INVALID_MSG;
+  }
   int32_t len = 0;
   int32_t code = 0;
   tEncodeSize(tEncodeMqMetaRsp, pRsp, len, code);
@@ -529,6 +556,9 @@ int32_t tqSendMetaPollRsp(STqHandle* pHandle, const SRpcMsg* pMsg, const SMqPoll
 
 int32_t tqDoSendDataRsp(const SRpcHandleInfo* pRpcHandleInfo, const SMqDataRsp* pRsp, int32_t epoch, int64_t consumerId,
                         int32_t type, int64_t sver, int64_t ever) {
+  if (pRpcHandleInfo == NULL || pRsp == NULL) {
+    return TSDB_CODE_TMQ_INVALID_MSG;
+  }
   int32_t len = 0;
   int32_t code = 0;