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

fix:[TD-30883]send hb before close in tmq

This commit is contained in:
wangmm0220 2024-07-08 15:49:00 +08:00
parent 7ff7ef1d73
commit 2176fa3b1a
1 changed files with 3 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -100,6 +100,7 @@ struct tmq_t {
int64_t totalRows; int64_t totalRows;
// timer // timer
tmr_h hbLiveTimer;
tmr_h epTimer; tmr_h epTimer;
tmr_h commitTimer; tmr_h commitTimer;
STscObj* pTscObj; // connection STscObj* pTscObj; // connection
@ -878,7 +879,7 @@ void tmqSendHbReq(void* param, void* tmrId) {
OVER: OVER:
tDestroySMqHbReq(&req); tDestroySMqHbReq(&req);
if(tmrId != NULL){ if(tmrId != NULL){
taosTmrReset(tmqSendHbReq, DEFAULT_HEARTBEAT_INTERVAL, param, tmqMgmt.timer, &tmrId); taosTmrReset(tmqSendHbReq, DEFAULT_HEARTBEAT_INTERVAL, param, tmqMgmt.timer, &tmq->hbLiveTimer);
} }
taosReleaseRef(tmqMgmt.rsetId, refId); taosReleaseRef(tmqMgmt.rsetId, refId);
} }
@ -1163,7 +1164,7 @@ tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
goto _failed; goto _failed;
} }
taosTmrStart(tmqSendHbReq, DEFAULT_HEARTBEAT_INTERVAL, (void*)pTmq->refId, tmqMgmt.timer); pTmq->hbLiveTimer = taosTmrStart(tmqSendHbReq, DEFAULT_HEARTBEAT_INTERVAL, (void*)pTmq->refId, tmqMgmt.timer);
char buf[TSDB_OFFSET_LEN] = {0}; char buf[TSDB_OFFSET_LEN] = {0};
STqOffsetVal offset = {.type = pTmq->resetOffsetCfg}; STqOffsetVal offset = {.type = pTmq->resetOffsetCfg};