Merge pull request #20628 from taosdata/fix/liaohj

fix(tmq): wait for 2mins when subscribe topics.

source/client/src/clientTmq.c

@@ -1116,6 +1116,7 @@ _failed:
 }
 
 int32_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) {
+  const int32_t   MAX_RETRY_COUNT = 120 * 2;  // let's wait for 2 mins at most
   const SArray*   container = &topic_list->container;
   int32_t         sz = taosArrayGetSize(container);
   void*           buf = NULL;
@@ -1209,7 +1210,7 @@ int32_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) {
 
   int32_t retryCnt = 0;
   while (TSDB_CODE_MND_CONSUMER_NOT_READY == tmqAskEp(tmq, false)) {
-    if (retryCnt++ > 40) {
+    if (retryCnt++ > MAX_RETRY_COUNT) {
       goto FAIL;
     }
 
@@ -1811,7 +1812,6 @@ static void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
     if (pRspWrapper == NULL) {
       taosReadAllQitems(tmq->mqueue, tmq->qall);
       taosGetQitem(tmq->qall, (void**)&pRspWrapper);
-
       if (pRspWrapper == NULL) {
         return NULL;
       }
@@ -1831,7 +1831,6 @@ static void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
       SMqDataRsp* pDataRsp = &pollRspWrapper->dataRsp;
 
       if (pDataRsp->head.epoch == consumerEpoch) {
-        // todo fix it: race condition
         SMqClientVg* pVg = pollRspWrapper->vgHandle;
 
         // update the epset
@@ -1843,6 +1842,7 @@ static void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
           pVg->epSet = *pollRspWrapper->pEpset;
         }
 
+        // update the local offset value only for the returned values.
         pVg->currentOffset = pDataRsp->rspOffset;
         atomic_store_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE);
 

source/dnode/vnode/src/inc/tq.h

@@ -109,23 +109,18 @@ typedef struct {
 } STqPushEntry;
 
 struct STQ {
-  SVnode* pVnode;
-  char*   path;
-  int64_t walLogLastVer;
-
-  SRWLatch pushLock;
-
-  SHashObj* pPushMgr;    // consumerId -> STqPushEntry
-  SHashObj* pHandle;     // subKey -> STqHandle
-  SHashObj* pCheckInfo;  // topic -> SAlterCheckInfo
-
+  SVnode*         pVnode;
+  char*           path;
+  int64_t         walLogLastVer;
+  SRWLatch        lock;
+  SHashObj*       pPushMgr;    // consumerId -> STqPushEntry
+  SHashObj*       pHandle;     // subKey -> STqHandle
+  SHashObj*       pCheckInfo;  // topic -> SAlterCheckInfo
   STqOffsetStore* pOffsetStore;
-
-  TDB* pMetaDB;
-  TTB* pExecStore;
-  TTB* pCheckStore;
-
-  SStreamMeta* pStreamMeta;
+  TDB*            pMetaDB;
+  TTB*            pExecStore;
+  TTB*            pCheckStore;
+  SStreamMeta*    pStreamMeta;
 };
 
 typedef struct {
@@ -164,7 +159,7 @@ typedef struct {
   int32_t size;
 } STqOffsetHead;
 
-STqOffsetStore* tqOffsetOpen();
+STqOffsetStore* tqOffsetOpen(STQ* pTq);
 void            tqOffsetClose(STqOffsetStore*);
 STqOffset*      tqOffsetRead(STqOffsetStore* pStore, const char* subscribeKey);
 int32_t         tqOffsetWrite(STqOffsetStore* pStore, const STqOffset* pOffset);

source/dnode/vnode/src/tq/tqPush.c

@@ -213,7 +213,7 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver)
 
   if (msgType == TDMT_VND_SUBMIT) {
     // lock push mgr to avoid potential msg lost
-    taosWLockLatch(&pTq->pushLock);
+    taosWLockLatch(&pTq->lock);
 
     int32_t numOfRegisteredPush = taosHashGetSize(pTq->pPushMgr);
     if (numOfRegisteredPush > 0) {
@@ -231,7 +231,7 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver)
         taosArrayDestroy(cachedKeyLens);
 
         // unlock
-        taosWUnLockLatch(&pTq->pushLock);
+        taosWUnLockLatch(&pTq->lock);
         return -1;
       }
 
@@ -320,7 +320,7 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver)
       taosMemoryFree(data);
     }
     // unlock
-    taosWUnLockLatch(&pTq->pushLock);
+    taosWUnLockLatch(&pTq->lock);
   }
 
   if (!tsDisableStream && vnodeIsRoleLeader(pTq->pVnode)) {