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

feat:[TD-30270] checkout subscribe logic if unsubscribe twice

This commit is contained in:
wangmm0220 2024-09-04 11:35:01 +08:00
parent c532e9367d
commit 0336d3aa64
2 changed files with 7 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
source/dnode/mnode/impl/src/mndConsumer.c
View File

@ -581,6 +581,10 @@ int32_t mndProcessSubscribeReq(SRpcMsg *pMsg) {
if(unSubscribe){
SMqConsumerObj *pConsumerTmp = NULL;
MND_TMQ_RETURN_CHECK(mndAcquireConsumer(pMnode, subscribe.consumerId, &pConsumerTmp));
if (taosArrayGetSize(pConsumerTmp->assignedTopics) == 0){
mndReleaseConsumer(pMnode, pConsumerTmp);
return 0;
}
mndReleaseConsumer(pMnode, pConsumerTmp);
}
MND_TMQ_RETURN_CHECK(checkAndSortTopic(pMnode, subscribe.topicNames));

9
tests/system-test/7-tmq/subscribeStb3.py
View File

@ -20,12 +20,9 @@ class actionType(Enum):
class TDTestCase:
hostname = socket.gethostname()
#rpcDebugFlagVal = '143'
#clientCfgDict = {'serverPort': '', 'firstEp': '', 'secondEp':'', 'rpcDebugFlag':'135', 'fqdn':''}
#clientCfgDict["rpcDebugFlag"] = rpcDebugFlagVal
#updatecfgDict = {'clientCfg': {}, 'serverPort': '', 'firstEp': '', 'secondEp':'', 'rpcDebugFlag':'135', 'fqdn':''}
#updatecfgDict["rpcDebugFlag"] = rpcDebugFlagVal
#print ("===================: ", updatecfgDict)
clientCfgDict = {'debugFlag': 135}
updatecfgDict = {'debugFlag': 135, 'asynclog': 0}
updatecfgDict["clientCfg"] = clientCfgDict
def init(self, conn, logSql, replicaVar=1):
self.replicaVar = int(replicaVar)