From 1d852d4f89273f6a13b8b58bfa092b40ff6d746a Mon Sep 17 00:00:00 2001
From: wangmm0220 <wangmm0220@gmail.com>
Date: Thu, 10 Oct 2024 11:10:34 +0800
Subject: [PATCH] fix:[TS-4592] pollFalg error in multi thread

---
 source/client/src/clientTmq.c | 9 +++++----
 1 file changed, 5 insertions(+), 4 deletions(-)

diff --git a/source/client/src/clientTmq.c b/source/client/src/clientTmq.c
index 42de2925da..fcdc5bc236 100644
--- a/source/client/src/clientTmq.c
+++ b/source/client/src/clientTmq.c
@@ -134,6 +134,7 @@ struct tmq_t {
   // poll info
   int64_t pollCnt;
   int64_t totalRows;
+  int8_t  pollFlag;
 
   // timer
   tmr_h       hbLiveTimer;
@@ -287,7 +288,6 @@ typedef struct {
 static   TdThreadOnce   tmqInit = PTHREAD_ONCE_INIT;  // initialize only once
 volatile int32_t        tmqInitRes = 0;               // initialize rsp code
 static   SMqMgmt        tmqMgmt = {0};
-static   int8_t         pollFlag = 0;
 
 tmq_conf_t* tmq_conf_new() {
   tmq_conf_t* conf = taosMemoryCalloc(1, sizeof(tmq_conf_t));
@@ -977,7 +977,7 @@ void tmqSendHbReq(void* param, void* tmrId) {
   SMqHbReq req = {0};
   req.consumerId = tmq->consumerId;
   req.epoch = tmq->epoch;
-  req.pollFlag = atomic_load_8(&pollFlag);
+  req.pollFlag = atomic_load_8(&tmq->pollFlag);
   req.topics = taosArrayInit(taosArrayGetSize(tmq->clientTopics), sizeof(TopicOffsetRows));
   if (req.topics == NULL) {
     goto END;
@@ -1057,7 +1057,7 @@ void tmqSendHbReq(void* param, void* tmrId) {
   if (code != 0) {
     tqErrorC("tmqSendHbReq asyncSendMsgToServer failed");
   }
-  (void)atomic_val_compare_exchange_8(&pollFlag, 1, 0);
+  (void)atomic_val_compare_exchange_8(&tmq->pollFlag, 1, 0);
 
 END:
   tDestroySMqHbReq(&req);
@@ -1640,6 +1640,7 @@ tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
   pTmq->status = TMQ_CONSUMER_STATUS__INIT;
   pTmq->pollCnt = 0;
   pTmq->epoch = 0;
+  pTmq->pollFlag = 0;
 
   // set conf
   tstrncpy(pTmq->clientId, conf->clientId, TSDB_CLIENT_ID_LEN);
@@ -2441,7 +2442,7 @@ TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t timeout) {
     return NULL;
   }
 
-  (void)atomic_val_compare_exchange_8(&pollFlag, 0, 1);
+  (void)atomic_val_compare_exchange_8(&tmq->pollFlag, 0, 1);
 
   while (1) {
     tmqHandleAllDelayedTask(tmq);