From d3f54e678f15233bce75fb5f77d4acffda543f30 Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Wed, 20 Sep 2023 18:27:43 +0800 Subject: [PATCH 1/8] feat:support vnode split in tmq --- source/dnode/mnode/impl/inc/mndSubscribe.h | 8 +- source/dnode/mnode/impl/src/mndConsumer.c | 34 +- source/dnode/mnode/impl/src/mndSubscribe.c | 116 +++++-- tests/system-test/7-tmq/tmqVnodeSplit.py | 329 +++++++++++++++++++ tests/system-test/7-tmq/tmqVnodeTransform.py | 4 +- 5 files changed, 454 insertions(+), 37 deletions(-) create mode 100644 tests/system-test/7-tmq/tmqVnodeSplit.py diff --git a/source/dnode/mnode/impl/inc/mndSubscribe.h b/source/dnode/mnode/impl/inc/mndSubscribe.h index ba4328b8fe..10864da5fb 100644 --- a/source/dnode/mnode/impl/inc/mndSubscribe.h +++ b/source/dnode/mnode/impl/inc/mndSubscribe.h @@ -32,11 +32,11 @@ void mndReleaseSubscribe(SMnode *pMnode, SMqSubscribeObj *pSub); int32_t mndMakeSubscribeKey(char *key, const char *cgroup, const char *topicName); -static FORCE_INLINE int32_t mndMakePartitionKey(char *key, const char *cgroup, const char *topicName, int32_t vgId) { - return snprintf(key, TSDB_PARTITION_KEY_LEN, "%d:%s:%s", vgId, cgroup, topicName); -} +//static FORCE_INLINE int32_t mndMakePartitionKey(char *key, const char *cgroup, const char *topicName, int32_t vgId) { +// return snprintf(key, TSDB_PARTITION_KEY_LEN, "%d:%s:%s", vgId, cgroup, topicName); +//} -int32_t mndDropSubByDB(SMnode *pMnode, STrans *pTrans, SDbObj *pDb); +//int32_t mndDropSubByDB(SMnode *pMnode, STrans *pTrans, SDbObj *pDb); int32_t mndDropSubByTopic(SMnode *pMnode, STrans *pTrans, const char *topic); int32_t mndSetDropSubCommitLogs(SMnode *pMnode, STrans *pTrans, SMqSubscribeObj *pSub); diff --git a/source/dnode/mnode/impl/src/mndConsumer.c b/source/dnode/mnode/impl/src/mndConsumer.c index c1494fd0d0..30ed4ed07f 100644 --- a/source/dnode/mnode/impl/src/mndConsumer.c +++ b/source/dnode/mnode/impl/src/mndConsumer.c @@ -311,6 +311,34 @@ static int32_t mndProcessMqTimerMsg(SRpcMsg *pMsg) { taosArrayPush(pRebSub->removedConsumers, &pConsumer->consumerId); } taosRUnLockLatch(&pConsumer->lock); + }else{ + int32_t newTopicNum = taosArrayGetSize(pConsumer->currentTopics); + for (int32_t i = 0; i < newTopicNum; i++) { + char * topic = taosArrayGetP(pConsumer->currentTopics, i); + SMqSubscribeObj *pSub = mndAcquireSubscribe(pMnode, pConsumer->cgroup, topic); + if (pSub == NULL) { + continue; + } + taosRLockLatch(&pSub->lock); + + // 2.2 iterate all vg assigned to the consumer of that topic + SMqConsumerEp *pConsumerEp = taosHashGet(pSub->consumerHash, &pConsumer->consumerId, sizeof(int64_t)); + int32_t vgNum = taosArrayGetSize(pConsumerEp->vgs); + + for (int32_t j = 0; j < vgNum; j++) { + SMqVgEp *pVgEp = taosArrayGetP(pConsumerEp->vgs, j); + SVgObj * pVgroup = mndAcquireVgroup(pMnode, pVgEp->vgId); + if (!pVgroup) { + char key[TSDB_SUBSCRIBE_KEY_LEN]; + mndMakeSubscribeKey(key, pConsumer->cgroup, topic); + mndGetOrCreateRebSub(pRebMsg->rebSubHash, key); + mInfo("vnode splitted, rebalance will be triggered"); + } + mndReleaseVgroup(pMnode, pVgroup); + } + taosRUnLockLatch(&pSub->lock); + mndReleaseSubscribe(pMnode, pSub); + } } } else if (status == MQ_CONSUMER_STATUS_LOST) { if (hbStatus > MND_CONSUMER_LOST_CLEAR_THRESHOLD) { // clear consumer if lost a day @@ -343,7 +371,7 @@ static int32_t mndProcessMqTimerMsg(SRpcMsg *pMsg) { } if (taosHashGetSize(pRebMsg->rebSubHash) != 0) { - mInfo("mq rebalance will be triggered"); + mInfo("mq rebalance will be triggered"); SRpcMsg rpcMsg = { .msgType = TDMT_MND_TMQ_DO_REBALANCE, .pCont = pRebMsg, @@ -548,8 +576,8 @@ static int32_t mndProcessAskEpReq(SRpcMsg *pMsg) { for (int32_t j = 0; j < vgNum; j++) { SMqVgEp *pVgEp = taosArrayGetP(pConsumerEp->vgs, j); - char offsetKey[TSDB_PARTITION_KEY_LEN]; - mndMakePartitionKey(offsetKey, pConsumer->cgroup, topic, pVgEp->vgId); +// char offsetKey[TSDB_PARTITION_KEY_LEN]; +// mndMakePartitionKey(offsetKey, pConsumer->cgroup, topic, pVgEp->vgId); if(epoch == -1){ SVgObj *pVgroup = mndAcquireVgroup(pMnode, pVgEp->vgId); diff --git a/source/dnode/mnode/impl/src/mndSubscribe.c b/source/dnode/mnode/impl/src/mndSubscribe.c index 93bafb34e1..b6ce8c17e6 100644 --- a/source/dnode/mnode/impl/src/mndSubscribe.c +++ b/source/dnode/mnode/impl/src/mndSubscribe.c @@ -361,8 +361,68 @@ static void transferVgroupsForConsumers(SMqRebOutputObj *pOutput, SHashObj *pHas } } +static int32_t processRemoveAddVgs(SMnode *pMnode, SMqRebOutputObj *pOutput){ + int32_t totalVgNum = 0; + SVgObj* pVgroup = NULL; + void* pIter = NULL; + SArray* newVgs = taosArrayInit(0, POINTER_BYTES); + while (1) { + pIter = sdbFetch(pMnode->pSdb, SDB_VGROUP, pIter, (void**)&pVgroup); + if (pIter == NULL) { + break; + } + + if (!mndVgroupInDb(pVgroup, pOutput->pSub->dbUid)) { + sdbRelease(pMnode->pSdb, pVgroup); + continue; + } + + totalVgNum++; + SMqVgEp* pVgEp = taosMemoryMalloc(sizeof(SMqVgEp)); + pVgEp->epSet = mndGetVgroupEpset(pMnode, pVgroup); + pVgEp->vgId = pVgroup->vgId; + taosArrayPush(newVgs, &pVgEp); + sdbRelease(pMnode->pSdb, pVgroup); + } + + pIter = NULL; + while (1) { + pIter = taosHashIterate(pOutput->pSub->consumerHash, pIter); + if (pIter == NULL) break; + SMqConsumerEp *pConsumerEp = (SMqConsumerEp *)pIter; + + int32_t j = 0; + while (j < taosArrayGetSize(pConsumerEp->vgs)) { + SMqVgEp *pVgEp = taosArrayGetP(pConsumerEp->vgs, j); + bool find = false; + for(int32_t k = 0; k < taosArrayGetSize(newVgs); k++){ + SMqVgEp *pnewVgEp = taosArrayGetP(newVgs, k); + if(pVgEp->vgId == pnewVgEp->vgId){ + taosArrayRemove(newVgs, k); + find = true; + break; + } + } + if(!find){ + mInfo("processRemoveAddVgs old vgId:%d", pVgEp->vgId); + tDeleteSMqVgEp(pVgEp); + taosArrayRemove(pConsumerEp->vgs, j); + continue; + } + j++; + } + } + + if(taosArrayGetSize(pOutput->pSub->unassignedVgs) == 0 && taosArrayGetSize(newVgs) != 0){ + taosArrayAddAll(pOutput->pSub->unassignedVgs, newVgs); + mInfo("processRemoveAddVgs add new vg num:%d", (int)taosArrayGetSize(newVgs)); + } + taosArrayDestroy(newVgs); + return totalVgNum; +} + static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqRebOutputObj *pOutput) { - int32_t totalVgNum = pOutput->pSub->vgNum; + int32_t totalVgNum = processRemoveAddVgs(pMnode, pOutput); const char *pSubKey = pOutput->pSub->key; int32_t numOfRemoved = taosArrayGetSize(pInput->pRebInfo->removedConsumers); @@ -1093,33 +1153,33 @@ int32_t mndSetDropSubCommitLogs(SMnode *pMnode, STrans *pTrans, SMqSubscribeObj return 0; } -int32_t mndDropSubByDB(SMnode *pMnode, STrans *pTrans, SDbObj *pDb) { - int32_t code = 0; - SSdb *pSdb = pMnode->pSdb; - - void *pIter = NULL; - SMqSubscribeObj *pSub = NULL; - while (1) { - pIter = sdbFetch(pSdb, SDB_SUBSCRIBE, pIter, (void **)&pSub); - if (pIter == NULL) break; - - if (pSub->dbUid != pDb->uid) { - sdbRelease(pSdb, pSub); - continue; - } - - if (mndSetDropSubCommitLogs(pMnode, pTrans, pSub) < 0) { - sdbRelease(pSdb, pSub); - sdbCancelFetch(pSdb, pIter); - code = -1; - break; - } - - sdbRelease(pSdb, pSub); - } - - return code; -} +//int32_t mndDropSubByDB(SMnode *pMnode, STrans *pTrans, SDbObj *pDb) { +// int32_t code = 0; +// SSdb *pSdb = pMnode->pSdb; +// +// void *pIter = NULL; +// SMqSubscribeObj *pSub = NULL; +// while (1) { +// pIter = sdbFetch(pSdb, SDB_SUBSCRIBE, pIter, (void **)&pSub); +// if (pIter == NULL) break; +// +// if (pSub->dbUid != pDb->uid) { +// sdbRelease(pSdb, pSub); +// continue; +// } +// +// if (mndSetDropSubCommitLogs(pMnode, pTrans, pSub) < 0) { +// sdbRelease(pSdb, pSub); +// sdbCancelFetch(pSdb, pIter); +// code = -1; +// break; +// } +// +// sdbRelease(pSdb, pSub); +// } +// +// return code; +//} int32_t mndDropSubByTopic(SMnode *pMnode, STrans *pTrans, const char *topicName) { SSdb *pSdb = pMnode->pSdb; diff --git a/tests/system-test/7-tmq/tmqVnodeSplit.py b/tests/system-test/7-tmq/tmqVnodeSplit.py new file mode 100644 index 0000000000..ee4dc32e40 --- /dev/null +++ b/tests/system-test/7-tmq/tmqVnodeSplit.py @@ -0,0 +1,329 @@ + +import taos +import sys +import time +import socket +import os +import threading +import math + +from util.log import * +from util.sql import * +from util.cases import * +from util.dnodes import * +from util.common import * +from util.cluster import * +sys.path.append("./7-tmq") +from tmqCommon import * + +class TDTestCase: + def __init__(self): + self.vgroups = 1 + self.ctbNum = 10 + self.rowsPerTbl = 10000 + + def init(self, conn, logSql, replicaVar=1): + self.replicaVar = int(replicaVar) + tdLog.debug(f"start to excute {__file__}") + tdSql.init(conn.cursor(), False) + + def getDataPath(self): + selfPath = os.path.dirname(os.path.realpath(__file__)) + + return selfPath + '/../../../sim/dnode%d/data/vnode/vnode%d/wal/*'; + + def prepareTestEnv(self): + tdLog.printNoPrefix("======== prepare test env include database, stable, ctables, and insert data: ") + paraDict = {'dbName': 'dbt', + 'dropFlag': 1, + 'event': '', + 'vgroups': 1, + 'stbName': 'stb', + 'colPrefix': 'c', + 'tagPrefix': 't', + 'colSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1},{'type': 'TIMESTAMP', 'count':1}], + 'tagSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1}], + 'ctbPrefix': 'ctb', + 'ctbStartIdx': 0, + 'ctbNum': 10, + 'rowsPerTbl': 10000, + 'batchNum': 10, + 'startTs': 1640966400000, # 2022-01-01 00:00:00.000 + 'pollDelay': 60, + 'showMsg': 1, + 'showRow': 1, + 'snapshot': 0} + + paraDict['vgroups'] = self.vgroups + paraDict['ctbNum'] = self.ctbNum + paraDict['rowsPerTbl'] = self.rowsPerTbl + + tmqCom.initConsumerTable() + tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], wal_retention_period=36000,vgroups=paraDict["vgroups"],replica=self.replicaVar) + tdLog.info("create stb") + tmqCom.create_stable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"]) + # tdLog.info("create ctb") + # tmqCom.create_ctable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=paraDict['ctbPrefix'], + # ctbNum=paraDict["ctbNum"],ctbStartIdx=paraDict['ctbStartIdx']) + # tdLog.info("insert data") + # tmqCom.insert_data_interlaceByMultiTbl(tsql=tdSql,dbName=paraDict["dbName"],ctbPrefix=paraDict["ctbPrefix"], + # ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"], + # startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx']) + + # tdLog.info("restart taosd to ensure that the data falls into the disk") + # tdDnodes.stop(1) + # tdDnodes.start(1) + # tdSql.query("flush database %s"%(paraDict['dbName'])) + return + + def restartAndRemoveWal(self): + tdDnodes = cluster.dnodes + tdSql.query("select * from information_schema.ins_vnodes") + for result in tdSql.queryResult: + if result[2] == 'dbt': + tdLog.debug("dnode is %d"%(result[0])) + dnodeId = result[0] + vnodeId = result[1] + + tdDnodes[dnodeId - 1].stoptaosd() + time.sleep(1) + dataPath = self.getDataPath() + dataPath = dataPath%(dnodeId,vnodeId) + os.system('rm -rf ' + dataPath) + tdLog.debug("dataPath:%s"%dataPath) + tdDnodes[dnodeId - 1].starttaosd() + time.sleep(1) + break + tdLog.debug("restart dnode ok") + + def redistributeVgroups(self): + tdSql.query("select * from information_schema.ins_vnodes") + vnodeId = 0 + for result in tdSql.queryResult: + if result[2] == 'dbt': + vnodeId = result[1] + tdLog.debug("vnode is %d"%(vnodeId)) + break + splitSql = "split vgroup %d" %(vnodeId) + tdLog.debug("splitSql:%s"%(splitSql)) + tdSql.query(splitSql) + tdLog.debug("splitSql ok") + + def tmqCase1(self): + tdLog.printNoPrefix("======== test case 1: ") + paraDict = {'dbName': 'dbt', + 'dropFlag': 1, + 'event': '', + 'vgroups': 1, + 'stbName': 'stb', + 'colPrefix': 'c', + 'tagPrefix': 't', + 'colSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1},{'type': 'TIMESTAMP', 'count':1}], + 'tagSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1}], + 'ctbPrefix': 'ctb', + 'ctbStartIdx': 0, + 'ctbNum': 10, + 'rowsPerTbl': 10000, + 'batchNum': 10, + 'startTs': 1640966400000, # 2022-01-01 00:00:00.000 + 'pollDelay': 60, + 'showMsg': 1, + 'showRow': 1, + 'snapshot': 0} + + paraDict['vgroups'] = self.vgroups + paraDict['ctbNum'] = self.ctbNum + paraDict['rowsPerTbl'] = self.rowsPerTbl + + topicNameList = ['topic1'] + # expectRowsList = [] + tmqCom.initConsumerTable() + + tdLog.info("create topics from stb with filter") + queryString = "select * from %s.%s"%(paraDict['dbName'], paraDict['stbName']) + # sqlString = "create topic %s as stable %s" %(topicNameList[0], paraDict['stbName']) + sqlString = "create topic %s as %s" %(topicNameList[0], queryString) + tdLog.info("create topic sql: %s"%sqlString) + tdSql.execute(sqlString) + # tdSql.query(queryString) + # expectRowsList.append(tdSql.getRows()) + + # init consume info, and start tmq_sim, then check consume result + tdLog.info("insert consume info to consume processor") + consumerId = 0 + expectrowcnt = paraDict["rowsPerTbl"] * paraDict["ctbNum"] * 2 + topicList = topicNameList[0] + ifcheckdata = 1 + ifManualCommit = 1 + keyList = 'group.id:cgrp1, enable.auto.commit:true, auto.commit.interval.ms:200, auto.offset.reset:earliest' + tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit) + + tdLog.info("start consume processor") + tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot']) + tdLog.info("wait the consume result") + + tdLog.info("create ctb1") + tmqCom.create_ctable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=paraDict['ctbPrefix'], + ctbNum=paraDict["ctbNum"],ctbStartIdx=paraDict['ctbStartIdx']) + tdLog.info("insert data") + pInsertThread = tmqCom.asyncInsertDataByInterlace(paraDict) + + tmqCom.getStartConsumeNotifyFromTmqsim() + tmqCom.getStartCommitNotifyFromTmqsim() + + #restart dnode & remove wal + self.restartAndRemoveWal() + + # redistribute vgroup + self.redistributeVgroups(); + + tdLog.info("create ctb2") + paraDict['ctbPrefix'] = "ctbn" + tmqCom.create_ctable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=paraDict['ctbPrefix'], + ctbNum=paraDict["ctbNum"],ctbStartIdx=paraDict['ctbStartIdx']) + tdLog.info("insert data") + pInsertThread1 = tmqCom.asyncInsertDataByInterlace(paraDict) + pInsertThread.join() + pInsertThread1.join() + + expectRows = 1 + resultList = tmqCom.selectConsumeResult(expectRows) + + if expectrowcnt / 2 >= resultList[0]: + tdLog.info("expect consume rows: %d, act consume rows: %d"%(expectrowcnt / 2, resultList[0])) + tdLog.exit("%d tmq consume rows error!"%consumerId) + + # tmqCom.checkFileContent(consumerId, queryString) + + time.sleep(10) + for i in range(len(topicNameList)): + tdSql.query("drop topic %s"%topicNameList[i]) + + tdLog.printNoPrefix("======== test case 1 end ...... ") + + def tmqCase2(self): + tdLog.printNoPrefix("======== test case 2: ") + paraDict = {'dbName':'dbt'} + + ntbName = "ntb" + + topicNameList = ['topic2'] + tmqCom.initConsumerTable() + + sqlString = "create table %s.%s(ts timestamp, i nchar(8))" %(paraDict['dbName'], ntbName) + tdLog.info("create nomal table sql: %s"%sqlString) + tdSql.execute(sqlString) + + tdLog.info("create topics from nomal table") + queryString = "select * from %s.%s"%(paraDict['dbName'], ntbName) + sqlString = "create topic %s as %s" %(topicNameList[0], queryString) + tdLog.info("create topic sql: %s"%sqlString) + tdSql.execute(sqlString) + tdSql.query("flush database %s"%(paraDict['dbName'])) + #restart dnode & remove wal + self.restartAndRemoveWal() + + # redistribute vgroup + self.redistributeVgroups(); + + sqlString = "alter table %s.%s modify column i nchar(16)" %(paraDict['dbName'], ntbName) + tdLog.info("alter table sql: %s"%sqlString) + tdSql.error(sqlString) + + time.sleep(1) + for i in range(len(topicNameList)): + tdSql.query("drop topic %s"%topicNameList[i]) + + tdLog.printNoPrefix("======== test case 2 end ...... ") + + def tmqCase3(self): + tdLog.printNoPrefix("======== test case 3: ") + paraDict = {'dbName': 'dbt', + 'dropFlag': 1, + 'event': '', + 'vgroups': 1, + 'stbName': 'stbn', + 'colPrefix': 'c', + 'tagPrefix': 't', + 'colSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1},{'type': 'TIMESTAMP', 'count':1}], + 'tagSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1}], + 'ctbPrefix': 'ctb', + 'ctbStartIdx': 0, + 'ctbNum': 10, + 'rowsPerTbl': 10000, + 'batchNum': 10, + 'startTs': 1640966400000, # 2022-01-01 00:00:00.000 + 'pollDelay': 2, + 'showMsg': 1, + 'showRow': 1, + 'snapshot': 0} + + paraDict['vgroups'] = self.vgroups + paraDict['ctbNum'] = self.ctbNum + paraDict['rowsPerTbl'] = self.rowsPerTbl + + topicNameList = ['topic3'] + tmqCom.initConsumerTable() + + tdLog.info("create stb") + tmqCom.create_stable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"]) + + tdLog.info("create ctb") + tmqCom.create_ctable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=paraDict['ctbPrefix'], + ctbNum=paraDict["ctbNum"],ctbStartIdx=paraDict['ctbStartIdx']) + tdLog.info("insert data") + tmqCom.insert_data_interlaceByMultiTbl(tsql=tdSql,dbName=paraDict["dbName"],ctbPrefix=paraDict["ctbPrefix"], + ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"], + startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx']) + + tdLog.info("create topics from stb with filter") + queryString = "select * from %s.%s"%(paraDict['dbName'], paraDict['stbName']) + sqlString = "create topic %s as %s" %(topicNameList[0], queryString) + tdLog.info("create topic sql: %s"%sqlString) + tdSql.execute(sqlString) + + # init consume info, and start tmq_sim, then check consume result + tdLog.info("insert consume info to consume processor") + consumerId = 0 + expectrowcnt = paraDict["rowsPerTbl"] * paraDict["ctbNum"] + topicList = topicNameList[0] + ifcheckdata = 1 + ifManualCommit = 1 + keyList = 'group.id:cgrp1, enable.auto.commit:true, auto.commit.interval.ms:200, auto.offset.reset:earliest' + tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit) + + tdLog.info("start consume processor") + tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot']) + tdLog.info("wait the consume result") + + time.sleep(5) + #restart dnode & remove wal + self.restartAndRemoveWal() + + # redistribute vgroup + self.redistributeVgroups(); + + tdLog.info("start consume processor") + tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot']) + tdLog.info("wait the consume result") + + time.sleep(10) + for i in range(len(topicNameList)): + tdSql.query("drop topic %s"%topicNameList[i]) + + tdLog.printNoPrefix("======== test case 3 end ...... ") + + def run(self): + + tdSql.prepare() + self.prepareTestEnv() + self.tmqCase1() + + def stop(self): + tdSql.close() + tdLog.success(f"{__file__} successfully executed") + +event = threading.Event() + +tdCases.addLinux(__file__, TDTestCase()) +tdCases.addWindows(__file__, TDTestCase()) diff --git a/tests/system-test/7-tmq/tmqVnodeTransform.py b/tests/system-test/7-tmq/tmqVnodeTransform.py index 8db9ce0e13..8b2524a6bc 100644 --- a/tests/system-test/7-tmq/tmqVnodeTransform.py +++ b/tests/system-test/7-tmq/tmqVnodeTransform.py @@ -324,8 +324,8 @@ class TDTestCase: tdSql.prepare() self.prepareTestEnv() self.tmqCase1() - # self.tmqCase2() - # self.tmqCase3() + self.tmqCase2() + self.tmqCase3() def stop(self): tdSql.close() From f9c5aa01af8827c41e1fad4a4896793aa2f9f4bc Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Thu, 21 Sep 2023 09:20:34 +0800 Subject: [PATCH 2/8] fix:enable split if there are topics & fix memory leak --- source/dnode/mnode/impl/src/mndConsumer.c | 2 +- source/dnode/mnode/impl/src/mndSubscribe.c | 2 +- source/dnode/mnode/impl/src/mndVgroup.c | 16 ++++++++-------- source/dnode/vnode/src/tq/tq.c | 2 +- 4 files changed, 11 insertions(+), 11 deletions(-) diff --git a/source/dnode/mnode/impl/src/mndConsumer.c b/source/dnode/mnode/impl/src/mndConsumer.c index 30ed4ed07f..7273e13317 100644 --- a/source/dnode/mnode/impl/src/mndConsumer.c +++ b/source/dnode/mnode/impl/src/mndConsumer.c @@ -332,7 +332,7 @@ static int32_t mndProcessMqTimerMsg(SRpcMsg *pMsg) { char key[TSDB_SUBSCRIBE_KEY_LEN]; mndMakeSubscribeKey(key, pConsumer->cgroup, topic); mndGetOrCreateRebSub(pRebMsg->rebSubHash, key); - mInfo("vnode splitted, rebalance will be triggered"); + mInfo("vnode splitted, vgId:%d rebalance will be triggered", pVgEp->vgId); } mndReleaseVgroup(pMnode, pVgroup); } diff --git a/source/dnode/mnode/impl/src/mndSubscribe.c b/source/dnode/mnode/impl/src/mndSubscribe.c index b6ce8c17e6..de3866f974 100644 --- a/source/dnode/mnode/impl/src/mndSubscribe.c +++ b/source/dnode/mnode/impl/src/mndSubscribe.c @@ -417,7 +417,7 @@ static int32_t processRemoveAddVgs(SMnode *pMnode, SMqRebOutputObj *pOutput){ taosArrayAddAll(pOutput->pSub->unassignedVgs, newVgs); mInfo("processRemoveAddVgs add new vg num:%d", (int)taosArrayGetSize(newVgs)); } - taosArrayDestroy(newVgs); + taosArrayDestroyP(newVgs, (FDelete)tDeleteSMqVgEp); return totalVgNum; } diff --git a/source/dnode/mnode/impl/src/mndVgroup.c b/source/dnode/mnode/impl/src/mndVgroup.c index d819b71f18..9d27a27365 100644 --- a/source/dnode/mnode/impl/src/mndVgroup.c +++ b/source/dnode/mnode/impl/src/mndVgroup.c @@ -2672,14 +2672,14 @@ int32_t mndSplitVgroup(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SVgObj *pVgro SDbObj dbObj = {0}; SArray *pArray = mndBuildDnodesArray(pMnode, 0); - int32_t numOfTopics = 0; - if (mndGetNumOfTopics(pMnode, pDb->name, &numOfTopics) != 0) { - goto _OVER; - } - if (numOfTopics > 0) { - terrno = TSDB_CODE_MND_TOPIC_MUST_BE_DELETED; - goto _OVER; - } +// int32_t numOfTopics = 0; +// if (mndGetNumOfTopics(pMnode, pDb->name, &numOfTopics) != 0) { +// goto _OVER; +// } +// if (numOfTopics > 0) { +// terrno = TSDB_CODE_MND_TOPIC_MUST_BE_DELETED; +// goto _OVER; +// } int32_t numOfStreams = 0; if (mndGetNumOfStreams(pMnode, pDb->name, &numOfStreams) != 0) { diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index 56ea636234..adf3abe4d9 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -712,7 +712,7 @@ int32_t tqProcessSubscribeReq(STQ* pTq, int64_t sversion, char* msg, int32_t msg continue; } if (pHandle->consumerId == req.newConsumerId) { // do nothing - tqInfo("vgId:%d no switch consumer:0x%" PRIx64 " remains, because redo wal log", req.vgId, req.newConsumerId); + tqInfo("vgId:%d no switch consumer:0x%" PRIx64 " remains", req.vgId, req.newConsumerId); } else { tqInfo("vgId:%d switch consumer from Id:0x%" PRIx64 " to Id:0x%" PRIx64, req.vgId, pHandle->consumerId, req.newConsumerId); From ce2fc8b319b52d249348e55a394f7257ae029a50 Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Thu, 21 Sep 2023 10:05:52 +0800 Subject: [PATCH 3/8] fix:memory leak --- source/dnode/mnode/impl/src/mndSubscribe.c | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/source/dnode/mnode/impl/src/mndSubscribe.c b/source/dnode/mnode/impl/src/mndSubscribe.c index de3866f974..408b664e50 100644 --- a/source/dnode/mnode/impl/src/mndSubscribe.c +++ b/source/dnode/mnode/impl/src/mndSubscribe.c @@ -398,6 +398,7 @@ static int32_t processRemoveAddVgs(SMnode *pMnode, SMqRebOutputObj *pOutput){ for(int32_t k = 0; k < taosArrayGetSize(newVgs); k++){ SMqVgEp *pnewVgEp = taosArrayGetP(newVgs, k); if(pVgEp->vgId == pnewVgEp->vgId){ + tDeleteSMqVgEp(pnewVgEp); taosArrayRemove(newVgs, k); find = true; break; @@ -416,8 +417,10 @@ static int32_t processRemoveAddVgs(SMnode *pMnode, SMqRebOutputObj *pOutput){ if(taosArrayGetSize(pOutput->pSub->unassignedVgs) == 0 && taosArrayGetSize(newVgs) != 0){ taosArrayAddAll(pOutput->pSub->unassignedVgs, newVgs); mInfo("processRemoveAddVgs add new vg num:%d", (int)taosArrayGetSize(newVgs)); + taosArrayDestroy(newVgs); + }else{ + taosArrayDestroyP(newVgs, (FDelete)tDeleteSMqVgEp); } - taosArrayDestroyP(newVgs, (FDelete)tDeleteSMqVgEp); return totalVgNum; } From 70b5509217f7ebae7ff60f85b66fc7fd1933c849 Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Thu, 21 Sep 2023 18:16:05 +0800 Subject: [PATCH 4/8] fix:enable split if there are topics --- source/client/src/clientTmq.c | 2 +- tests/parallel_test/cases.task | 1 + tests/system-test/7-tmq/tmqVnodeSplit.py | 153 +++---------------- tests/system-test/7-tmq/tmqVnodeTransform.py | 6 +- 4 files changed, 23 insertions(+), 139 deletions(-) diff --git a/source/client/src/clientTmq.c b/source/client/src/clientTmq.c index 6241c089e9..c9c8a66c7a 100644 --- a/source/client/src/clientTmq.c +++ b/source/client/src/clientTmq.c @@ -1768,7 +1768,7 @@ static void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout) { tscError("consumer:0x%" PRIx64 " msg from vgId:%d discarded, since %s", tmq->consumerId, pollRspWrapper->vgId, tstrerror(pRspWrapper->code)); taosWLockLatch(&tmq->lock); SMqClientVg* pVg = getVgInfo(tmq, pollRspWrapper->topicName, pollRspWrapper->vgId); - pVg->emptyBlockReceiveTs = taosGetTimestampMs(); + if(pVg) pVg->emptyBlockReceiveTs = taosGetTimestampMs(); taosWUnLockLatch(&tmq->lock); } setVgIdle(tmq, pollRspWrapper->topicName, pollRspWrapper->vgId); diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task index 9c6cd5c99a..18fb3cc4dd 100644 --- a/tests/parallel_test/cases.task +++ b/tests/parallel_test/cases.task @@ -164,6 +164,7 @@ ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmq3mnodeSwitch.py -N 6 -M 3 -i True ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmq3mnodeSwitch.py -N 6 -M 3 -n 3 -i True ,,y,system-test,./pytest.sh python3 test.py -f 7-tmq/tmqVnodeTransform.py -N 2 -n 1 +,,y,system-test,./pytest.sh python3 test.py -f 7-tmq/tmqVnodeSplit.py -N 2 -n 1 ,,y,system-test,./pytest.sh python3 test.py -f 7-tmq/tmqVnodeReplicate.py -M 3 -N 3 -n 3 ,,y,system-test,./pytest.sh python3 ./test.py -f 99-TDcase/TD-19201.py ,,y,system-test,./pytest.sh python3 ./test.py -f 99-TDcase/TD-21561.py diff --git a/tests/system-test/7-tmq/tmqVnodeSplit.py b/tests/system-test/7-tmq/tmqVnodeSplit.py index ee4dc32e40..c6cdc2bf83 100644 --- a/tests/system-test/7-tmq/tmqVnodeSplit.py +++ b/tests/system-test/7-tmq/tmqVnodeSplit.py @@ -28,9 +28,9 @@ class TDTestCase: tdSql.init(conn.cursor(), False) def getDataPath(self): - selfPath = os.path.dirname(os.path.realpath(__file__)) + selfPath = tdCom.getBuildPath() - return selfPath + '/../../../sim/dnode%d/data/vnode/vnode%d/wal/*'; + return selfPath + '/../sim/dnode%d/data/vnode/vnode%d/wal/*'; def prepareTestEnv(self): tdLog.printNoPrefix("======== prepare test env include database, stable, ctables, and insert data: ") @@ -58,25 +58,14 @@ class TDTestCase: paraDict['ctbNum'] = self.ctbNum paraDict['rowsPerTbl'] = self.rowsPerTbl + tdCom.drop_all_db(); tmqCom.initConsumerTable() tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], wal_retention_period=36000,vgroups=paraDict["vgroups"],replica=self.replicaVar) tdLog.info("create stb") tmqCom.create_stable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"]) - # tdLog.info("create ctb") - # tmqCom.create_ctable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=paraDict['ctbPrefix'], - # ctbNum=paraDict["ctbNum"],ctbStartIdx=paraDict['ctbStartIdx']) - # tdLog.info("insert data") - # tmqCom.insert_data_interlaceByMultiTbl(tsql=tdSql,dbName=paraDict["dbName"],ctbPrefix=paraDict["ctbPrefix"], - # ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"], - # startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx']) - - # tdLog.info("restart taosd to ensure that the data falls into the disk") - # tdDnodes.stop(1) - # tdDnodes.start(1) - # tdSql.query("flush database %s"%(paraDict['dbName'])) return - def restartAndRemoveWal(self): + def restartAndRemoveWal(self, deleteWal): tdDnodes = cluster.dnodes tdSql.query("select * from information_schema.ins_vnodes") for result in tdSql.queryResult: @@ -89,14 +78,17 @@ class TDTestCase: time.sleep(1) dataPath = self.getDataPath() dataPath = dataPath%(dnodeId,vnodeId) - os.system('rm -rf ' + dataPath) tdLog.debug("dataPath:%s"%dataPath) + if deleteWal: + if os.system('rm -rf ' + dataPath) != 0: + tdLog.exit("rm error") + tdDnodes[dnodeId - 1].starttaosd() time.sleep(1) break tdLog.debug("restart dnode ok") - def redistributeVgroups(self): + def splitVgroups(self): tdSql.query("select * from information_schema.ins_vnodes") vnodeId = 0 for result in tdSql.queryResult: @@ -109,7 +101,7 @@ class TDTestCase: tdSql.query(splitSql) tdLog.debug("splitSql ok") - def tmqCase1(self): + def tmqCase1(self, deleteWal=False): tdLog.printNoPrefix("======== test case 1: ") paraDict = {'dbName': 'dbt', 'dropFlag': 1, @@ -172,10 +164,10 @@ class TDTestCase: tmqCom.getStartCommitNotifyFromTmqsim() #restart dnode & remove wal - self.restartAndRemoveWal() + self.restartAndRemoveWal(deleteWal) - # redistribute vgroup - self.redistributeVgroups(); + # split vgroup + self.splitVgroups() tdLog.info("create ctb2") paraDict['ctbPrefix'] = "ctbn" @@ -195,129 +187,18 @@ class TDTestCase: # tmqCom.checkFileContent(consumerId, queryString) - time.sleep(10) + time.sleep(2) for i in range(len(topicNameList)): tdSql.query("drop topic %s"%topicNameList[i]) tdLog.printNoPrefix("======== test case 1 end ...... ") - def tmqCase2(self): - tdLog.printNoPrefix("======== test case 2: ") - paraDict = {'dbName':'dbt'} - - ntbName = "ntb" - - topicNameList = ['topic2'] - tmqCom.initConsumerTable() - - sqlString = "create table %s.%s(ts timestamp, i nchar(8))" %(paraDict['dbName'], ntbName) - tdLog.info("create nomal table sql: %s"%sqlString) - tdSql.execute(sqlString) - - tdLog.info("create topics from nomal table") - queryString = "select * from %s.%s"%(paraDict['dbName'], ntbName) - sqlString = "create topic %s as %s" %(topicNameList[0], queryString) - tdLog.info("create topic sql: %s"%sqlString) - tdSql.execute(sqlString) - tdSql.query("flush database %s"%(paraDict['dbName'])) - #restart dnode & remove wal - self.restartAndRemoveWal() - - # redistribute vgroup - self.redistributeVgroups(); - - sqlString = "alter table %s.%s modify column i nchar(16)" %(paraDict['dbName'], ntbName) - tdLog.info("alter table sql: %s"%sqlString) - tdSql.error(sqlString) - - time.sleep(1) - for i in range(len(topicNameList)): - tdSql.query("drop topic %s"%topicNameList[i]) - - tdLog.printNoPrefix("======== test case 2 end ...... ") - - def tmqCase3(self): - tdLog.printNoPrefix("======== test case 3: ") - paraDict = {'dbName': 'dbt', - 'dropFlag': 1, - 'event': '', - 'vgroups': 1, - 'stbName': 'stbn', - 'colPrefix': 'c', - 'tagPrefix': 't', - 'colSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1},{'type': 'TIMESTAMP', 'count':1}], - 'tagSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1}], - 'ctbPrefix': 'ctb', - 'ctbStartIdx': 0, - 'ctbNum': 10, - 'rowsPerTbl': 10000, - 'batchNum': 10, - 'startTs': 1640966400000, # 2022-01-01 00:00:00.000 - 'pollDelay': 2, - 'showMsg': 1, - 'showRow': 1, - 'snapshot': 0} - - paraDict['vgroups'] = self.vgroups - paraDict['ctbNum'] = self.ctbNum - paraDict['rowsPerTbl'] = self.rowsPerTbl - - topicNameList = ['topic3'] - tmqCom.initConsumerTable() - - tdLog.info("create stb") - tmqCom.create_stable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"]) - - tdLog.info("create ctb") - tmqCom.create_ctable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=paraDict['ctbPrefix'], - ctbNum=paraDict["ctbNum"],ctbStartIdx=paraDict['ctbStartIdx']) - tdLog.info("insert data") - tmqCom.insert_data_interlaceByMultiTbl(tsql=tdSql,dbName=paraDict["dbName"],ctbPrefix=paraDict["ctbPrefix"], - ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"], - startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx']) - - tdLog.info("create topics from stb with filter") - queryString = "select * from %s.%s"%(paraDict['dbName'], paraDict['stbName']) - sqlString = "create topic %s as %s" %(topicNameList[0], queryString) - tdLog.info("create topic sql: %s"%sqlString) - tdSql.execute(sqlString) - - # init consume info, and start tmq_sim, then check consume result - tdLog.info("insert consume info to consume processor") - consumerId = 0 - expectrowcnt = paraDict["rowsPerTbl"] * paraDict["ctbNum"] - topicList = topicNameList[0] - ifcheckdata = 1 - ifManualCommit = 1 - keyList = 'group.id:cgrp1, enable.auto.commit:true, auto.commit.interval.ms:200, auto.offset.reset:earliest' - tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit) - - tdLog.info("start consume processor") - tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot']) - tdLog.info("wait the consume result") - - time.sleep(5) - #restart dnode & remove wal - self.restartAndRemoveWal() - - # redistribute vgroup - self.redistributeVgroups(); - - tdLog.info("start consume processor") - tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot']) - tdLog.info("wait the consume result") - - time.sleep(10) - for i in range(len(topicNameList)): - tdSql.query("drop topic %s"%topicNameList[i]) - - tdLog.printNoPrefix("======== test case 3 end ...... ") - def run(self): - tdSql.prepare() self.prepareTestEnv() - self.tmqCase1() + self.tmqCase1(True) + self.prepareTestEnv() + self.tmqCase1(False) def stop(self): tdSql.close() diff --git a/tests/system-test/7-tmq/tmqVnodeTransform.py b/tests/system-test/7-tmq/tmqVnodeTransform.py index 8b2524a6bc..fa50e46853 100644 --- a/tests/system-test/7-tmq/tmqVnodeTransform.py +++ b/tests/system-test/7-tmq/tmqVnodeTransform.py @@ -28,9 +28,9 @@ class TDTestCase: tdSql.init(conn.cursor(), False) def getDataPath(self): - selfPath = os.path.dirname(os.path.realpath(__file__)) + selfPath = tdCom.getBuildPath() - return selfPath + '/../../../sim/dnode%d/data/vnode/vnode%d/wal/*'; + return selfPath + '/../sim/dnode%d/data/vnode/vnode%d/wal/*'; def prepareTestEnv(self): tdLog.printNoPrefix("======== prepare test env include database, stable, ctables, and insert data: ") @@ -58,6 +58,7 @@ class TDTestCase: paraDict['ctbNum'] = self.ctbNum paraDict['rowsPerTbl'] = self.rowsPerTbl + tdCom.drop_all_db(); tmqCom.initConsumerTable() tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], wal_retention_period=36000,vgroups=paraDict["vgroups"],replica=self.replicaVar) tdLog.info("create stb") @@ -325,6 +326,7 @@ class TDTestCase: self.prepareTestEnv() self.tmqCase1() self.tmqCase2() + self.prepareTestEnv() self.tmqCase3() def stop(self): From 803b17a2db639e7efcf6c060e288b287f842f472 Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Fri, 22 Sep 2023 09:31:58 +0800 Subject: [PATCH 5/8] fix:memory leak --- source/client/src/clientRawBlockWrite.c | 4 +- source/dnode/vnode/src/tq/tqHandleSnapshot.c | 11 ++- source/dnode/vnode/src/tq/tqMeta.c | 70 +++++++++----------- source/libs/stream/src/streamMeta.c | 2 + 4 files changed, 41 insertions(+), 46 deletions(-) diff --git a/source/client/src/clientRawBlockWrite.c b/source/client/src/clientRawBlockWrite.c index f5cdfeadad..b7e92d2e65 100644 --- a/source/client/src/clientRawBlockWrite.c +++ b/source/client/src/clientRawBlockWrite.c @@ -1411,7 +1411,7 @@ int taos_write_raw_block_with_fields(TAOS* taos, int rows, char* pData, const ch code = smlBuildOutput(pQuery, pVgHash); if (code != TSDB_CODE_SUCCESS) { uError("smlBuildOutput failed"); - return code; + goto end; } launchQueryImpl(pRequest, pQuery, true, NULL); @@ -1496,7 +1496,7 @@ int taos_write_raw_block(TAOS* taos, int rows, char* pData, const char* tbname) code = smlBuildOutput(pQuery, pVgHash); if (code != TSDB_CODE_SUCCESS) { uError("smlBuildOutput failed"); - return code; + goto end; } launchQueryImpl(pRequest, pQuery, true, NULL); diff --git a/source/dnode/vnode/src/tq/tqHandleSnapshot.c b/source/dnode/vnode/src/tq/tqHandleSnapshot.c index 7d3e2f7837..38f9dcc57a 100644 --- a/source/dnode/vnode/src/tq/tqHandleSnapshot.c +++ b/source/dnode/vnode/src/tq/tqHandleSnapshot.c @@ -178,17 +178,14 @@ int32_t tqSnapWrite(STqSnapWriter* pWriter, uint8_t* pData, uint32_t nData) { tDecoderInit(pDecoder, pData + sizeof(SSnapDataHdr), nData - sizeof(SSnapDataHdr)); code = tDecodeSTqHandle(pDecoder, &handle); - if (code) goto _err; + if (code) goto end; taosWLockLatch(&pTq->lock); code = tqMetaSaveHandle(pTq, handle.subKey, &handle); taosWUnLockLatch(&pTq->lock); - if (code < 0) goto _err; - tDecoderClear(pDecoder); - return code; - -_err: +end: tDecoderClear(pDecoder); - tqError("vgId:%d, vnode snapshot tq write failed since %s", TD_VID(pTq->pVnode), tstrerror(code)); + tqDestroyTqHandle(&handle); + tqInfo("vgId:%d, vnode snapshot tq write result:%d", TD_VID(pTq->pVnode), code); return code; } diff --git a/source/dnode/vnode/src/tq/tqMeta.c b/source/dnode/vnode/src/tq/tqMeta.c index bea63fccb9..4c403dc18f 100644 --- a/source/dnode/vnode/src/tq/tqMeta.c +++ b/source/dnode/vnode/src/tq/tqMeta.c @@ -198,58 +198,51 @@ int32_t tqMetaRestoreCheckInfo(STQ* pTq) { int32_t tqMetaSaveHandle(STQ* pTq, const char* key, const STqHandle* pHandle) { int32_t code; int32_t vlen; + void* buf = NULL; + SEncoder encoder; tEncodeSize(tEncodeSTqHandle, pHandle, vlen, code); if (code < 0) { - return -1; + goto end; } tqDebug("tq save %s(%d) handle consumer:0x%" PRIx64 " epoch:%d vgId:%d", pHandle->subKey, (int32_t)strlen(pHandle->subKey), pHandle->consumerId, pHandle->epoch, TD_VID(pTq->pVnode)); - void* buf = taosMemoryCalloc(1, vlen); + buf = taosMemoryCalloc(1, vlen); if (buf == NULL) { - return -1; + code = TSDB_CODE_OUT_OF_MEMORY; + goto end; } - SEncoder encoder; + tEncoderInit(&encoder, buf, vlen); - if (tEncodeSTqHandle(&encoder, pHandle) < 0) { - tEncoderClear(&encoder); - taosMemoryFree(buf); - return -1; + code = tEncodeSTqHandle(&encoder, pHandle); + if (code < 0) { + goto end; } - TXN* txn; - - if (tdbBegin(pTq->pMetaDB, &txn, tdbDefaultMalloc, tdbDefaultFree, NULL, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED) < - 0) { - tEncoderClear(&encoder); - taosMemoryFree(buf); - return -1; + TXN* txn = NULL; + code = tdbBegin(pTq->pMetaDB, &txn, tdbDefaultMalloc, tdbDefaultFree, NULL, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED); + if (code < 0) { + goto end; } - if (tdbTbUpsert(pTq->pExecStore, key, (int)strlen(key), buf, vlen, txn) < 0) { - tEncoderClear(&encoder); - taosMemoryFree(buf); - return -1; + code = tdbTbUpsert(pTq->pExecStore, key, (int)strlen(key), buf, vlen, txn); + if (code < 0) { + goto end; } - if (tdbCommit(pTq->pMetaDB, txn) < 0) { - tEncoderClear(&encoder); - taosMemoryFree(buf); - return -1; - } - - if (tdbPostCommit(pTq->pMetaDB, txn) < 0) { - tEncoderClear(&encoder); - taosMemoryFree(buf); - return -1; + code = tdbCommit(pTq->pMetaDB, txn); + if (code < 0) { + goto end; } + code = tdbPostCommit(pTq->pMetaDB, txn); +end: tEncoderClear(&encoder); taosMemoryFree(buf); - return 0; + return code; } int32_t tqMetaDeleteHandle(STQ* pTq, const char* key) { @@ -349,15 +342,18 @@ static int buildHandle(STQ* pTq, STqHandle* handle){ static int restoreHandle(STQ* pTq, void* pVal, int vLen, STqHandle* handle){ int32_t vgId = TD_VID(pTq->pVnode); SDecoder decoder; + int32_t code = 0; tDecoderInit(&decoder, (uint8_t*)pVal, vLen); - tDecodeSTqHandle(&decoder, handle); - tDecoderClear(&decoder); - - if(buildHandle(pTq, handle) < 0){ - return -1; - } + code = tDecodeSTqHandle(&decoder, handle); + if (code) goto end; + code = buildHandle(pTq, handle); + if (code) goto end; tqInfo("restoreHandle %s consumer 0x%" PRIx64 " vgId:%d", handle->subKey, handle->consumerId, vgId); - return taosHashPut(pTq->pHandle, handle->subKey, strlen(handle->subKey), handle, sizeof(STqHandle)); + code = taosHashPut(pTq->pHandle, handle->subKey, strlen(handle->subKey), handle, sizeof(STqHandle)); + +end: + tDecoderClear(&decoder); + return code; } int32_t tqCreateHandle(STQ* pTq, SMqRebVgReq* req, STqHandle* handle){ diff --git a/source/libs/stream/src/streamMeta.c b/source/libs/stream/src/streamMeta.c index 652ef7cde7..70371c4add 100644 --- a/source/libs/stream/src/streamMeta.c +++ b/source/libs/stream/src/streamMeta.c @@ -238,6 +238,8 @@ int32_t streamMetaReopen(SStreamMeta* pMeta, int64_t chkpId) { return -1; } } + taosMemoryFree(defaultPath); + taosMemoryFree(newPath); pMeta->streamBackend = streamBackendInit(pMeta->path, pMeta->chkpId); while (pMeta->streamBackend == NULL) { From 86a4bfe28d94ef2e7e3284826f0a176cfdb4fcc8 Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Fri, 22 Sep 2023 10:02:16 +0800 Subject: [PATCH 6/8] fix:memory leak --- source/dnode/vnode/src/tq/tqHandleSnapshot.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/source/dnode/vnode/src/tq/tqHandleSnapshot.c b/source/dnode/vnode/src/tq/tqHandleSnapshot.c index 38f9dcc57a..3ce838ce8b 100644 --- a/source/dnode/vnode/src/tq/tqHandleSnapshot.c +++ b/source/dnode/vnode/src/tq/tqHandleSnapshot.c @@ -174,7 +174,7 @@ int32_t tqSnapWrite(STqSnapWriter* pWriter, uint8_t* pData, uint32_t nData) { STQ* pTq = pWriter->pTq; SDecoder decoder = {0}; SDecoder* pDecoder = &decoder; - STqHandle handle; + STqHandle handle = {0}; tDecoderInit(pDecoder, pData + sizeof(SSnapDataHdr), nData - sizeof(SSnapDataHdr)); code = tDecodeSTqHandle(pDecoder, &handle); From 5555371fbc390d713ba0118873714ccbcd01380b Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Fri, 22 Sep 2023 10:04:53 +0800 Subject: [PATCH 7/8] fix:memory leak --- source/dnode/vnode/src/tq/tqOffsetSnapshot.c | 1 + source/libs/stream/src/streamBackendRocksdb.c | 1 + source/libs/stream/src/streamSnapshot.c | 5 +++++ 3 files changed, 7 insertions(+) diff --git a/source/dnode/vnode/src/tq/tqOffsetSnapshot.c b/source/dnode/vnode/src/tq/tqOffsetSnapshot.c index 85d4dc6c0f..8a7f672e5d 100644 --- a/source/dnode/vnode/src/tq/tqOffsetSnapshot.c +++ b/source/dnode/vnode/src/tq/tqOffsetSnapshot.c @@ -85,6 +85,7 @@ int32_t tqOffsetSnapRead(STqOffsetReader* pReader, uint8_t** ppData) { *ppData = (uint8_t*)buf; pReader->readEnd = 1; + taosCloseFile(&pFile); return 0; } diff --git a/source/libs/stream/src/streamBackendRocksdb.c b/source/libs/stream/src/streamBackendRocksdb.c index 19f4ebbeea..0cc8bfb6fe 100644 --- a/source/libs/stream/src/streamBackendRocksdb.c +++ b/source/libs/stream/src/streamBackendRocksdb.c @@ -905,6 +905,7 @@ int32_t streamBackendTriggerChkp(void* arg, char* dst) { } _ERROR: + taosMemoryFree(ppCf); taosReleaseRef(streamBackendId, backendRid); taosArrayDestroy(refs); return code; diff --git a/source/libs/stream/src/streamSnapshot.c b/source/libs/stream/src/streamSnapshot.c index 8a4500dd86..1e27aec03b 100644 --- a/source/libs/stream/src/streamSnapshot.c +++ b/source/libs/stream/src/streamSnapshot.c @@ -341,11 +341,15 @@ int32_t streamSnapRead(SStreamSnapReader* pReader, uint8_t** ppData, int64_t* si qDebug("%s start to read file %s, current offset:%" PRId64 ", size:%" PRId64 ", file no.%d", STREAM_STATE_TRANSFER, item->name, (int64_t)pHandle->offset, item->size, pHandle->currFileIdx); uint8_t* buf = taosMemoryCalloc(1, sizeof(SStreamSnapBlockHdr) + kBlockSize); + if(buf == NULL){ + return TSDB_CODE_OUT_OF_MEMORY; + } int64_t nread = taosPReadFile(pHandle->fd, buf + sizeof(SStreamSnapBlockHdr), kBlockSize, pHandle->offset); if (nread == -1) { code = TAOS_SYSTEM_ERROR(terrno); qError("%s snap failed to read snap, file name:%s, type:%d,reason:%s", STREAM_STATE_TRANSFER, item->name, item->type, tstrerror(code)); + taosMemoryFree(buf); return -1; } else if (nread > 0 && nread <= kBlockSize) { // left bytes less than kBlockSize @@ -368,6 +372,7 @@ int32_t streamSnapRead(SStreamSnapReader* pReader, uint8_t** ppData, int64_t* si // finish *ppData = NULL; *size = 0; + taosMemoryFree(buf); return 0; } item = taosArrayGet(pHandle->pFileList, pHandle->currFileIdx); From 19806cbcabe6cbe23d27a605a2d47ef999e3e10a Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Fri, 22 Sep 2023 10:12:11 +0800 Subject: [PATCH 8/8] fix:memory leak --- source/dnode/vnode/src/tq/tqStreamStateSnap.c | 1 + 1 file changed, 1 insertion(+) diff --git a/source/dnode/vnode/src/tq/tqStreamStateSnap.c b/source/dnode/vnode/src/tq/tqStreamStateSnap.c index 4a1b3961cd..c4ddaa9e54 100644 --- a/source/dnode/vnode/src/tq/tqStreamStateSnap.c +++ b/source/dnode/vnode/src/tq/tqStreamStateSnap.c @@ -104,6 +104,7 @@ int32_t streamStateSnapRead(SStreamStateReader* pReader, uint8_t** ppData) { pHdr->type = SNAP_DATA_STREAM_STATE_BACKEND; pHdr->size = len; memcpy(pHdr->data, rowData, len); + taosMemoryFree(rowData); tqDebug("vgId:%d, vnode stream-state snapshot read data success", TD_VID(pReader->pTq->pVnode)); return code;