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

fix:[TD-31017]process return value in client for tmq

This commit is contained in:
wangmm0220 2024-07-23 16:29:10 +08:00
parent 3bd21d446d
commit 37fd3d4fa8
1 changed files with 7 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
source/client/src/clientTmq.c
View File

@ -697,7 +697,7 @@ static void asyncCommitAllOffsets(tmq_t* tmq, tmq_commit_cb* pCommitFp, void* us
for (int32_t i = 0; i < numOfTopics; i++) { for (int32_t i = 0; i < numOfTopics; i++) {
SMqClientTopic* pTopic = taosArrayGet(tmq->clientTopics, i); SMqClientTopic* pTopic = taosArrayGet(tmq->clientTopics, i);
if (pTopic != NULL) { if (pTopic == NULL) {
code = TSDB_CODE_TMQ_INVALID_TOPIC; code = TSDB_CODE_TMQ_INVALID_TOPIC;
taosRUnLockLatch(&tmq->lock); taosRUnLockLatch(&tmq->lock);
goto end; goto end;
@ -707,7 +707,7 @@ static void asyncCommitAllOffsets(tmq_t* tmq, tmq_commit_cb* pCommitFp, void* us
numOfVgroups); numOfVgroups);
for (int32_t j = 0; j < numOfVgroups; j++) { for (int32_t j = 0; j < numOfVgroups; j++) {
SMqClientVg* pVg = taosArrayGet(pTopic->vgs, j); SMqClientVg* pVg = taosArrayGet(pTopic->vgs, j);
if (pVg != NULL) { if (pVg == NULL) {
code = TSDB_CODE_INVALID_PARA; code = TSDB_CODE_INVALID_PARA;
taosRUnLockLatch(&tmq->lock); taosRUnLockLatch(&tmq->lock);
goto end; goto end;
@ -2446,6 +2446,8 @@ static int32_t innerClose(tmq_t* tmq){
} }
int32_t tmq_unsubscribe(tmq_t* tmq) { int32_t tmq_unsubscribe(tmq_t* tmq) {
if (tmq == NULL) return TSDB_CODE_INVALID_PARA; if (tmq == NULL) return TSDB_CODE_INVALID_PARA;
tscInfo("consumer:0x%" PRIx64 " start to unsubscribe consumer, status:%d", tmq->consumerId, tmq->status);
int32_t code = 0; int32_t code = 0;
if (atomic_load_8(&tmq->status) != TMQ_CONSUMER_STATUS__CLOSED) { if (atomic_load_8(&tmq->status) != TMQ_CONSUMER_STATUS__CLOSED) {
code = innerClose(tmq); code = innerClose(tmq);
@ -2466,10 +2468,12 @@ int32_t tmq_consumer_close(tmq_t* tmq) {
code = innerClose(tmq); code = innerClose(tmq);
if(code == 0){ if(code == 0){
atomic_store_8(&tmq->status, TMQ_CONSUMER_STATUS__CLOSED); atomic_store_8(&tmq->status, TMQ_CONSUMER_STATUS__CLOSED);
(void)taosRemoveRef(tmqMgmt.rsetId, tmq->refId);
} }
} }
if (code == 0){
(void)taosRemoveRef(tmqMgmt.rsetId, tmq->refId);
}
return code; return code;
} }