Merge pull request #21499 from taosdata/mark/tmq

fix:put poll to push manager if wal not exist when offset is latest

include/libs/wal/wal.h

@@ -205,7 +205,8 @@ int32_t walFetchHead(SWalReader *pRead, int64_t ver, SWalCkHead *pHead);
 int32_t walFetchBody(SWalReader *pRead, SWalCkHead **ppHead);
 int32_t walSkipFetchBody(SWalReader *pRead, const SWalCkHead *pHead);
 
-SWalRef *walRefFirstVer(SWal *, SWalRef *);
+void walRefFirstVer(SWal *, SWalRef *);
+void walRefLastVer(SWal *, SWalRef *);
 SWalRef *walRefCommittedVer(SWal *);
 
 SWalRef *walOpenRef(SWal *);

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

@@ -106,7 +106,6 @@ typedef struct {
 //  STqPushHandle pushHandle;    // push
   STqExecHandle execHandle;    // exec
   SRpcMsg*      msg;
-  int32_t       noDataPollCnt;
   tq_handle_status        status;
 } STqHandle;
 

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

@@ -377,7 +377,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
     }
     taosWUnLockLatch(&pTq->lock);
 
-    tqDebug("tmq poll: consumer:0x%" PRIx64 "vgId:%d, topic:%s, subscription is executing, wait for 5ms and retry, pHandle:%p", consumerId, vgId, req.subKey, pHandle);
+    tqDebug("tmq poll: consumer:0x%" PRIx64 "vgId:%d, topic:%s, subscription is executing, wait for 10ms and retry, pHandle:%p", consumerId, vgId, req.subKey, pHandle);
     taosMsleep(10);
   }
 
