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

fix:[TD-23972] push subscribe msg to vnode even though consumer not change

This commit is contained in:
wangmm0220 2023-05-19 15:14:39 +08:00
parent 238254e49e
commit f7ab8dabf0
2 changed files with 32 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -133,10 +133,10 @@ static int32_t mndBuildSubChangeReq(void **pBuf, int32_t *pLen, const SMqSubscri
static int32_t mndPersistSubChangeVgReq(SMnode *pMnode, STrans *pTrans, const SMqSubscribeObj *pSub,
const SMqRebOutputVg *pRebVg) {
if (pRebVg->oldConsumerId == pRebVg->newConsumerId) {
terrno = TSDB_CODE_MND_INVALID_SUB_OPTION;
return -1;
}
// if (pRebVg->oldConsumerId == pRebVg->newConsumerId) {
// terrno = TSDB_CODE_MND_INVALID_SUB_OPTION;
// return -1;
// }
void *buf;
int32_t tlen;
@ -269,6 +269,18 @@ static void addUnassignedVgroups(SMqRebOutputObj *pOutput, SHashObj *pHash) {
}
}
static void putNoTransferToOutput(SMqRebOutputObj *pOutput, SMqConsumerEp *pConsumerEp){
for(int i = 0; i < taosArrayGetSize(pConsumerEp->vgs); i++){
SMqVgEp *pVgEp = (SMqVgEp *)taosArrayGetP(pConsumerEp->vgs, i);
SMqRebOutputVg outputVg = {
.oldConsumerId = pConsumerEp->consumerId,
.newConsumerId = pConsumerEp->consumerId,
.pVgEp = pVgEp,
};
taosArrayPush(pOutput->rebVgs, &outputVg);
}
}
static void transferVgroupsForConsumers(SMqRebOutputObj *pOutput, SHashObj *pHash, int32_t minVgCnt,
int32_t imbConsumerNum) {
const char *pSubKey = pOutput->pSub->key;
@ -318,6 +330,7 @@ static void transferVgroupsForConsumers(SMqRebOutputObj *pOutput, SHashObj *pHas
}
}
}
putNoTransferToOutput(pOutput, pConsumerEp);
}
}

28
source/dnode/vnode/src/tq/tq.c
View File

@ -569,20 +569,22 @@ int32_t tqProcessSubscribeReq(STQ* pTq, int64_t sversion, char* msg, int32_t msg
req.newConsumerId);
atomic_store_64(&pHandle->consumerId, req.newConsumerId);
// atomic_store_32(&pHandle->epoch, 0);
// kill executing task
qTaskInfo_t pTaskInfo = pHandle->execHandle.task;
if (pTaskInfo != NULL) {
qKillTask(pTaskInfo, TSDB_CODE_SUCCESS);
}
if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
qStreamCloseTsdbReader(pTaskInfo);
}
// remove if it has been register in the push manager, and return one empty block to consumer
taosWLockLatch(&pTq->lock);
tqUnregisterPushHandle(pTq, pHandle);
taosWUnLockLatch(&pTq->lock);
ret = tqMetaSaveHandle(pTq, req.subKey, pHandle);
}
// kill executing task
qTaskInfo_t pTaskInfo = pHandle->execHandle.task;
if (pTaskInfo != NULL) {
qKillTask(pTaskInfo, TSDB_CODE_SUCCESS);
}
taosWLockLatch(&pTq->lock);
if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
qStreamCloseTsdbReader(pTaskInfo);
}
// remove if it has been register in the push manager, and return one empty block to consumer
tqUnregisterPushHandle(pTq, pHandle);
taosWUnLockLatch(&pTq->lock);
ret = tqMetaSaveHandle(pTq, req.subKey, pHandle);
}
end: