From 522e688387d37184636f2c74a4109ac5514b895b Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Fri, 5 Jan 2024 22:16:58 +0800
Subject: [PATCH] fix(stream): check if in restart procedure in call back.

---
 source/dnode/vnode/src/tqCommon/tqCommon.c | 29 +++++++++++++---------
 source/libs/stream/src/streamQueue.c       |  1 +
 2 files changed, 18 insertions(+), 12 deletions(-)

diff --git a/source/dnode/vnode/src/tqCommon/tqCommon.c b/source/dnode/vnode/src/tqCommon/tqCommon.c
index ad6d11aaf2..16d073797b 100644
--- a/source/dnode/vnode/src/tqCommon/tqCommon.c
+++ b/source/dnode/vnode/src/tqCommon/tqCommon.c
@@ -774,22 +774,27 @@ int32_t tqStreamTaskProcessRunReq(SStreamMeta* pMeta, SRpcMsg* pMsg, bool isLead
 
 int32_t tqStartTaskCompleteCallback(SStreamMeta* pMeta) {
   STaskStartInfo* pStartInfo = &pMeta->startInfo;
+  int32_t vgId = pMeta->vgId;
+
   streamMetaWLock(pMeta);
+  if (pStartInfo->taskStarting == 1) {
+    tqDebug("vgId:%d already in start tasks procedure in other thread, restartCounter:%d, do nothing", vgId,
+            pMeta->startInfo.restartCount);
+  } else {  // not in starting procedure
+    if (pStartInfo->restartCount > 0) {
+      pStartInfo->restartCount -= 1;
+      tqDebug("vgId:%d role:%d need to restart all tasks again, restartCounter:%d", vgId, pMeta->role,
+              pStartInfo->restartCount);
 
-  if (pStartInfo->restartCount > 0) {
-    pStartInfo->restartCount -= 1;
-
-    ASSERT(pStartInfo->taskStarting == 0);
-    tqDebug("vgId:%d role:%d need to restart all tasks again, restartCounter:%d", pMeta->vgId, pMeta->role,
-            pStartInfo->restartCount);
-
-    streamMetaWUnLock(pMeta);
-    restartStreamTasks(pMeta, (pMeta->role == NODE_ROLE_LEADER));
-  } else {
-    streamMetaWUnLock(pMeta);
-    tqDebug("vgId:%d start all tasks completed in callbackFn", pMeta->vgId);
+      streamMetaWUnLock(pMeta);
+      restartStreamTasks(pMeta, (pMeta->role == NODE_ROLE_LEADER));
+      return TSDB_CODE_SUCCESS;
+    } else {
+      tqDebug("vgId:%d start all tasks completed in callbackFn", pMeta->vgId);
+    }
   }
 
+  streamMetaWUnLock(pMeta);
   return TSDB_CODE_SUCCESS;
 }
 
diff --git a/source/libs/stream/src/streamQueue.c b/source/libs/stream/src/streamQueue.c
index 459efb838c..479612d9d1 100644
--- a/source/libs/stream/src/streamQueue.c
+++ b/source/libs/stream/src/streamQueue.c
@@ -157,6 +157,7 @@ int32_t streamTaskGetDataFromInputQ(SStreamTask* pTask, SStreamQueueItem** pInpu
   *numOfBlocks = 0;
   *blockSize = 0;
 
+  // todo remove it
   // no available token in bucket for sink task, let's wait for a little bit
   if (taskLevel == TASK_LEVEL__SINK && (!streamTaskExtractAvailableToken(pTask->outputInfo.pTokenBucket, pTask->id.idStr))) {
     stDebug("s-task:%s no available token in bucket for sink data, wait for 10ms", id);