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

fix:[TD-30915]tmq exit elegantly

This commit is contained in:
wangmm0220 2024-07-08 09:12:50 +08:00
parent c3e09d0852
commit fc544705fe
3 changed files with 29 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
source/client/inc/clientInt.h
View File

@ -454,6 +454,8 @@ enum {
void sqlReqLog(int64_t rid, bool killed, int32_t code, int8_t type);
void tmqMgmtClose(void);
#ifdef __cplusplus
}
#endif

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

@ -57,8 +57,6 @@ void taos_cleanup(void) {
}
monitorClose();
taosHashCleanup(appInfo.pInstMap);
taosHashCleanup(appInfo.pInstMapByClusterId);
tscStopCrashReport();
hbMgrCleanUp();
@ -88,6 +86,9 @@ void taos_cleanup(void) {
tscInfo("all local resources released");
taosCleanupCfg();
taosCloseLog();
tmqMgmtClose();
taosHashCleanup(appInfo.pInstMap);
taosHashCleanup(appInfo.pInstMapByClusterId);
}
static setConfRet taos_set_config_imp(const char *config) {

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

@ -29,7 +29,6 @@
#define DEFAULT_HEARTBEAT_INTERVAL 3000
struct SMqMgmt {
int8_t inited;
tmr_h timer;
int32_t rsetId;
};
@ -1066,6 +1065,18 @@ void tmqFreeImpl(void* handle) {
taos_close_internal(tmq->pTscObj);
taosMemoryFree(tmq);
if(tmq->commitTimer) {
taosTmrStopA(tmq->commitTimer);
tmq->commitTimer = NULL;
}
if(tmq->epTimer) {
taosTmrStopA(tmq->epTimer);
tmq->epTimer = NULL;
}
if(tmq->hbLiveTimer) {
taosTmrStopA(tmq->hbLiveTimer);
tmq->hbLiveTimer = NULL;
}
tscDebug("consumer:0x%" PRIx64 " closed", id);
}
@ -1083,6 +1094,18 @@ static void tmqMgmtInit(void) {
}
}
void tmqMgmtClose(void) {
if (tmqMgmt.timer) {
taosTmrCleanUp(tmqMgmt.timer);
tmqMgmt.timer = NULL;
}
if (tmqMgmt.rsetId >= 0) {
taosCloseRef(tmqMgmt.rsetId);
tmqMgmt.rsetId = -1;
}
}
#define SET_ERROR_MSG_TMQ(MSG) \
if (errstr != NULL) snprintf(errstr, errstrLen, MSG);