set task status

source/dnode/vnode/src/tq/tq.c

@@ -1440,7 +1440,9 @@ int32_t tqProcessTaskPauseReq(STQ* pTq, int64_t sversion, char* msg, int32_t msg
     return code;
   }
   SStreamTask* pHistoryTask = streamMetaAcquireTask(pTq->pStreamMeta, pTask->historyTaskId.taskId);
-  code = tqProcessTaskPauseImpl(pTq->pStreamMeta, pHistoryTask);
+  if (pHistoryTask) {
+    code = tqProcessTaskPauseImpl(pTq->pStreamMeta, pHistoryTask);
+  }
   return code;
 }
 
@@ -1486,7 +1488,9 @@ int32_t tqProcessTaskResumeReq(STQ* pTq, int64_t sversion, char* msg, int32_t ms
   }
 
   SStreamTask* pHistoryTask = streamMetaAcquireTask(pTq->pStreamMeta, pTask->historyTaskId.taskId);
-  code = tqProcessTaskResumeImpl(pTq, pHistoryTask, sversion, pReq->igUntreated);
+  if (pHistoryTask) {
+    code = tqProcessTaskResumeImpl(pTq, pHistoryTask, sversion, pReq->igUntreated);
+  }
   return code;
 }
 

source/libs/stream/src/streamTask.c

@@ -42,7 +42,11 @@ SStreamTask* tNewStreamTask(int64_t streamId, int8_t taskLevel, int8_t fillHisto
 
   pTask->id.idStr = taosStrdup(buf);
   pTask->status.schedStatus = TASK_SCHED_STATUS__INACTIVE;
-  pTask->status.taskStatus = TASK_STATUS__SCAN_HISTORY;
+  if (fillHistory) {
+    pTask->status.taskStatus = TASK_STATUS__SCAN_HISTORY;
+  } else {
+    pTask->status.taskStatus = TASK_STATUS__NORMAL;
+  }
   pTask->inputStatus = TASK_INPUT_STATUS__NORMAL;
   pTask->outputStatus = TASK_OUTPUT_STATUS__NORMAL;