@@ -410,7 +410,7 @@ int32_t tqProcessDeleteSubReq(STQ* pTq, int64_t sversion, char* msg, int32_t msg
   STqHandle* pHandle = taosHashGet(pTq->pHandle, pReq->subKey, strlen(pReq->subKey));
   if (pHandle) {
     while (tqIsHandleExec(pHandle)) {
-      tqDebug("vgId:%d, topic:%s, subscription is executing, wait for 5ms and retry, pHandle:%p", vgId, pHandle->subKey, pHandle);
+      tqDebug("vgId:%d, topic:%s, subscription is executing, wait for 10ms and retry, pHandle:%p", vgId, pHandle->subKey, pHandle);
       taosMsleep(10);
     }
     if (pHandle->pRef) {

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

@@ -16,7 +16,6 @@
 #include "tq.h"
 
 #define IS_OFFSET_RESET_TYPE(_t)  ((_t) < 0)
-#define NO_POLL_CNT 5
 
 static int32_t tqSendMetaPollRsp(STQ* pTq, const SRpcMsg* pMsg, const SMqPollReq* pReq, const SMqMetaRsp* pRsp);
 
@@ -88,15 +87,13 @@ static int32_t tqInitTaosxRsp(STaosxRsp* pRsp, const SMqPollReq* pReq) {
   return 0;
 }
 
-static int32_t extractResetOffsetVal(STqOffsetVal* pOffsetVal, STQ* pTq, STqHandle* pHandle, const SMqPollReq* pRequest,
-                                     SRpcMsg* pMsg, bool* pBlockReturned) {
+static int32_t extractResetOffsetVal(STqOffsetVal* pOffsetVal, STQ* pTq, STqHandle* pHandle, const SMqPollReq* pRequest, SRpcMsg* pMsg, bool* pBlockReturned) {
   uint64_t     consumerId = pRequest->consumerId;
   STqOffsetVal reqOffset = pRequest->reqOffset;
   STqOffset*   pOffset = tqOffsetRead(pTq->pOffsetStore, pRequest->subKey);
   int32_t      vgId = TD_VID(pTq->pVnode);
 
   *pBlockReturned = false;
-
   // In this vnode, data has been polled by consumer for this topic, so let's continue from the last offset value.
   if (pOffset != NULL) {
     *pOffsetVal = pOffset->val;
@@ -119,21 +116,16 @@ static int32_t extractResetOffsetVal(STqOffsetVal* pOffsetVal, STQ* pTq, STqHand
           tqOffsetResetToData(pOffsetVal, 0, 0);
         }
       } else {
-        pHandle->pRef = walRefFirstVer(pTq->pVnode->pWal, pHandle->pRef);
-        if (pHandle->pRef == NULL) {
-          terrno = TSDB_CODE_OUT_OF_MEMORY;
-          return -1;
-        }
-
-        // offset set to previous version when init
+        walRefFirstVer(pTq->pVnode->pWal, pHandle->pRef);
         tqOffsetResetToLog(pOffsetVal, pHandle->pRef->refVer - 1);
       }
     } else if (reqOffset.type == TMQ_OFFSET__RESET_LATEST) {
+      walRefLastVer(pTq->pVnode->pWal, pHandle->pRef);
       if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
         SMqDataRsp dataRsp = {0};
         tqInitDataRsp(&dataRsp, pRequest, pHandle->execHandle.subType);
 
-        tqOffsetResetToLog(&dataRsp.rspOffset, walGetLastVer(pTq->pVnode->pWal));
+        tqOffsetResetToLog(&dataRsp.rspOffset, pHandle->pRef->refVer);
         tqDebug("tmq poll: consumer:0x%" PRIx64 ", subkey %s, vgId:%d, (latest) offset reset to %" PRId64, consumerId,
                 pHandle->subKey, vgId, dataRsp.rspOffset.version);
         int32_t code = tqSendDataRsp(pTq, pMsg, pRequest, &dataRsp, TMQ_MSG_TYPE__POLL_RSP);
@@ -144,7 +136,7 @@ static int32_t extractResetOffsetVal(STqOffsetVal* pOffsetVal, STQ* pTq, STqHand
       } else {
         STaosxRsp taosxRsp = {0};
         tqInitTaosxRsp(&taosxRsp, pRequest);
-        tqOffsetResetToLog(&taosxRsp.rspOffset, walGetLastVer(pTq->pVnode->pWal));
+        tqOffsetResetToLog(&taosxRsp.rspOffset, pHandle->pRef->refVer);
         int32_t code = tqSendDataRsp(pTq, pMsg, pRequest, (SMqDataRsp*)&taosxRsp, TMQ_MSG_TYPE__TAOSX_RSP);
         tDeleteSTaosxRsp(&taosxRsp);
 
@@ -173,25 +165,18 @@ static int32_t extractDataAndRspForNormalSubscribe(STQ* pTq, STqHandle* pHandle,
 
   qSetTaskId(pHandle->execHandle.task, consumerId, pRequest->reqId);
   code = tqScanData(pTq, pHandle, &dataRsp, pOffset);
-  if(code != 0) {
+  if(code != 0 && terrno != TSDB_CODE_WAL_LOG_NOT_EXIST) {
     goto end;
   }
 
   //   till now, all data has been transferred to consumer, new data needs to push client once arrived.
-  if (dataRsp.blockNum == 0 && dataRsp.reqOffset.type == TMQ_OFFSET__LOG &&
-      dataRsp.reqOffset.version == dataRsp.rspOffset.version && pHandle->consumerId == pRequest->consumerId) {
-    if(pHandle->noDataPollCnt >= NO_POLL_CNT){  // send poll result to client if no data 5 times to avoid lost data
-      pHandle->noDataPollCnt = 0;
-      // lock
-      taosWLockLatch(&pTq->lock);
-      code = tqRegisterPushHandle(pTq, pHandle, pMsg);
-      taosWUnLockLatch(&pTq->lock);
-      tDeleteSMqDataRsp(&dataRsp);
-      return code;
-    }
-    else{
-      pHandle->noDataPollCnt++;
-    }
+  if (terrno == TSDB_CODE_WAL_LOG_NOT_EXIST && dataRsp.blockNum == 0) {
+    // lock
+    taosWLockLatch(&pTq->lock);
+    code = tqRegisterPushHandle(pTq, pHandle, pMsg);
+    taosWUnLockLatch(&pTq->lock);
+    tDeleteSMqDataRsp(&dataRsp);
+    return code;
   }
 
   // NOTE: this pHandle->consumerId may have been changed already.

source/libs/wal/src/walRef.c

@@ -63,21 +63,22 @@ int32_t walSetRefVer(SWalRef *pRef, int64_t ver) {
   return 0;
 }
 
-SWalRef *walRefFirstVer(SWal *pWal, SWalRef *pRef) {
-  if (pRef == NULL) {
-    pRef = walOpenRef(pWal);
-    if (pRef == NULL) {
-      return NULL;
-    }
-  }
+void walRefFirstVer(SWal *pWal, SWalRef *pRef) {
   taosThreadMutexLock(&pWal->mutex);
   int64_t ver = walGetFirstVer(pWal);
   pRef->refVer = ver;
 
   taosThreadMutexUnlock(&pWal->mutex);
   wDebug("vgId:%d, wal ref version %" PRId64 " for first", pWal->cfg.vgId, ver);
+}
 
-  return pRef;
+void walRefLastVer(SWal *pWal, SWalRef *pRef) {
+  taosThreadMutexLock(&pWal->mutex);
+  int64_t ver = walGetLastVer(pWal);
+  pRef->refVer = ver;
+
+  taosThreadMutexUnlock(&pWal->mutex);
+  wDebug("vgId:%d, wal ref version %" PRId64 " for last", pWal->cfg.vgId, ver);
 }
 
 SWalRef *walRefCommittedVer(SWal *pWal) {