From 9a76ae72d625c3e97ab9968f47321b7512e9cac9 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Wed, 16 Aug 2023 19:11:43 +0800
Subject: [PATCH] fix(stream): add the back pressure for sink node.

---
 source/dnode/vnode/src/tq/tqRestore.c | 7 +++++++
 source/libs/stream/src/streamExec.c   | 6 ++++++
 2 files changed, 13 insertions(+)

diff --git a/source/dnode/vnode/src/tq/tqRestore.c b/source/dnode/vnode/src/tq/tqRestore.c
index d363031db1..ed612587f5 100644
--- a/source/dnode/vnode/src/tq/tqRestore.c
+++ b/source/dnode/vnode/src/tq/tqRestore.c
@@ -287,6 +287,13 @@ int32_t createStreamTaskRunReq(SStreamMeta* pStreamMeta, bool* pScanIdle) {
       continue;
     }
 
+    // downstream task has blocked the output, stopped for a while
+    if (pTask->inputStatus == TASK_INPUT_STATUS__BLOCKED) {
+      tqDebug("s-task:%s inputQ is blocked, do nothing", pTask->id.idStr);
+      streamMetaReleaseTask(pStreamMeta, pTask);
+      continue;
+    }
+
     *pScanIdle = false;
 
     // seek the stored version and extract data from WAL
diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c
index 02bbce6485..d761c5ad24 100644
--- a/source/libs/stream/src/streamExec.c
+++ b/source/libs/stream/src/streamExec.c
@@ -191,6 +191,12 @@ int32_t streamScanExec(SStreamTask* pTask, int32_t batchSize) {
         return 0;
       }
 
+      if (pTask->inputStatus == TASK_INPUT_STATUS__BLOCKED) {
+        qDebug("s-task:%s inputQ is blocked, wait for 5sec and retry", pTask->id.idStr);
+        taosMsleep(10000);
+        continue;
+      }
+
       SSDataBlock* output = NULL;
       uint64_t     ts = 0;
       code = qExecTask(exec, &output, &ts);