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

Merge branch 'feature/3_liaohj' of https://github.com/taosdata/TDengine into feature/3_liaohj

This commit is contained in:
yihaoDeng 2023-03-04 18:10:15 +08:00
parent 6fe2a9b10d 33f2d07605
commit a0f68b28ad
1 changed files with 2 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1228,7 +1228,7 @@ int32_t tmqPollCb(void* param, SDataBuf* pMsg, int32_t code) {
// in case of consumer mismatch, wait for 500ms and retry // in case of consumer mismatch, wait for 500ms and retry
if (code == TSDB_CODE_TMQ_CONSUMER_MISMATCH) { if (code == TSDB_CODE_TMQ_CONSUMER_MISMATCH) {
taosMsleep(500); // taosMsleep(500);
atomic_store_8(&tmq->status, TMQ_CONSUMER_STATUS__RECOVER); atomic_store_8(&tmq->status, TMQ_CONSUMER_STATUS__RECOVER);
} else if (code == TSDB_CODE_TQ_NO_COMMITTED_OFFSET) { } else if (code == TSDB_CODE_TQ_NO_COMMITTED_OFFSET) {
SMqPollRspWrapper* pRspWrapper = taosAllocateQitem(sizeof(SMqPollRspWrapper), DEF_QITEM, 0); SMqPollRspWrapper* pRspWrapper = taosAllocateQitem(sizeof(SMqPollRspWrapper), DEF_QITEM, 0);
@ -1918,6 +1918,7 @@ TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t timeout) {
// in no topic status, delayed task also need to be processed // in no topic status, delayed task also need to be processed
if (atomic_load_8(&tmq->status) == TMQ_CONSUMER_STATUS__INIT) { if (atomic_load_8(&tmq->status) == TMQ_CONSUMER_STATUS__INIT) {
tscDebug("consumer:0x%" PRIx64 " poll return since consumer is init", tmq->consumerId); tscDebug("consumer:0x%" PRIx64 " poll return since consumer is init", tmq->consumerId);
// sleep for a while
return NULL; return NULL;
} }