From 105542bb211ef59c8c8d334c40975c7744c78034 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Wed, 13 Dec 2023 17:44:00 +0800
Subject: [PATCH] fix(stream):handle the case of delete orphaned tasks.

---
 source/dnode/mnode/impl/src/mndStream.c      | 19 +++++++++++++------
 source/dnode/mnode/impl/src/mndStreamTrans.c |  3 +--
 2 files changed, 14 insertions(+), 8 deletions(-)

diff --git a/source/dnode/mnode/impl/src/mndStream.c b/source/dnode/mnode/impl/src/mndStream.c
index e8d5dfd1f5..bd67af712a 100644
--- a/source/dnode/mnode/impl/src/mndStream.c
+++ b/source/dnode/mnode/impl/src/mndStream.c
@@ -704,10 +704,10 @@ static int32_t mndPersistTaskDropReq(SMnode *pMnode, STrans *pTrans, SStreamTask
   pReq->streamId = pTask->id.streamId;
 
   STransAction action = {0};
-  SEpSet  epset = {0};
-  if(pTask->info.nodeId == SNODE_HANDLE){
+  SEpSet       epset = {0};
+  if (pTask->info.nodeId == SNODE_HANDLE) {
     SSnodeObj *pObj = NULL;
-    void  *pIter = NULL;
+    void      *pIter = NULL;
     while (1) {
       pIter = sdbFetch(pMnode->pSdb, SDB_SNODE, pIter, (void **)&pObj);
       if (pIter == NULL) {
@@ -717,10 +717,16 @@ static int32_t mndPersistTaskDropReq(SMnode *pMnode, STrans *pTrans, SStreamTask
       addEpIntoEpSet(&epset, pObj->pDnode->fqdn, pObj->pDnode->port);
       sdbRelease(pMnode->pSdb, pObj);
     }
-  }else{
+  } else {
     SVgObj *pVgObj = mndAcquireVgroup(pMnode, pTask->info.nodeId);
-    epset = mndGetVgroupEpset(pMnode, pVgObj);
-    mndReleaseVgroup(pMnode, pVgObj);
+    if (pVgObj != NULL) {
+      epset = mndGetVgroupEpset(pMnode, pVgObj);
+      mndReleaseVgroup(pMnode, pVgObj);
+    } else {
+      mDebug("orphaned task:0x%x need to be dropped, nodeId:%d, no redo action", pTask->id.taskId, pTask->info.nodeId);
+      taosMemoryFree(pReq);
+      return 0;
+    }
   }
 
   // The epset of nodeId of this task may have been expired now, let's use the newest epset from mnode.
@@ -1657,6 +1663,7 @@ static void setTaskAttrInResBlock(SStreamObj *pStream, SStreamTask *pTask, SSDat
 
   STaskStatusEntry *pe = taosHashGet(execInfo.pTaskMap, &id, sizeof(id));
   if (pe == NULL) {
+    mError("task:0x%" PRIx64 " not exists in vnode, no valid status/stage info", id.taskId);
     return;
   }
 
diff --git a/source/dnode/mnode/impl/src/mndStreamTrans.c b/source/dnode/mnode/impl/src/mndStreamTrans.c
index fa36d69d6e..43e85a9405 100644
--- a/source/dnode/mnode/impl/src/mndStreamTrans.c
+++ b/source/dnode/mnode/impl/src/mndStreamTrans.c
@@ -48,8 +48,7 @@ int32_t clearFinishedTrans(SMnode* pMnode) {
       void* pKey = taosHashGetKey(pEntry, &keyLen);
       // key is the name of src/dst db name
       SKeyInfo info = {.pKey = pKey, .keyLen = keyLen};
-
-      mDebug("transId:%d %s startTs:%" PRId64 "cleared due to finished", pEntry->transId, pEntry->name,
+      mDebug("transId:%d %s startTs:%" PRId64 " cleared since finished", pEntry->transId, pEntry->name,
              pEntry->startTime);
       taosArrayPush(pList, &info);
     } else {