From 809dd064a7071544a877e95b4f5d301438f12250 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Wed, 8 Nov 2023 11:04:16 +0800
Subject: [PATCH] refactor(stream): create sim env for stream processing.

---
 source/dnode/vnode/src/tq/tq.c | 38 ++++++++++++++++++++++++++++++++++
 1 file changed, 38 insertions(+)

diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c
index fd35f5f7eb..a8d25b6bb3 100644
--- a/source/dnode/vnode/src/tq/tq.c
+++ b/source/dnode/vnode/src/tq/tq.c
@@ -1920,7 +1920,45 @@ int32_t tqProcessTaskUpdateReq(STQ* pTq, SRpcMsg* pMsg) {
       streamMetaWUnLock(pMeta);
     } else {
       streamMetaWUnLock(pMeta);
+#if 0
       tqStartStreamTaskAsync(pTq, true);
+#else
+      // For debug purpose.
+      // the following procedure consume many CPU resource, result in the re-election of leader
+      // with high probability. So we employ it as a test case for the stream processing framework, with
+      // checkpoint/restart/nodeUpdate etc.
+      while (streamMetaTaskInTimer(pMeta)) {
+        tqDebug("vgId:%d some tasks in timer, wait for 100ms and recheck", pMeta->vgId);
+        taosMsleep(100);
+      }
+
+      streamMetaWLock(pMeta);
+
+      int32_t code = streamMetaReopen(pMeta);
+      if (code != 0) {
+        tqError("vgId:%d failed to reopen stream meta", vgId);
+        streamMetaWUnLock(pMeta);
+        taosArrayDestroy(req.pNodeList);
+        return -1;
+      }
+
+      if (streamMetaLoadAllTasks(pTq->pStreamMeta) < 0) {
+        tqError("vgId:%d failed to load stream tasks", vgId);
+        streamMetaWUnLock(pMeta);
+        taosArrayDestroy(req.pNodeList);
+        return -1;
+      }
+
+      if (vnodeIsRoleLeader(pTq->pVnode) && !tsDisableStream) {
+        tqInfo("vgId:%d restart all stream tasks after all tasks being updated", vgId);
+        tqResetStreamTaskStatus(pTq);
+        tqStartStreamTaskAsync(pTq, false);
+      } else {
+        tqInfo("vgId:%d, follower node not start stream tasks", vgId);
+      }
+
+      streamMetaWUnLock(pMeta);
+#endif
     }
   }