From 80fd3e044546a16221816829e768ceaf095671cb Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Mon, 4 Mar 2024 19:31:47 +0800
Subject: [PATCH] fix(stream): remove related fill-history if task in stop
 status.

---
 include/libs/stream/tstream.h              |  2 +-
 source/dnode/vnode/src/tqCommon/tqCommon.c | 36 +++++++++++++++-------
 source/libs/stream/src/streamTask.c        | 14 +--------
 3 files changed, 27 insertions(+), 25 deletions(-)

diff --git a/include/libs/stream/tstream.h b/include/libs/stream/tstream.h
index cf980400d6..0df6b76cf7 100644
--- a/include/libs/stream/tstream.h
+++ b/include/libs/stream/tstream.h
@@ -782,7 +782,7 @@ bool    streamTaskIsAllUpstreamClosed(SStreamTask* pTask);
 bool    streamTaskSetSchedStatusWait(SStreamTask* pTask);
 int8_t  streamTaskSetSchedStatusActive(SStreamTask* pTask);
 int8_t  streamTaskSetSchedStatusInactive(SStreamTask* pTask);
-int32_t streamTaskClearHTaskAttr(SStreamTask* pTask, int32_t clearRelHalt, bool metaLock);
+int32_t streamTaskClearHTaskAttr(SStreamTask* pTask, int32_t clearRelHalt);
 
 int32_t streamTaskHandleEvent(SStreamTaskSM* pSM, EStreamTaskEvent event);
 int32_t streamTaskOnHandleEventSuccess(SStreamTaskSM* pSM, EStreamTaskEvent event);
diff --git a/source/dnode/vnode/src/tqCommon/tqCommon.c b/source/dnode/vnode/src/tqCommon/tqCommon.c
index a2d45062b9..8c66ecc6a9 100644
--- a/source/dnode/vnode/src/tqCommon/tqCommon.c
+++ b/source/dnode/vnode/src/tqCommon/tqCommon.c
@@ -142,8 +142,10 @@ int32_t tqStreamTaskProcessUpdateReq(SStreamMeta* pMeta, SMsgCb* cb, SRpcMsg* pM
   if (HAS_RELATED_FILLHISTORY_TASK(pTask)) {
     ppHTask = (SStreamTask**)taosHashGet(pMeta->pTasksMap, &pTask->hTaskInfo.id, sizeof(pTask->hTaskInfo.id));
     if (ppHTask == NULL || *ppHTask == NULL) {
-      tqError("vgId:%d failed to acquire fill-history task:0x%x when handling update, it may have been dropped already",
-              vgId, req.taskId);
+      tqError(
+          "vgId:%d failed to acquire fill-history task:0x%x when handling update, may have been dropped already, rel "
+          "stream task:0x%x",
+          vgId, (uint32_t)pTask->hTaskInfo.id.taskId, req.taskId);
       CLEAR_RELATED_FILLHISTORY_TASK(pTask);
     } else {
       tqDebug("s-task:%s fill-history task update nodeEp along with stream task", (*ppHTask)->id.idStr);
@@ -612,23 +614,35 @@ int32_t tqStreamTaskProcessDeployReq(SStreamMeta* pMeta, SMsgCb* cb, int64_t sve
 
 int32_t tqStreamTaskProcessDropReq(SStreamMeta* pMeta, char* msg, int32_t msgLen) {
   SVDropStreamTaskReq* pReq = (SVDropStreamTaskReq*)msg;
+  int32_t              vgId = pMeta->vgId;
+  STaskId              hTaskId = {0};
 
-  int32_t vgId = pMeta->vgId;
   tqDebug("vgId:%d receive msg to drop s-task:0x%x", vgId, pReq->taskId);
 
-  SStreamTask* pTask = streamMetaAcquireTask(pMeta, pReq->streamId, pReq->taskId);
-  if (pTask != NULL) {
-    // drop the related fill-history task firstly
+  streamMetaWLock(pMeta);
+
+  STaskId       id = {.streamId = pReq->streamId, .taskId = pReq->taskId};
+  SStreamTask** ppTask = (SStreamTask**)taosHashGet(pMeta->pTasksMap, &id, sizeof(id));
+  if ((ppTask != NULL) && ((*ppTask) != NULL)) {
+    streamMetaAcquireOneTask(*ppTask);
+    SStreamTask* pTask = *ppTask;
+
     if (HAS_RELATED_FILLHISTORY_TASK(pTask)) {
-      STaskId* pHTaskId = &pTask->hTaskInfo.id;
-      streamMetaUnregisterTask(pMeta, pHTaskId->streamId, pHTaskId->taskId);
-      tqDebug("s-task:0x%x vgId:%d drop fill-history task:0x%x firstly", pReq->taskId, vgId,
-              (int32_t)pHTaskId->taskId);
+      hTaskId.streamId = pTask->hTaskInfo.id.streamId;
+      hTaskId.taskId = pTask->hTaskInfo.id.taskId;
+      streamTaskClearHTaskAttr(pTask, pReq->resetRelHalt);
     }
+
     streamMetaReleaseTask(pMeta, pTask);
   }
 
-  streamTaskClearHTaskAttr(pTask, pReq->resetRelHalt, true);
+  streamMetaWUnLock(pMeta);
+
+  // drop the related fill-history task firstly
+  if (hTaskId.taskId != 0 && hTaskId.streamId != 0) {
+    streamMetaUnregisterTask(pMeta, hTaskId.streamId, hTaskId.taskId);
+    tqDebug("s-task:0x%x vgId:%d drop rel fill-history task:0x%x firstly", pReq->taskId, vgId, (int32_t)hTaskId.taskId);
+  }
 
   // drop the stream task now
   streamMetaUnregisterTask(pMeta, pReq->streamId, pReq->taskId);
diff --git a/source/libs/stream/src/streamTask.c b/source/libs/stream/src/streamTask.c
index 9639921c77..68b4ba2296 100644
--- a/source/libs/stream/src/streamTask.c
+++ b/source/libs/stream/src/streamTask.c
@@ -766,21 +766,13 @@ int8_t streamTaskSetSchedStatusInactive(SStreamTask* pTask) {
   return status;
 }
 
-int32_t streamTaskClearHTaskAttr(SStreamTask* pTask, int32_t resetRelHalt, bool metaLock) {
-  if (pTask == NULL) {
-    return TSDB_CODE_SUCCESS;
-  }
-
+int32_t streamTaskClearHTaskAttr(SStreamTask* pTask, int32_t resetRelHalt) {
   SStreamMeta* pMeta = pTask->pMeta;
   STaskId      sTaskId = {.streamId = pTask->streamTaskId.streamId, .taskId = pTask->streamTaskId.taskId};
   if (pTask->info.fillHistory == 0) {
     return TSDB_CODE_SUCCESS;
   }
 
-  if (metaLock) {
-    streamMetaWLock(pMeta);
-  }
-
   SStreamTask** ppStreamTask = (SStreamTask**)taosHashGet(pMeta->pTasksMap, &sTaskId, sizeof(sTaskId));
   if (ppStreamTask != NULL) {
     stDebug("s-task:%s clear the related stream task:0x%x attr to fill-history task", pTask->id.idStr,
@@ -798,10 +790,6 @@ int32_t streamTaskClearHTaskAttr(SStreamTask* pTask, int32_t resetRelHalt, bool
     taosThreadMutexUnlock(&(*ppStreamTask)->lock);
   }
 
-  if (metaLock) {
-    streamMetaWUnLock(pMeta);
-  }
-
   return TSDB_CODE_SUCCESS;
 }