From 930b267a75fe975a88b75af01f8f676176eecd38 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Mon, 17 Apr 2023 22:59:00 +0800
Subject: [PATCH] fix(stream): set the correct start offset for stream task.

---
 source/libs/stream/src/streamDispatch.c | 12 +++++-------
 source/libs/stream/src/streamExec.c     |  5 ++++-
 source/libs/stream/src/streamMeta.c     |  3 +--
 source/libs/stream/src/streamRecover.c  |  8 ++++++--
 4 files changed, 16 insertions(+), 12 deletions(-)

diff --git a/source/libs/stream/src/streamDispatch.c b/source/libs/stream/src/streamDispatch.c
index a9f6d29bf5..549374ed94 100644
--- a/source/libs/stream/src/streamDispatch.c
+++ b/source/libs/stream/src/streamDispatch.c
@@ -230,23 +230,21 @@ int32_t streamDispatchOneCheckReq(SStreamTask* pTask, const SStreamTaskCheckReq*
   SEncoder encoder;
   tEncoderInit(&encoder, abuf, tlen);
   if ((code = tEncodeSStreamTaskCheckReq(&encoder, pReq)) < 0) {
-    goto FAIL;
+    rpcFreeCont(buf);
+    return code;
   }
+
   tEncoderClear(&encoder);
 
   msg.contLen = tlen + sizeof(SMsgHead);
   msg.pCont = buf;
   msg.msgType = TDMT_STREAM_TASK_CHECK;
 
-  qDebug("dispatch from s-task:%s to downstream s-task:%"PRIx64":%d node %d: check msg", pTask->id.idStr,
-      pReq->streamId, pReq->downstreamTaskId, nodeId);
+  qDebug("dispatch from s-task:%s to downstream s-task:%" PRIx64 ":%d node %d: check msg", pTask->id.idStr,
+         pReq->streamId, pReq->downstreamTaskId, nodeId);
 
   tmsgSendReq(pEpSet, &msg);
-
   return 0;
-FAIL:
-  if (buf) rpcFreeCont(buf);
-  return code;
 }
 
 int32_t streamDispatchOneRecoverFinishReq(SStreamTask* pTask, const SStreamRecoverFinishReq* pReq, int32_t vgId,
diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c
index 3d896c08ac..9a6ff302ef 100644
--- a/source/libs/stream/src/streamExec.c
+++ b/source/libs/stream/src/streamExec.c
@@ -189,7 +189,10 @@ int32_t streamScanExec(SStreamTask* pTask, int32_t batchSz) {
       qDebug("task %d scan exec dispatch block num %d", pTask->id.taskId, batchCnt);
       streamDispatch(pTask);
     }
-    if (finished) break;
+
+    if (finished) {
+      break;
+    }
   }
   return 0;
 }
diff --git a/source/libs/stream/src/streamMeta.c b/source/libs/stream/src/streamMeta.c
index 51cc315780..860514bb44 100644
--- a/source/libs/stream/src/streamMeta.c
+++ b/source/libs/stream/src/streamMeta.c
@@ -296,8 +296,7 @@ int32_t streamLoadTasks(SStreamMeta* pMeta, int64_t ver) {
     tDecodeStreamTask(&decoder, pTask);
     tDecoderClear(&decoder);
 
-    // todo set correct initial version.
-    if (pMeta->expandFunc(pMeta->ahandle, pTask, 0) < 0) {
+    if (pMeta->expandFunc(pMeta->ahandle, pTask, pTask->chkInfo.version) < 0) {
       tdbFree(pKey);
       tdbFree(pVal);
       tdbTbcClose(pCur);
diff --git a/source/libs/stream/src/streamRecover.c b/source/libs/stream/src/streamRecover.c
index 03afc0692d..55c745e417 100644
--- a/source/libs/stream/src/streamRecover.c
+++ b/source/libs/stream/src/streamRecover.c
@@ -102,8 +102,10 @@ int32_t streamRecheckOneDownstream(SStreamTask* pTask, const SStreamTaskCheckRsp
       .downstreamNodeId = pRsp->downstreamNodeId,
       .childId = pRsp->childId,
   };
-  qDebug("task %d at node %d check downstream task %d at node %d (recheck)", pTask->id.taskId, pTask->nodeId,
+
+  qDebug("s-task:%s at node %d check downstream task %d at node %d (recheck)", pTask->id.idStr, pTask->nodeId,
          req.downstreamTaskId, req.downstreamNodeId);
+
   if (pTask->outputType == TASK_OUTPUT__FIXED_DISPATCH) {
     streamDispatchOneCheckReq(pTask, &req, pRsp->downstreamNodeId, &pTask->fixedEpDispatcher.epSet);
   } else if (pTask->outputType == TASK_OUTPUT__SHUFFLE_DISPATCH) {
@@ -116,6 +118,7 @@ int32_t streamRecheckOneDownstream(SStreamTask* pTask, const SStreamTaskCheckRsp
       }
     }
   }
+
   return 0;
 }
 
@@ -158,9 +161,10 @@ int32_t streamProcessTaskCheckRsp(SStreamTask* pTask, const SStreamTaskCheckRsp*
     } else {
       ASSERT(0);
     }
-  } else {
+  } else { // not ready, it should wait for at least 100ms and then retry
     streamRecheckOneDownstream(pTask, pRsp);
   }
+
   return 0;
 }
 
-- 
GitLab