diff --git a/source/client/src/clientRawBlockWrite.c b/source/client/src/clientRawBlockWrite.c index a77567dca3dedb06c969d74d68ffe49616d293b7..514f99d463fefc206fde683d83d3fd5354ee22a1 100644 --- a/source/client/src/clientRawBlockWrite.c +++ b/source/client/src/clientRawBlockWrite.c @@ -219,8 +219,6 @@ static char* processAlterStb(SMqMetaRsp* metaRsp) { goto _err; } string = buildAlterSTableJson(req.alterOriData, req.alterOriDataLen); - tDecoderClear(&coder); - return string; _err: tDecoderClear(&coder); @@ -397,6 +395,10 @@ static char* processAutoCreateTable(STaosxRsp* rsp) { _exit: for(int i = 0; i < rsp->createTableNum; i++){ tDecoderClear(&decoder[i]); + taosMemoryFreeClear(pCreateReq[i].comment); + if (pCreateReq[i].type == TSDB_CHILD_TABLE) { + taosArrayDestroy(pCreateReq[i].ctb.tagName); + } } taosMemoryFree(decoder); taosMemoryFree(pCreateReq); @@ -512,6 +514,7 @@ static char* processAlterTable(SMqMetaRsp* metaRsp) { string = cJSON_PrintUnformatted(json); _exit: + cJSON_Delete(json); tDecoderClear(&decoder); return string; } @@ -543,6 +546,7 @@ static char* processDropSTable(SMqMetaRsp* metaRsp) { string = cJSON_PrintUnformatted(json); _exit: + cJSON_Delete(json); tDecoderClear(&decoder); return string; } @@ -583,6 +587,7 @@ static char* processDropTable(SMqMetaRsp* metaRsp) { string = cJSON_PrintUnformatted(json); _exit: + cJSON_Delete(json); tDecoderClear(&decoder); return string; } @@ -688,6 +693,7 @@ static int32_t taosDropStb(TAOS* taos, void* meta, int32_t metaLen) { goto end; } + pRequest->syncQuery = true; if (!pRequest->pDb) { code = TSDB_CODE_PAR_DB_NOT_SPECIFIED; goto end; @@ -920,7 +926,7 @@ static int32_t taosDropTable(TAOS* taos, void* meta, int32_t metaLen) { if (code != TSDB_CODE_SUCCESS) { goto end; } - + pRequest->syncQuery = true; if (!pRequest->pDb) { code = TSDB_CODE_PAR_DB_NOT_SPECIFIED; goto end; @@ -1093,6 +1099,7 @@ static int32_t taosAlterTable(TAOS* taos, void* meta, int32_t metaLen) { goto end; } + pRequest->syncQuery = true; if (!pRequest->pDb) { code = TSDB_CODE_PAR_DB_NOT_SPECIFIED; goto end; @@ -1212,6 +1219,7 @@ int taos_write_raw_block(TAOS* taos, int rows, char* pData, const char* tbname) goto end; } + pRequest->syncQuery = true; if (!pRequest->pDb) { uError("WriteRaw:not use db"); code = TSDB_CODE_PAR_DB_NOT_SPECIFIED; @@ -1399,6 +1407,7 @@ static int32_t tmqWriteRawDataImpl(TAOS* taos, void* data, int32_t dataLen) { return terrno; } + pRequest->syncQuery = true; rspObj.resIter = -1; rspObj.resType = RES_TYPE__TMQ; @@ -1664,6 +1673,7 @@ static int32_t tmqWriteRawMetaDataImpl(TAOS* taos, void* data, int32_t dataLen) return terrno; } + pRequest->syncQuery = true; rspObj.resIter = -1; rspObj.resType = RES_TYPE__TMQ_METADATA; @@ -1739,23 +1749,29 @@ static int32_t tmqWriteRawMetaDataImpl(TAOS* taos, void* data, int32_t dataLen) int32_t* lenTmp = taosArrayGet(rspObj.rsp.createTableLen, j); SDecoder decoderTmp = {0}; - SVCreateTbReq* pCreateReq; + SVCreateTbReq pCreateReq = {0}; tDecoderInit(&decoderTmp, *dataTmp, *lenTmp); - if (tDecodeSVCreateTbReq(&decoderTmp, pCreateReq) < 0) { + if (tDecodeSVCreateTbReq(&decoderTmp, &pCreateReq) < 0) { tDecoderClear(&decoderTmp); + taosMemoryFreeClear(pCreateReq.comment); + taosArrayDestroy(pCreateReq.ctb.tagName); goto end; } - ASSERT (pCreateReq->type == TSDB_CHILD_TABLE); - if(strcmp(tbName, pCreateReq->name) == 0){ + ASSERT (pCreateReq.type == TSDB_CHILD_TABLE); + if(strcmp(tbName, pCreateReq.name) == 0){ schemaLen = *lenTmp; schemaData = *dataTmp; - strcpy(pName.tname, pCreateReq->ctb.name); + strcpy(pName.tname, pCreateReq.ctb.name); tDecoderClear(&decoderTmp); + taosMemoryFreeClear(pCreateReq.comment); + taosArrayDestroy(pCreateReq.ctb.tagName); break; } tDecoderClear(&decoderTmp); + taosMemoryFreeClear(pCreateReq.comment); + taosArrayDestroy(pCreateReq.ctb.tagName); } code = catalogGetTableMeta(pCatalog, &conn, &pName, &pTableMeta); @@ -1884,6 +1900,8 @@ static int32_t tmqWriteRawMetaDataImpl(TAOS* taos, void* data, int32_t dataLen) subReq->length += sizeof(SSubmitBlk) + schemaLen + totalLen; subReq->numOfBlocks++; taosMemoryFreeClear(pTableMeta); + rspObj.resInfo.pRspMsg = NULL; + doFreeReqResultInfo(&rspObj.resInfo); } pQuery = (SQuery*)nodesMakeNode(QUERY_NODE_QUERY); diff --git a/source/client/src/clientTmq.c b/source/client/src/clientTmq.c index 047191ec27677141a9258343cec4681124f969de..c67556faa07b8f42d216b963f99019a1b8735622 100644 --- a/source/client/src/clientTmq.c +++ b/source/client/src/clientTmq.c @@ -1816,7 +1816,7 @@ int32_t tmq_get_vgroup_id(TAOS_RES* res) { } else if (TD_RES_TMQ_META(res)) { SMqMetaRspObj* pMetaRspObj = (SMqMetaRspObj*)res; return pMetaRspObj->vgId; - } else if (TD_RES_TMQ_META(res)) { + } else if (TD_RES_TMQ_METADATA(res)) { SMqTaosxRspObj* pRspObj = (SMqTaosxRspObj*)res; return pRspObj->vgId; } else { @@ -1832,7 +1832,14 @@ const char* tmq_get_table_name(TAOS_RES* res) { return NULL; } return (const char*)taosArrayGetP(pRspObj->rsp.blockTbName, pRspObj->resIter); - } + } else if (TD_RES_TMQ_METADATA(res)) { + SMqTaosxRspObj* pRspObj = (SMqTaosxRspObj*)res; + if (!pRspObj->rsp.withTbName || pRspObj->rsp.blockTbName == NULL || pRspObj->resIter < 0 || + pRspObj->resIter >= pRspObj->rsp.blockNum) { + return NULL; + } + return (const char*)taosArrayGetP(pRspObj->rsp.blockTbName, pRspObj->resIter); + } return NULL; } diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c index ea25094d10863ac1e61146b15076d93cf6ad7fcd..111c8da8f62f4df348f1e9e6bf5c1d84863f62ac 100644 --- a/source/common/src/tmsg.c +++ b/source/common/src/tmsg.c @@ -6007,12 +6007,18 @@ int32_t tDecodeSTaosxRsp(SDecoder *pDecoder, STaosxRsp *pRsp) { void tDeleteSTaosxRsp(STaosxRsp *pRsp) { taosArrayDestroy(pRsp->blockDataLen); + pRsp->blockDataLen = NULL; taosArrayDestroyP(pRsp->blockData, (FDelete)taosMemoryFree); + pRsp->blockData = NULL; taosArrayDestroyP(pRsp->blockSchema, (FDelete)tDeleteSSchemaWrapper); + pRsp->blockSchema = NULL; taosArrayDestroyP(pRsp->blockTbName, (FDelete)taosMemoryFree); + pRsp->blockTbName = NULL; taosArrayDestroy(pRsp->createTableLen); + pRsp->createTableLen = NULL; taosArrayDestroyP(pRsp->createTableReq, (FDelete)taosMemoryFree); + pRsp->createTableReq = NULL; } int32_t tEncodeSSingleDeleteReq(SEncoder *pEncoder, const SSingleDeleteReq *pReq) { diff --git a/source/dnode/mnode/impl/inc/mndDef.h b/source/dnode/mnode/impl/inc/mndDef.h index 9632be1b24ba4561804a1f7ec0e7bf1205f23e4c..4625cbfb096d6df86c60962a59b579cfa0dc43bc 100644 --- a/source/dnode/mnode/impl/inc/mndDef.h +++ b/source/dnode/mnode/impl/inc/mndDef.h @@ -539,7 +539,7 @@ typedef struct { } SMqConsumerEp; SMqConsumerEp* tCloneSMqConsumerEp(const SMqConsumerEp* pEp); -void tDeleteSMqConsumerEp(SMqConsumerEp* pEp); +void tDeleteSMqConsumerEp(void* pEp); int32_t tEncodeSMqConsumerEp(void** buf, const SMqConsumerEp* pEp); void* tDecodeSMqConsumerEp(const void* buf, SMqConsumerEp* pEp); diff --git a/source/dnode/mnode/impl/src/mndConsumer.c b/source/dnode/mnode/impl/src/mndConsumer.c index 614348c2098e9ca447f8077c1bbc1860e6ebb73f..168f3cd79b0a524b3e2d0cab6f140b898d49e375 100644 --- a/source/dnode/mnode/impl/src/mndConsumer.c +++ b/source/dnode/mnode/impl/src/mndConsumer.c @@ -197,11 +197,12 @@ static int32_t mndProcessMqTimerMsg(SRpcMsg *pMsg) { SMqConsumerLostMsg *pLostMsg = rpcMallocCont(sizeof(SMqConsumerLostMsg)); pLostMsg->consumerId = pConsumer->consumerId; - SRpcMsg *pRpcMsg = taosMemoryCalloc(1, sizeof(SRpcMsg)); - pRpcMsg->msgType = TDMT_MND_MQ_CONSUMER_LOST; - pRpcMsg->pCont = pLostMsg; - pRpcMsg->contLen = sizeof(SMqConsumerLostMsg); - tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, pRpcMsg); + SRpcMsg pRpcMsg = { + .msgType = TDMT_MND_MQ_CONSUMER_LOST, + .pCont = pLostMsg, + .contLen = sizeof(SMqConsumerLostMsg), + }; + tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &pRpcMsg); } if (status == MQ_CONSUMER_STATUS__LOST_REBD || status == MQ_CONSUMER_STATUS__READY) { // do nothing @@ -280,11 +281,12 @@ static int32_t mndProcessMqHbReq(SRpcMsg *pMsg) { SMqConsumerRecoverMsg *pRecoverMsg = rpcMallocCont(sizeof(SMqConsumerRecoverMsg)); pRecoverMsg->consumerId = consumerId; - SRpcMsg *pRpcMsg = taosMemoryCalloc(1, sizeof(SRpcMsg)); - pRpcMsg->msgType = TDMT_MND_MQ_CONSUMER_RECOVER; - pRpcMsg->pCont = pRecoverMsg; - pRpcMsg->contLen = sizeof(SMqConsumerRecoverMsg); - tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, pRpcMsg); + SRpcMsg pRpcMsg = { + .msgType = TDMT_MND_MQ_CONSUMER_RECOVER, + .pCont = pRecoverMsg, + .contLen = sizeof(SMqConsumerRecoverMsg), + }; + tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &pRpcMsg); } mndReleaseConsumer(pMnode, pConsumer); @@ -318,11 +320,12 @@ static int32_t mndProcessAskEpReq(SRpcMsg *pMsg) { SMqConsumerRecoverMsg *pRecoverMsg = rpcMallocCont(sizeof(SMqConsumerRecoverMsg)); pRecoverMsg->consumerId = consumerId; - SRpcMsg *pRpcMsg = taosMemoryCalloc(1, sizeof(SRpcMsg)); - pRpcMsg->msgType = TDMT_MND_MQ_CONSUMER_RECOVER; - pRpcMsg->pCont = pRecoverMsg; - pRpcMsg->contLen = sizeof(SMqConsumerRecoverMsg); - tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, pRpcMsg); + SRpcMsg pRpcMsg = { + .msgType = TDMT_MND_MQ_CONSUMER_RECOVER, + .pCont = pRecoverMsg, + .contLen = sizeof(SMqConsumerRecoverMsg), + }; + tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &pRpcMsg); } #endif diff --git a/source/dnode/mnode/impl/src/mndDef.c b/source/dnode/mnode/impl/src/mndDef.c index e6f1a40993fcb7adf2fec121b5e99374c48aae8f..67e599cfc777359c03e3dbfccea0b5fe8b29b85f 100644 --- a/source/dnode/mnode/impl/src/mndDef.c +++ b/source/dnode/mnode/impl/src/mndDef.c @@ -343,8 +343,8 @@ SMqConsumerEp *tCloneSMqConsumerEp(const SMqConsumerEp *pConsumerEpOld) { return pConsumerEpNew; } -void tDeleteSMqConsumerEp(SMqConsumerEp *pConsumerEp) { - // +void tDeleteSMqConsumerEp(void *data) { + SMqConsumerEp *pConsumerEp = (SMqConsumerEp*)data; taosArrayDestroyP(pConsumerEp->vgs, (FDelete)tDeleteSMqVgEp); } diff --git a/source/dnode/vnode/src/meta/metaQuery.c b/source/dnode/vnode/src/meta/metaQuery.c index a34569b08ecdd87483dbf31e8c6d7e406e8ae766..9b199ecd37d13cc52166c9d9242cca61c68b206b 100644 --- a/source/dnode/vnode/src/meta/metaQuery.c +++ b/source/dnode/vnode/src/meta/metaQuery.c @@ -319,8 +319,12 @@ _query: pSchema = tCloneSSchemaWrapper(&meNew.stbEntry.schemaRow); tDecoderClear(&dcNew); tdbTbcClose(pCur); + tdbFree(pKey); + tdbFree(pVal); goto _exit; } + tdbFree(pKey); + tdbFree(pVal); tdbTbcClose(pCur); } } else if (me.type == TSDB_CHILD_TABLE) { @@ -347,11 +351,13 @@ _query: tDecoderClear(&dc); _exit: + tDecoderClear(&dc); metaULock(pMeta); tdbFree(pData); return pSchema; _err: + tDecoderClear(&dc); metaULock(pMeta); tdbFree(pData); return NULL; @@ -382,11 +388,9 @@ int metaTtlSmaller(SMeta *pMeta, uint64_t ttl, SArray *uidList) { } ttlKey = *(STtlIdxKey *)pKey; taosArrayPush(uidList, &ttlKey.uid); + tdbFree(pKey); } tdbTbcClose(pCur); - - tdbFree(pKey); - return 0; } diff --git a/source/dnode/vnode/src/meta/metaSnapshot.c b/source/dnode/vnode/src/meta/metaSnapshot.c index 0edbd092e6b06883cc1e2b6be66e0ea55b8563a1..9fdbe50f88ab9fbdd2570ea76f51d7cb26772d06 100644 --- a/source/dnode/vnode/src/meta/metaSnapshot.c +++ b/source/dnode/vnode/src/meta/metaSnapshot.c @@ -353,6 +353,8 @@ int32_t buildSnapContext(SMeta* pMeta, int64_t snapVersion, int64_t suid, int8_t metaDebug("tmqsnap init idVersion uid:%" PRIi64 " version:%" PRIi64 " index:%d", *uid, idData->version, idData->index); } + tdbFree(pKey); + tdbFree(pVal); return TDB_CODE_SUCCESS; } @@ -528,6 +530,7 @@ int32_t getMetafromSnapShot(SSnapContext* ctx, void **pBuf, int32_t *contLen, in } } } + taosArrayDestroy(pTagVals); } // SIdInfo* sidInfo = (SIdInfo*)taosHashGet(ctx->idVersion, &me.ctbEntry.suid, sizeof(tb_uid_t)); // if(sidInfo->version >= idInfo->version){ diff --git a/source/dnode/vnode/src/meta/metaTable.c b/source/dnode/vnode/src/meta/metaTable.c index 92cf90d32868e010424522af6b729c2de68eec66..7768b484e95a9c6fba6dd950082aaec83c0298d0 100644 --- a/source/dnode/vnode/src/meta/metaTable.c +++ b/source/dnode/vnode/src/meta/metaTable.c @@ -1191,10 +1191,11 @@ static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry) { const void *pTagData = NULL; // int32_t nTagData = 0; SDecoder dc = {0}; - + int32_t ret = 0; // get super table if (tdbTbGet(pMeta->pUidIdx, &pCtbEntry->ctbEntry.suid, sizeof(tb_uid_t), &pData, &nData) != 0) { - return -1; + ret = -1; + goto end; } tbDbKey.uid = pCtbEntry->ctbEntry.suid; tbDbKey.version = ((SUidIdxVal *)pData)[0].version; @@ -1220,17 +1221,20 @@ static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry) { // nTagData = ((const STag *)pCtbEntry->ctbEntry.pTags)->len; pTagData = pCtbEntry->ctbEntry.pTags; nTagData = ((const STag *)pCtbEntry->ctbEntry.pTags)->len; - return metaSaveJsonVarToIdx(pMeta, pCtbEntry, pTagColumn); + ret = metaSaveJsonVarToIdx(pMeta, pCtbEntry, pTagColumn); + goto end; } if (metaCreateTagIdxKey(pCtbEntry->ctbEntry.suid, pTagColumn->colId, pTagData, nTagData, pTagColumn->type, pCtbEntry->uid, &pTagIdxKey, &nTagIdxKey) < 0) { - return -1; + ret = -1; + goto end; } tdbTbUpsert(pMeta->pTagIdx, pTagIdxKey, nTagIdxKey, NULL, 0, &pMeta->txn); +end: metaDestroyTagIdxKey(pTagIdxKey); tDecoderClear(&dc); tdbFree(pData); - return 0; + return ret; } static int metaSaveToSkmDb(SMeta *pMeta, const SMetaEntry *pME) { diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index f9c2757c3778446dfc75cca2fa55556e54d28d44..73f9e2f679670c6a1ba79a79b9d895aa5e152f93 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -51,6 +51,20 @@ void tqCleanUp() { } } +static void destroySTqHandle(void* data) { + STqHandle* pData = (STqHandle*)data; + qDestroyTask(pData->execHandle.task); + if (pData->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) { + } else if (pData->execHandle.subType == TOPIC_SUB_TYPE__DB) { + tqCloseReader(pData->execHandle.pExecReader); + walCloseReader(pData->pWalReader); + taosHashCleanup(pData->execHandle.execDb.pFilterOutTbUid); + } else if (pData->execHandle.subType == TOPIC_SUB_TYPE__TABLE){ + walCloseReader(pData->pWalReader); + tqCloseReader(pData->execHandle.pExecReader); + } +} + STQ* tqOpen(const char* path, SVnode* pVnode) { STQ* pTq = taosMemoryCalloc(1, sizeof(STQ)); if (pTq == NULL) { @@ -62,6 +76,8 @@ STQ* tqOpen(const char* path, SVnode* pVnode) { pTq->pHandle = taosHashInit(64, MurmurHash3_32, true, HASH_ENTRY_LOCK); + taosHashSetFreeFp(pTq->pHandle, destroySTqHandle); + pTq->pPushMgr = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_ENTRY_LOCK); pTq->pCheckInfo = taosHashInit(64, MurmurHash3_32, true, HASH_ENTRY_LOCK); @@ -520,6 +536,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) { int64_t fetchVer = fetchOffsetNew.version + 1; pCkHead = taosMemoryMalloc(sizeof(SWalCkHead) + 2048); if (pCkHead == NULL) { + tDeleteSTaosxRsp(&taosxRsp); return -1; } @@ -580,14 +597,17 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) { if (tqSendMetaPollRsp(pTq, pMsg, pReq, &metaRsp) < 0) { code = -1; taosMemoryFree(pCkHead); + tDeleteSTaosxRsp(&taosxRsp); return code; } code = 0; if (pCkHead) taosMemoryFree(pCkHead); + tDeleteSTaosxRsp(&taosxRsp); return code; } } } + tDeleteSTaosxRsp(&taosxRsp); return 0; } diff --git a/source/dnode/vnode/src/tq/tqOffset.c b/source/dnode/vnode/src/tq/tqOffset.c index 5c1d5d65b4f74297fcc0db81d16788f15ee58ab7..7097591c35820398547bc6093ee74c4e89a02ad8 100644 --- a/source/dnode/vnode/src/tq/tqOffset.c +++ b/source/dnode/vnode/src/tq/tqOffset.c @@ -145,8 +145,10 @@ int32_t tqOffsetCommitFile(STqOffsetStore* pStore) { ASSERT(0); tqError("write offset incomplete, len %d, write len %" PRId64, bodyLen, writeLen); taosHashCancelIterate(pStore->pHash, pIter); + taosMemoryFree(buf); return -1; } + taosMemoryFree(buf); } // close and rename file taosCloseFile(&pFile); diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index 815d6a71ef4666e71842884409818a1719a07c5d..397d8460c7f419c39a145ce75b11e00ba007be6b 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -3645,6 +3645,11 @@ static void cleanupTableSchemaInfo(SSchemaInfo* pSchemaInfo) { tDeleteSSchemaWrapper(pSchemaInfo->qsw); } +static void cleanupStreamInfo(SStreamTaskInfo* pStreamInfo) { + tDeleteSSchemaWrapper(pStreamInfo->schema); + cleanupQueryTableDataCond(&pStreamInfo->tableCond); +} + static int32_t sortTableGroup(STableListInfo* pTableListInfo) { taosArrayClear(pTableListInfo->pGroupList); SArray* sortSupport = taosArrayInit(16, sizeof(uint64_t)); @@ -4338,6 +4343,7 @@ void doDestroyTask(SExecTaskInfo* pTaskInfo) { doDestroyTableList(&pTaskInfo->tableqinfoList); destroyOperatorInfo(pTaskInfo->pRoot); cleanupTableSchemaInfo(&pTaskInfo->schemaInfo); + cleanupStreamInfo(&pTaskInfo->streamInfo); nodesDestroyNode((SNode*)pTaskInfo->pSubplan); diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index ad9cd1ffe7909c9a67e5af2e98193995757a05c2..0df70b48185bcfc6375fda6d07f5605ee3a213db 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -1769,6 +1769,7 @@ static SSDataBlock* doRawScan(SOperatorInfo* pOperator) { qDebug("tmqsnap change get data uid:%ld", mtInfo.uid); qStreamPrepareScan(pTaskInfo, &pTaskInfo->streamInfo.prepareStatus, pInfo->sContext->subType); } + tDeleteSSchemaWrapper(mtInfo.schema); qDebug("tmqsnap stream scan tsdb return null"); return NULL; } else if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__SNAPSHOT_META) { diff --git a/source/libs/qworker/src/qwMsg.c b/source/libs/qworker/src/qwMsg.c index ecc25e9c2df6fd1d0efb35cc21d871fde64784c5..c38c9ac1cf522d747188e5e3fff5c7a81944a490 100644 --- a/source/libs/qworker/src/qwMsg.c +++ b/source/libs/qworker/src/qwMsg.c @@ -632,6 +632,7 @@ int32_t qWorkerProcessDeleteMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, SD QW_ERR_JRET(qwProcessDelete(QW_FPARAMS(), &qwMsg, pRes)); + taosMemoryFreeClear(req.msg); QW_SCH_TASK_DLOG("processDelete end, node:%p", node); _return: diff --git a/utils/test/c/tmq_taosx_ci.c b/utils/test/c/tmq_taosx_ci.c index 038723955f37bbdfece0e7cca29720afefb2f20b..3a598ba98b59d4e55f1a12cc7dd60745ac915d74 100644 --- a/utils/test/c/tmq_taosx_ci.c +++ b/utils/test/c/tmq_taosx_ci.c @@ -68,6 +68,7 @@ static void msg_process(TAOS_RES* msg) { tmq_raw_data raw = {0}; tmq_get_raw(msg, &raw); + printf("write raw data type: %d\n", raw.raw_type); int32_t ret = tmq_write_raw(pConn, raw); printf("write raw data: %s\n", tmq_err2str(ret)); tmq_free_raw(raw);