Merge pull request #19190 from taosdata/feature/stream_main

fix memory leak

source/libs/stream/src/streamDispatch.c

@@ -487,8 +487,6 @@ int32_t streamDispatchAllBlocks(SStreamTask* pTask, const SStreamDataBlock* pDat
       taosMemoryFree(pReqs);
     }
     return code;
-  } else {
-    ASSERT(0);
   }
   return 0;
 }
@@ -514,7 +512,6 @@ int32_t streamDispatch(SStreamTask* pTask) {
 
   int32_t code = 0;
   if (streamDispatchAllBlocks(pTask, pBlock) < 0) {
-    ASSERT(0);
     code = -1;
     streamQueueProcessFail(pTask->outputQueue);
     atomic_store_8(&pTask->outputStatus, TASK_OUTPUT_STATUS__NORMAL);

source/libs/stream/src/streamRecover.c

@@ -143,6 +143,7 @@ int32_t streamProcessTaskCheckRsp(SStreamTask* pTask, const SStreamTaskCheckRsp*
       ASSERT(left >= 0);
       if (left == 0) {
         taosArrayDestroy(pTask->checkReqIds);
+        pTask->checkReqIds = NULL;
         streamTaskLaunchRecover(pTask, version);
       }
     } else if (pTask->outputType == TASK_OUTPUT__FIXED_DISPATCH) {

source/libs/stream/src/streamTask.c

@@ -175,6 +175,8 @@ void tFreeSStreamTask(SStreamTask* pTask) {
   }
   if (pTask->outputType == TASK_OUTPUT__SHUFFLE_DISPATCH) {
     taosArrayDestroy(pTask->shuffleDispatcher.dbInfo.pVgroupInfos);
+    taosArrayDestroy(pTask->checkReqIds);
+    pTask->checkReqIds = NULL;
   }
 
   if (pTask->pState) streamStateClose(pTask->pState);