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

fix:[TD-31115] modify trans conflict level in tmq

This commit is contained in:
wangmm0220 2024-08-05 22:58:10 +08:00
parent d0c101f605
commit 3a94a22801
1 changed files with 5 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -565,13 +565,16 @@ int32_t mndProcessSubscribeReq(SRpcMsg *pMsg) {
SCMSubscribeReq subscribe = {0}; SCMSubscribeReq subscribe = {0};
MND_TMQ_RETURN_CHECK(tDeserializeSCMSubscribeReq(msgStr, &subscribe, pMsg->contLen)); MND_TMQ_RETURN_CHECK(tDeserializeSCMSubscribeReq(msgStr, &subscribe, pMsg->contLen));
if(taosArrayGetSize(subscribe.topicNames) == 0){ bool ubSubscribe = (taosArrayGetSize(subscribe.topicNames) == 0);
if(ubSubscribe){
SMqConsumerObj *pConsumerTmp = NULL; SMqConsumerObj *pConsumerTmp = NULL;
MND_TMQ_RETURN_CHECK(mndAcquireConsumer(pMnode, subscribe.consumerId, &pConsumerTmp)); MND_TMQ_RETURN_CHECK(mndAcquireConsumer(pMnode, subscribe.consumerId, &pConsumerTmp));
mndReleaseConsumer(pMnode, pConsumerTmp); mndReleaseConsumer(pMnode, pConsumerTmp);
} }
MND_TMQ_RETURN_CHECK(checkAndSortTopic(pMnode, subscribe.topicNames)); MND_TMQ_RETURN_CHECK(checkAndSortTopic(pMnode, subscribe.topicNames));
pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pMsg, "subscribe"); pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY,
(ubSubscribe ? TRN_CONFLICT_NOTHING :TRN_CONFLICT_DB_INSIDE),
pMsg, "subscribe");
MND_TMQ_NULL_CHECK(pTrans); MND_TMQ_NULL_CHECK(pTrans);
MND_TMQ_RETURN_CHECK(validateTopics(pTrans, &subscribe, pMnode, pMsg->info.conn.user)); MND_TMQ_RETURN_CHECK(validateTopics(pTrans, &subscribe, pMnode, pMsg->info.conn.user));