From 4b7413535059aa74b5b5bcc474e5876a7f1904b4 Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Mon, 18 Sep 2023 19:47:28 +0800 Subject: [PATCH] fix:vnode tranform support in tmq --- source/dnode/mnode/impl/src/mndSubscribe.c | 7 +- .../dnode/vnode/src/tq/tqCheckInfoSnapshot.c | 38 ++++++----- source/dnode/vnode/src/tq/tqHandleSnapshot.c | 66 ++++++++++++------- tests/system-test/7-tmq/tmqDropConsumer.py | 3 +- tests/system-test/7-tmq/tmqVnodeTransform.py | 59 +++++++++-------- 5 files changed, 105 insertions(+), 68 deletions(-) diff --git a/source/dnode/mnode/impl/src/mndSubscribe.c b/source/dnode/mnode/impl/src/mndSubscribe.c index a8cd62db09..54f1024b23 100644 --- a/source/dnode/mnode/impl/src/mndSubscribe.c +++ b/source/dnode/mnode/impl/src/mndSubscribe.c @@ -875,7 +875,12 @@ end: mndReleaseSubscribe(pMnode, pSub); mndTransDrop(pTrans); - return code; + if (code != 0) { + mError("cgroup %s on topic:%s, failed to drop", dropReq.cgroup, dropReq.topic); + return code; + } + + return TSDB_CODE_ACTION_IN_PROGRESS; } void mndCleanupSubscribe(SMnode *pMnode) {} diff --git a/source/dnode/vnode/src/tq/tqCheckInfoSnapshot.c b/source/dnode/vnode/src/tq/tqCheckInfoSnapshot.c index 18f8f0fecc..346dcf5b50 100644 --- a/source/dnode/vnode/src/tq/tqCheckInfoSnapshot.c +++ b/source/dnode/vnode/src/tq/tqCheckInfoSnapshot.c @@ -79,25 +79,25 @@ int32_t tqCheckInfoRead(STqCheckInfoReader* pReader, uint8_t** ppData) { void* pVal = NULL; int32_t kLen = 0; int32_t vLen = 0; - SDecoder decoder; - STqCheckInfo info; +// SDecoder decoder; +// STqCheckInfo info; - *ppData = NULL; +// *ppData = NULL; if (tdbTbcNext(pReader->pCur, &pKey, &kLen, &pVal, &vLen)) { goto _exit; } - tDecoderInit(&decoder, (uint8_t*)pVal, vLen); - if (tDecodeSTqCheckInfo(&decoder, &info) < 0) { - tdbFree(pKey); - tdbFree(pVal); - code = TSDB_CODE_OUT_OF_MEMORY; - goto _err; - } - tdbFree(pKey); - tdbFree(pVal); - tDecoderClear(&decoder); - +// tDecoderInit(&decoder, (uint8_t*)pVal, vLen); +// if (tDecodeSTqCheckInfo(&decoder, &info) < 0) { +// tdbFree(pKey); +// tdbFree(pVal); +// code = TSDB_CODE_OUT_OF_MEMORY; +// goto _err; +// } +// tdbFree(pKey); +// tdbFree(pVal); +// tDecoderClear(&decoder); +// *ppData = taosMemoryMalloc(sizeof(SSnapDataHdr) + vLen); if (*ppData == NULL) { code = TSDB_CODE_OUT_OF_MEMORY; @@ -109,13 +109,17 @@ int32_t tqCheckInfoRead(STqCheckInfoReader* pReader, uint8_t** ppData) { pHdr->size = vLen; memcpy(pHdr->data, pVal, vLen); - tqInfo("vgId:%d, vnode check info tq read data, topic: %s vLen:%d", TD_VID(pReader->pTq->pVnode), - info.topic, vLen); - _exit: + tdbFree(pKey); + tdbFree(pVal); + + tqInfo("vgId:%d, vnode check info tq read data, vLen:%d", TD_VID(pReader->pTq->pVnode), vLen); return code; _err: + tdbFree(pKey); + tdbFree(pVal); + tqError("vgId:%d, vnode check info tq read data failed since %s", TD_VID(pReader->pTq->pVnode), tstrerror(code)); return code; } diff --git a/source/dnode/vnode/src/tq/tqHandleSnapshot.c b/source/dnode/vnode/src/tq/tqHandleSnapshot.c index 23015ddf39..93b8a0398a 100644 --- a/source/dnode/vnode/src/tq/tqHandleSnapshot.c +++ b/source/dnode/vnode/src/tq/tqHandleSnapshot.c @@ -75,31 +75,51 @@ int32_t tqSnapReaderClose(STqSnapReader** ppReader) { int32_t tqSnapRead(STqSnapReader* pReader, uint8_t** ppData) { int32_t code = 0; - const void* pKey = NULL; - const void* pVal = NULL; + void* pKey = NULL; + void* pVal = NULL; int32_t kLen = 0; int32_t vLen = 0; - SDecoder decoder; - STqHandle handle; +// SDecoder decoder; +// STqHandle handle; - *ppData = NULL; - for (;;) { - if (tdbTbcGet(pReader->pCur, &pKey, &kLen, &pVal, &vLen)) { - goto _exit; - } - tDecoderInit(&decoder, (uint8_t*)pVal, vLen); - tDecodeSTqHandle(&decoder, &handle); - tDecoderClear(&decoder); - - if (handle.snapshotVer <= pReader->sver && handle.snapshotVer >= pReader->ever) { - tdbTbcMoveToNext(pReader->pCur); - break; - } else { - tdbTbcMoveToNext(pReader->pCur); - } +// *ppData = NULL; + if (tdbTbcNext(pReader->pCur, &pKey, &kLen, &pVal, &vLen)) { + goto _exit; } +// tDecoderInit(&decoder, (uint8_t*)pVal, vLen); +// if (tDecodeSTqCheckInfo(&decoder, &info) < 0) { +// tdbFree(pKey); +// tdbFree(pVal); +// code = TSDB_CODE_OUT_OF_MEMORY; +// goto _err; +// } +// tdbFree(pKey); +// tdbFree(pVal); +// tDecoderClear(&decoder); + +// *ppData = NULL; +// for (;;) { +// if (tdbTbcGet(pReader->pCur, &pKey, &kLen, &pVal, &vLen)) { +// goto _exit; +// } +// +// tDecoderInit(&decoder, (uint8_t*)pVal, vLen); +// tDecodeSTqHandle(&decoder, &handle); +// tDecoderClear(&decoder); +// +// tqInfo("vgId:%d, vnode snapshot tq start read data, version:%" PRId64 " subKey: %s vLen:%d, sver:%"PRId64 " , ever:%" PRId64, TD_VID(pReader->pTq->pVnode), +// handle.snapshotVer, handle.subKey, vLen, +// pReader->sver, pReader->ever); +// if (handle.snapshotVer <= pReader->sver && handle.snapshotVer >= pReader->ever) { +// tdbTbcMoveToNext(pReader->pCur); +// break; +// } else { +// tdbTbcMoveToNext(pReader->pCur); +// } +// } + *ppData = taosMemoryMalloc(sizeof(SSnapDataHdr) + vLen); if (*ppData == NULL) { code = TSDB_CODE_OUT_OF_MEMORY; @@ -111,13 +131,15 @@ int32_t tqSnapRead(STqSnapReader* pReader, uint8_t** ppData) { pHdr->size = vLen; memcpy(pHdr->data, pVal, vLen); - tqInfo("vgId:%d, vnode snapshot tq read data, version:%" PRId64 " subKey: %s vLen:%d", TD_VID(pReader->pTq->pVnode), - handle.snapshotVer, handle.subKey, vLen); - _exit: + tdbFree(pKey); + tdbFree(pVal); + tqInfo("vgId:%d, vnode snapshot tq read data, vLen:%d", TD_VID(pReader->pTq->pVnode), vLen); return code; _err: + tdbFree(pKey); + tdbFree(pVal); tqError("vgId:%d, vnode snapshot tq read data failed since %s", TD_VID(pReader->pTq->pVnode), tstrerror(code)); return code; } diff --git a/tests/system-test/7-tmq/tmqDropConsumer.py b/tests/system-test/7-tmq/tmqDropConsumer.py index 137b5c6584..e3e9906ecf 100644 --- a/tests/system-test/7-tmq/tmqDropConsumer.py +++ b/tests/system-test/7-tmq/tmqDropConsumer.py @@ -12,7 +12,7 @@ sys.path.append("./7-tmq") from tmqCommon import * class TDTestCase: - updatecfgDict = {'debugFlag': 135} + # updatecfgDict = {'debugFlag': 135} def __init__(self): self.vgroups = 2 @@ -252,7 +252,6 @@ class TDTestCase: break tdLog.info("all consumers status into 'lost'") - # drop consumer groups tdLog.info("drop all consumers") for i in range(len(groupIdList)): diff --git a/tests/system-test/7-tmq/tmqVnodeTransform.py b/tests/system-test/7-tmq/tmqVnodeTransform.py index 85b9172646..fea459350c 100644 --- a/tests/system-test/7-tmq/tmqVnodeTransform.py +++ b/tests/system-test/7-tmq/tmqVnodeTransform.py @@ -49,7 +49,7 @@ class TDTestCase: 'rowsPerTbl': 10000, 'batchNum': 10, 'startTs': 1640966400000, # 2022-01-01 00:00:00.000 - 'pollDelay': 200, + 'pollDelay': 60, 'showMsg': 1, 'showRow': 1, 'snapshot': 0} @@ -62,18 +62,18 @@ class TDTestCase: 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("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") + # 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'])) + # tdSql.query("flush database %s"%(paraDict['dbName'])) return def tmqCase1(self): @@ -87,13 +87,13 @@ class TDTestCase: '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': 'ctbn', + 'ctbPrefix': 'ctb', 'ctbStartIdx': 0, 'ctbNum': 10, 'rowsPerTbl': 10000, 'batchNum': 10, 'startTs': 1640966400000, # 2022-01-01 00:00:00.000 - 'pollDelay': 200, + 'pollDelay': 60, 'showMsg': 1, 'showRow': 1, 'snapshot': 0} @@ -102,12 +102,6 @@ class TDTestCase: paraDict['ctbNum'] = self.ctbNum paraDict['rowsPerTbl'] = self.rowsPerTbl - 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") - pInsertThread = tmqCom.asyncInsertDataByInterlace(paraDict) - topicNameList = ['topic1'] # expectRowsList = [] tmqCom.initConsumerTable("cdb", self.replicaVar) @@ -124,7 +118,7 @@ class TDTestCase: # 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 + expectrowcnt = paraDict["rowsPerTbl"] * paraDict["ctbNum"] topicList = topicNameList[0] ifcheckdata = 1 ifManualCommit = 1 @@ -135,6 +129,15 @@ class TDTestCase: 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 tdDnodes = cluster.dnodes tdSql.query("select * from information_schema.ins_vnodes") @@ -145,13 +148,13 @@ class TDTestCase: vnodeId = result[1] tdDnodes[dnodeId - 1].stoptaosd() - time.sleep(2) + 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(2) + time.sleep(1) break tdLog.debug("restart dnode ok") @@ -171,18 +174,22 @@ class TDTestCase: break redistributeSql = "redistribute vgroup %d dnode %d" %(vnodeId, dnodesList[0]) tdLog.debug("redistributeSql:%s"%(redistributeSql)) - time.sleep(10) tdSql.query(redistributeSql) tdLog.debug("redistributeSql ok") - tmqCom.getStartConsumeNotifyFromTmqsim() - tmqCom.getStartCommitNotifyFromTmqsim() - + 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 != resultList[0]: + if expectrowcnt >= resultList[0]: tdLog.info("expect consume rows: %d, act consume rows: %d"%(expectrowcnt, resultList[0])) tdLog.exit("%d tmq consume rows error!"%consumerId)