diff --git a/source/dnode/mnode/impl/src/mndScheduler.c b/source/dnode/mnode/impl/src/mndScheduler.c
index 4f72b26a5e..8d33e61733 100644
--- a/source/dnode/mnode/impl/src/mndScheduler.c
+++ b/source/dnode/mnode/impl/src/mndScheduler.c
@@ -248,7 +248,7 @@ static int32_t doAddSinkTask(SStreamObj* pStream, SMnode* pMnode, SVgObj* pVgrou
     return code;
   }
 
-  mDebug("doAddSinkTask taskId:%s, vgId:%d, isFillHistory:%d", pTask->id.idStr, pVgroup->vgId, isFillhistory);
+  mDebug("doAddSinkTask taskId:%s, %p vgId:%d, isFillHistory:%d", pTask->id.idStr, pTask, pVgroup->vgId, isFillhistory);
 
   pTask->info.nodeId = pVgroup->vgId;
   pTask->info.epSet = mndGetVgroupEpset(pMnode, pVgroup);
@@ -364,12 +364,13 @@ static int32_t buildSourceTask(SStreamObj* pStream, SEpSet* pEpset, bool isFillh
 static void addNewTaskList(SStreamObj* pStream) {
   SArray* pTaskList = taosArrayInit(0, POINTER_BYTES);
   if (taosArrayPush(pStream->tasks, &pTaskList) == NULL) {
-    mError("failed to put array");
+    mError("failed to put into array");
   }
+
   if (pStream->conf.fillHistory) {
     pTaskList = taosArrayInit(0, POINTER_BYTES);
     if (taosArrayPush(pStream->pHTasksList, &pTaskList) == NULL) {
-      mError("failed to put array");
+      mError("failed to put into array");
     }
   }
 }
@@ -402,7 +403,8 @@ static int32_t doAddSourceTask(SMnode* pMnode, SSubplan* plan, SStreamObj* pStre
     return code;
   }
 
-  mDebug("doAddSourceTask taskId:%s, vgId:%d, isFillHistory:%d", pTask->id.idStr, pVgroup->vgId, isFillhistory);
+  mDebug("doAddSourceTask taskId:%s, %p vgId:%d, isFillHistory:%d", pTask->id.idStr, pTask, pVgroup->vgId,
+         isFillhistory);
 
   if (pStream->conf.fillHistory) {
     haltInitialTaskStatus(pTask, plan, isFillhistory);
@@ -512,19 +514,20 @@ static int32_t doAddAggTask(SStreamObj* pStream, SMnode* pMnode, SSubplan* plan,
                             SSnodeObj* pSnode, bool isFillhistory, bool useTriggerParam) {
   int32_t      code = 0;
   SStreamTask* pTask = NULL;
+  const char*  id = NULL;
 
   code = buildAggTask(pStream, pEpset, isFillhistory, useTriggerParam, &pTask);
   if (code != TSDB_CODE_SUCCESS) {
     return code;
   }
 
+  id = pTask->id.idStr;
   if (pSnode != NULL) {
     code = mndAssignStreamTaskToSnode(pMnode, pTask, plan, pSnode);
-    mDebug("doAddAggTask taskId:%s, snode id:%d, isFillHistory:%d", pTask->id.idStr, pSnode->id, isFillhistory);
-
+    mDebug("doAddAggTask taskId:%s, %p snode id:%d, isFillHistory:%d", id, pTask, pSnode->id, isFillhistory);
   } else {
     code = mndAssignStreamTaskToVgroup(pMnode, pTask, plan, pVgroup);
-    mDebug("doAddAggTask taskId:%s, vgId:%d, isFillHistory:%d", pTask->id.idStr, pVgroup->vgId, isFillhistory);
+    mDebug("doAddAggTask taskId:%s, %p vgId:%d, isFillHistory:%d", id, pTask, pVgroup->vgId, isFillhistory);
   }
   return code;
 }
@@ -678,7 +681,7 @@ static int32_t doScheduleStream(SStreamObj* pStream, SMnode* pMnode, SQueryPlan*
   if (numOfPlanLevel > 1 || externalTargetDB || multiTarget || pStream->fixedSinkVgId) {
     // add extra sink
     hasExtraSink = true;
-    int32_t code = addSinkTask(pMnode, pStream, pEpset);
+    code = addSinkTask(pMnode, pStream, pEpset);
     if (code != TSDB_CODE_SUCCESS) {
       return code;
     }
diff --git a/source/libs/stream/src/streamCheckpoint.c b/source/libs/stream/src/streamCheckpoint.c
index af2d8b559b..d1b57c32b9 100644
--- a/source/libs/stream/src/streamCheckpoint.c
+++ b/source/libs/stream/src/streamCheckpoint.c
@@ -1037,6 +1037,7 @@ void checkpointTriggerMonitorFn(void* param, void* tmrId) {
   if (++pTmrInfo->activeCounter < 50) {
     streamTmrStart(checkpointTriggerMonitorFn, 200, param, streamTimer, &pTmrInfo->tmrHandle, vgId,
                    "trigger-recv-monitor");
+    doCleanup(pTask, pNotSendList);
     return;
   }
 
diff --git a/source/libs/stream/src/streamStartTask.c b/source/libs/stream/src/streamStartTask.c
index ee9117ddc8..3518b8681d 100644
--- a/source/libs/stream/src/streamStartTask.c
+++ b/source/libs/stream/src/streamStartTask.c
@@ -355,7 +355,7 @@ int32_t streamMetaStartOneTask(SStreamMeta* pMeta, int64_t streamId, int32_t tas
   // fill-history task can only be launched by related stream tasks.
   STaskExecStatisInfo* pInfo = &pTask->execInfo;
   if (pTask->info.fillHistory == 1) {
-    stError("s-task:0x%x vgId:%d fill-histroy task, not start here", taskId, vgId);
+    stError("s-task:0x%x vgId:%d fill-history task, not start here", taskId, vgId);
     streamMetaReleaseTask(pMeta, pTask);
     return TSDB_CODE_SUCCESS;
   }
@@ -363,6 +363,7 @@ int32_t streamMetaStartOneTask(SStreamMeta* pMeta, int64_t streamId, int32_t tas
   // the start all tasks procedure may happen to start the newly deployed stream task, and results in the
   // concurrently start this task by two threads.
   streamMutexLock(&pTask->lock);
+
   SStreamTaskState status = streamTaskGetStatus(pTask);
   if (status.state != TASK_STATUS__UNINIT) {
     stError("s-task:0x%x vgId:%d status:%s not uninit status, not start stream task", taskId, vgId, status.name);
@@ -379,6 +380,7 @@ int32_t streamMetaStartOneTask(SStreamMeta* pMeta, int64_t streamId, int32_t tas
 
   if(pTask->status.downstreamReady != 0) {
     stFatal("s-task:0x%x downstream should be not ready, but it ready here, internal error happens", taskId);
+    streamMetaReleaseTask(pMeta, pTask);
     return TSDB_CODE_STREAM_INTERNAL_ERROR;
   }
 
@@ -395,7 +397,7 @@ int32_t streamMetaStartOneTask(SStreamMeta* pMeta, int64_t streamId, int32_t tas
     streamMutexUnlock(&pTask->lock);
   }
 
-  // concurrently start task may cause the later started task be failed, and also failed to added into meta result.
+  // concurrently start task may cause the latter started task be failed, and also failed to added into meta result.
   if (code == TSDB_CODE_SUCCESS) {
     code = streamTaskHandleEvent(pTask->status.pSM, TASK_EVENT_INIT);
     if (code != TSDB_CODE_SUCCESS) {