xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix:vnode tranform support in tmq

This commit is contained in:
wangmm0220 2023-09-18 19:47:28 +08:00
parent 49a193b328
commit 4b74135350
5 changed files with 105 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
source/dnode/mnode/impl/src/mndSubscribe.c
View File

@ -875,7 +875,12 @@ end:
mndReleaseSubscribe(pMnode, pSub); mndReleaseSubscribe(pMnode, pSub);
mndTransDrop(pTrans); 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) {} void mndCleanupSubscribe(SMnode *pMnode) {}

38
source/dnode/vnode/src/tq/tqCheckInfoSnapshot.c
View File

@ -79,25 +79,25 @@ int32_t tqCheckInfoRead(STqCheckInfoReader* pReader, uint8_t** ppData) {
void* pVal = NULL; void* pVal = NULL;
int32_t kLen = 0; int32_t kLen = 0;
int32_t vLen = 0; int32_t vLen = 0;
SDecoder decoder; // SDecoder decoder;
STqCheckInfo info; // STqCheckInfo info;
*ppData = NULL; // *ppData = NULL;
if (tdbTbcNext(pReader->pCur, &pKey, &kLen, &pVal, &vLen)) { if (tdbTbcNext(pReader->pCur, &pKey, &kLen, &pVal, &vLen)) {
goto _exit; goto _exit;
} }
tDecoderInit(&decoder, (uint8_t*)pVal, vLen); // tDecoderInit(&decoder, (uint8_t*)pVal, vLen);
if (tDecodeSTqCheckInfo(&decoder, &info) < 0) { // if (tDecodeSTqCheckInfo(&decoder, &info) < 0) {
tdbFree(pKey); // tdbFree(pKey);
tdbFree(pVal); // tdbFree(pVal);
code = TSDB_CODE_OUT_OF_MEMORY; // code = TSDB_CODE_OUT_OF_MEMORY;
goto _err; // goto _err;
} // }
tdbFree(pKey); // tdbFree(pKey);
tdbFree(pVal); // tdbFree(pVal);
tDecoderClear(&decoder); // tDecoderClear(&decoder);
//
*ppData = taosMemoryMalloc(sizeof(SSnapDataHdr) + vLen); *ppData = taosMemoryMalloc(sizeof(SSnapDataHdr) + vLen);
if (*ppData == NULL) { if (*ppData == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
@ -109,13 +109,17 @@ int32_t tqCheckInfoRead(STqCheckInfoReader* pReader, uint8_t** ppData) {
pHdr->size = vLen; pHdr->size = vLen;
memcpy(pHdr->data, pVal, 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: _exit:
tdbFree(pKey);
tdbFree(pVal);
tqInfo("vgId:%d, vnode check info tq read data, vLen:%d", TD_VID(pReader->pTq->pVnode), vLen);
return code; return code;
_err: _err:
tdbFree(pKey);
tdbFree(pVal);
tqError("vgId:%d, vnode check info tq read data failed since %s", TD_VID(pReader->pTq->pVnode), tstrerror(code)); tqError("vgId:%d, vnode check info tq read data failed since %s", TD_VID(pReader->pTq->pVnode), tstrerror(code));
return code; return code;
} }

66
source/dnode/vnode/src/tq/tqHandleSnapshot.c
View File

@ -75,31 +75,51 @@ int32_t tqSnapReaderClose(STqSnapReader** ppReader) {
int32_t tqSnapRead(STqSnapReader* pReader, uint8_t** ppData) { int32_t tqSnapRead(STqSnapReader* pReader, uint8_t** ppData) {
int32_t code = 0; int32_t code = 0;
const void* pKey = NULL; void* pKey = NULL;
const void* pVal = NULL; void* pVal = NULL;
int32_t kLen = 0; int32_t kLen = 0;
int32_t vLen = 0; int32_t vLen = 0;
SDecoder decoder; // SDecoder decoder;
STqHandle handle; // STqHandle handle;
*ppData = NULL;
for (;;) {
if (tdbTbcGet(pReader->pCur, &pKey, &kLen, &pVal, &vLen)) {
goto _exit;
}
tDecoderInit(&decoder, (uint8_t*)pVal, vLen); // *ppData = NULL;
tDecodeSTqHandle(&decoder, &handle); if (tdbTbcNext(pReader->pCur, &pKey, &kLen, &pVal, &vLen)) {
tDecoderClear(&decoder); goto _exit;
if (handle.snapshotVer <= pReader->sver && handle.snapshotVer >= pReader->ever) {
tdbTbcMoveToNext(pReader->pCur);
break;
} else {
tdbTbcMoveToNext(pReader->pCur);
}
} }
// 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); *ppData = taosMemoryMalloc(sizeof(SSnapDataHdr) + vLen);
if (*ppData == NULL) { if (*ppData == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
@ -111,13 +131,15 @@ int32_t tqSnapRead(STqSnapReader* pReader, uint8_t** ppData) {
pHdr->size = vLen; pHdr->size = vLen;
memcpy(pHdr->data, pVal, 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: _exit:
tdbFree(pKey);
tdbFree(pVal);
tqInfo("vgId:%d, vnode snapshot tq read data, vLen:%d", TD_VID(pReader->pTq->pVnode), vLen);
return code; return code;
_err: _err:
tdbFree(pKey);
tdbFree(pVal);
tqError("vgId:%d, vnode snapshot tq read data failed since %s", TD_VID(pReader->pTq->pVnode), tstrerror(code)); tqError("vgId:%d, vnode snapshot tq read data failed since %s", TD_VID(pReader->pTq->pVnode), tstrerror(code));
return code; return code;
} }

3
tests/system-test/7-tmq/tmqDropConsumer.py
View File

@ -12,7 +12,7 @@ sys.path.append("./7-tmq")
from tmqCommon import * from tmqCommon import *
class TDTestCase: class TDTestCase:
updatecfgDict = {'debugFlag': 135} # updatecfgDict = {'debugFlag': 135}
def __init__(self): def __init__(self):
self.vgroups = 2 self.vgroups = 2
@ -252,7 +252,6 @@ class TDTestCase:
break break
tdLog.info("all consumers status into 'lost'") tdLog.info("all consumers status into 'lost'")
# drop consumer groups # drop consumer groups
tdLog.info("drop all consumers") tdLog.info("drop all consumers")
for i in range(len(groupIdList)): for i in range(len(groupIdList)):

59
tests/system-test/7-tmq/tmqVnodeTransform.py
View File

@ -49,7 +49,7 @@ class TDTestCase:
'rowsPerTbl': 10000, 'rowsPerTbl': 10000,
'batchNum': 10, 'batchNum': 10,
'startTs': 1640966400000, # 2022-01-01 00:00:00.000 'startTs': 1640966400000, # 2022-01-01 00:00:00.000
'pollDelay': 200, 'pollDelay': 60,
'showMsg': 1, 'showMsg': 1,
'showRow': 1, 'showRow': 1,
'snapshot': 0} '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) tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], wal_retention_period=36000,vgroups=paraDict["vgroups"],replica=self.replicaVar)
tdLog.info("create stb") tdLog.info("create stb")
tmqCom.create_stable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"]) tmqCom.create_stable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"])
tdLog.info("create ctb") # tdLog.info("create ctb")
tmqCom.create_ctable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=paraDict['ctbPrefix'], # tmqCom.create_ctable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=paraDict['ctbPrefix'],
ctbNum=paraDict["ctbNum"],ctbStartIdx=paraDict['ctbStartIdx']) # ctbNum=paraDict["ctbNum"],ctbStartIdx=paraDict['ctbStartIdx'])
tdLog.info("insert data") # tdLog.info("insert data")
tmqCom.insert_data_interlaceByMultiTbl(tsql=tdSql,dbName=paraDict["dbName"],ctbPrefix=paraDict["ctbPrefix"], # tmqCom.insert_data_interlaceByMultiTbl(tsql=tdSql,dbName=paraDict["dbName"],ctbPrefix=paraDict["ctbPrefix"],
ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"], # ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"],
startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx']) # 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.stop(1)
# tdDnodes.start(1) # tdDnodes.start(1)
tdSql.query("flush database %s"%(paraDict['dbName'])) # tdSql.query("flush database %s"%(paraDict['dbName']))
return return
def tmqCase1(self): def tmqCase1(self):
@ -87,13 +87,13 @@ class TDTestCase:
'tagPrefix': 't', '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}], '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}], '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, 'ctbStartIdx': 0,
'ctbNum': 10, 'ctbNum': 10,
'rowsPerTbl': 10000, 'rowsPerTbl': 10000,
'batchNum': 10, 'batchNum': 10,
'startTs': 1640966400000, # 2022-01-01 00:00:00.000 'startTs': 1640966400000, # 2022-01-01 00:00:00.000
'pollDelay': 200, 'pollDelay': 60,
'showMsg': 1, 'showMsg': 1,
'showRow': 1, 'showRow': 1,
'snapshot': 0} 'snapshot': 0}
@ -102,12 +102,6 @@ class TDTestCase:
paraDict['ctbNum'] = self.ctbNum paraDict['ctbNum'] = self.ctbNum
paraDict['rowsPerTbl'] = self.rowsPerTbl 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'] topicNameList = ['topic1']
# expectRowsList = [] # expectRowsList = []
tmqCom.initConsumerTable("cdb", self.replicaVar) tmqCom.initConsumerTable("cdb", self.replicaVar)
@ -124,7 +118,7 @@ class TDTestCase:
# init consume info, and start tmq_sim, then check consume result # init consume info, and start tmq_sim, then check consume result
tdLog.info("insert consume info to consume processor") tdLog.info("insert consume info to consume processor")
consumerId = 0 consumerId = 0
expectrowcnt = paraDict["rowsPerTbl"] * paraDict["ctbNum"] * 2 expectrowcnt = paraDict["rowsPerTbl"] * paraDict["ctbNum"]
topicList = topicNameList[0] topicList = topicNameList[0]
ifcheckdata = 1 ifcheckdata = 1
ifManualCommit = 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']) 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("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 #restart dnode & remove wal
tdDnodes = cluster.dnodes tdDnodes = cluster.dnodes
tdSql.query("select * from information_schema.ins_vnodes") tdSql.query("select * from information_schema.ins_vnodes")
@ -145,13 +148,13 @@ class TDTestCase:
vnodeId = result[1] vnodeId = result[1]
tdDnodes[dnodeId - 1].stoptaosd() tdDnodes[dnodeId - 1].stoptaosd()
time.sleep(2) time.sleep(1)
dataPath = self.getDataPath() dataPath = self.getDataPath()
dataPath = dataPath%(dnodeId,vnodeId) dataPath = dataPath%(dnodeId,vnodeId)
os.system('rm -rf ' + dataPath) os.system('rm -rf ' + dataPath)
tdLog.debug("dataPath:%s"%dataPath) tdLog.debug("dataPath:%s"%dataPath)
tdDnodes[dnodeId - 1].starttaosd() tdDnodes[dnodeId - 1].starttaosd()
time.sleep(2) time.sleep(1)
break break
tdLog.debug("restart dnode ok") tdLog.debug("restart dnode ok")
@ -171,18 +174,22 @@ class TDTestCase:
break break
redistributeSql = "redistribute vgroup %d dnode %d" %(vnodeId, dnodesList[0]) redistributeSql = "redistribute vgroup %d dnode %d" %(vnodeId, dnodesList[0])
tdLog.debug("redistributeSql:%s"%(redistributeSql)) tdLog.debug("redistributeSql:%s"%(redistributeSql))
time.sleep(10)
tdSql.query(redistributeSql) tdSql.query(redistributeSql)
tdLog.debug("redistributeSql ok") tdLog.debug("redistributeSql ok")
tmqCom.getStartConsumeNotifyFromTmqsim() tdLog.info("create ctb2")
tmqCom.getStartCommitNotifyFromTmqsim() 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() pInsertThread.join()
pInsertThread1.join()
expectRows = 1 expectRows = 1
resultList = tmqCom.selectConsumeResult(expectRows) 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.info("expect consume rows: %d, act consume rows: %d"%(expectrowcnt, resultList[0]))
tdLog.exit("%d tmq consume rows error!"%consumerId) tdLog.exit("%d tmq consume rows error!"%consumerId)