diff --git a/source/dnode/mgmt/mgmt_snode/src/smWorker.c b/source/dnode/mgmt/mgmt_snode/src/smWorker.c
index 8300b13476d8fa4126527e87d77f8652c59b5de4..dbd081338acc3470915d03f1682e2f8b8a72bf6e 100644
--- a/source/dnode/mgmt/mgmt_snode/src/smWorker.c
+++ b/source/dnode/mgmt/mgmt_snode/src/smWorker.c
@@ -58,7 +58,11 @@ static void smProcessStreamQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
   dTrace("msg:%p, get from snode-stream queue", pMsg);
   int32_t code = sndProcessStreamMsg(pMgmt->pSnode, pMsg);
   if (code < 0) {
-    dGError("snd, msg:%p failed to process stream msg %s since %s", pMsg, TMSG_INFO(pMsg->msgType), terrstr(code));
+    if (pMsg) {
+      dGError("snd, msg:%p failed to process stream msg %s since %s", pMsg, TMSG_INFO(pMsg->msgType), terrstr(code));
+    } else {
+      dGError("snd, msg:%p failed to process stream empty msg since %s", pMsg, terrstr(code));
+    }
     smSendRsp(pMsg, terrno);
   }
 
diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
index f1e867a78ab8e3d63f2654e4cce9e1e24b54d55c..40d0e32c2b43f2c7170e8a8a085e8214399999c2 100644
--- a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
+++ b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
@@ -86,8 +86,12 @@ static void vmProcessStreamQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
   int32_t code = vnodeProcessFetchMsg(pVnode->pImpl, pMsg, pInfo);
   if (code != 0) {
     if (terrno != 0) code = terrno;
-    dGError("vgId:%d, msg:%p failed to process stream msg %s since %s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType),
-            terrstr(code));
+    if (pMsg) {
+      dGError("vgId:%d, msg:%p failed to process stream msg %s since %s", pVnode->vgId, pMsg, TMSG_INFO(pMsg->msgType),
+              terrstr(code));
+    } else {
+      dGError("vgId:%d, msg:%p failed to process stream empty msg since %s", pVnode->vgId, pMsg, terrstr(code));
+    }
     vmSendRsp(pMsg, code);
   }
 
diff --git a/source/dnode/mgmt/node_mgmt/src/dmTransport.c b/source/dnode/mgmt/node_mgmt/src/dmTransport.c
index 408244bfb5cfae400d8215ba694d1584174f3763..1e5f3139aa7ee86b3ff68efc3742c90f008a2c3b 100644
--- a/source/dnode/mgmt/node_mgmt/src/dmTransport.c
+++ b/source/dnode/mgmt/node_mgmt/src/dmTransport.c
@@ -157,7 +157,11 @@ static void dmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pRpc, SEpSet *pEpSet) {
 _OVER:
   if (code != 0) {
     if (terrno != 0) code = terrno;
-    dGTrace("msg:%p, failed to process %s since %s", pMsg, TMSG_INFO(pMsg->msgType), terrstr());
+    if (pMsg) {
+      dGTrace("msg:%p, failed to process %s since %s", pMsg, TMSG_INFO(pMsg->msgType), terrstr());
+    } else {
+      dGTrace("msg:%p, failed to process empty msg since %s", pMsg, terrstr());
+    }
 
     if (IsReq(pRpc)) {
       SRpcMsg rsp = {.code = code, .info = pRpc->info};
diff --git a/source/dnode/snode/src/snode.c b/source/dnode/snode/src/snode.c
index 08ea4808c54513c71e01a90204f2e6480c2c802f..f53350c10bbeea2e0b9b9d53edfb5caaa4304b8f 100644
--- a/source/dnode/snode/src/snode.c
+++ b/source/dnode/snode/src/snode.c
@@ -231,7 +231,7 @@ int32_t sndProcessTaskRetrieveReq(SSnode *pSnode, SRpcMsg *pMsg) {
 
 int32_t sndProcessTaskDispatchRsp(SSnode *pSnode, SRpcMsg *pMsg) {
   SStreamDispatchRsp *pRsp = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
-  int32_t             taskId = pRsp->upstreamTaskId;
+  int32_t             taskId = ntohl(pRsp->upstreamTaskId);
   SStreamTask        *pTask = streamMetaGetTask(pSnode->pMeta, taskId);
   if (pTask) {
     streamProcessDispatchRsp(pTask, pRsp, pMsg->code);
diff --git a/tests/script/tsim/stream/fillHistoryBasic1.sim b/tests/script/tsim/stream/fillHistoryBasic1.sim
index 4f4fa96bea41715175c7bef01d9ccddacb0608dd..772a09c0176512ca1617e53cc87e16ec20fb2a4a 100644
--- a/tests/script/tsim/stream/fillHistoryBasic1.sim
+++ b/tests/script/tsim/stream/fillHistoryBasic1.sim
@@ -139,7 +139,7 @@ if $data35 != 3 then
 endi
 
 sql insert into t1 values(1648791223001,12,14,13,11.1);
-sleep 500
+sleep 1000
 sql select * from streamt;
 
 print    count(*) , count(d) , sum(a) , max(b) , min(c)
@@ -256,7 +256,7 @@ if $data35 != 3 then
 endi
 
 sql insert into t1 values(1648791223002,12,14,13,11.1);
-sleep 100
+sleep 1000
 sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
 
 # row 1
@@ -286,7 +286,7 @@ if $data15 != 13 then
 endi
 
 sql insert into t1 values(1648791223003,12,14,13,11.1);
-sleep 100
+sleep 1000
 sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
 
 # row 1
@@ -318,7 +318,7 @@ endi
 sql insert into t1 values(1648791223001,1,1,1,1.1);
 sql insert into t1 values(1648791223002,2,2,2,2.1);
 sql insert into t1 values(1648791223003,3,3,3,3.1);
-sleep 100
+sleep 1000
 sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
 
 # row 1
@@ -350,7 +350,7 @@ endi
 sql insert into t1 values(1648791233003,3,2,3,2.1);
 sql insert into t1 values(1648791233002,5,6,7,8.1);
 sql insert into t1 values(1648791233002,3,2,3,2.1);
-sleep 100
+sleep 1000
 sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
 
 # row 2
@@ -380,7 +380,7 @@ if $data25 != 3 then
 endi
 
 sql insert into t1 values(1648791213004,4,2,3,4.1) (1648791213006,5,4,7,9.1) (1648791213004,40,20,30,40.1) (1648791213005,4,2,3,4.1);
-sleep 100
+sleep 1000
 sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
 
 # row 0
@@ -410,7 +410,7 @@ if $data05 != 3 then
 endi
 
 sql insert into t1 values(1648791223004,4,2,3,4.1) (1648791233006,5,4,7,9.1) (1648791223004,40,20,30,40.1) (1648791233005,4,2,3,4.1);
-sleep 100
+sleep 1000
 sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
 
 # row 1
@@ -594,7 +594,7 @@ if $data35 != 3 then
 endi
 
 sql insert into t1 values(1648791223001,12,14,13,11.1);
-sleep 500
+sleep 1000
 sql select * from streamt;
 
 print    count(*) , count(d) , sum(a) , max(b) , min(c)
@@ -711,7 +711,7 @@ if $data35 != 3 then
 endi
 
 sql insert into t1 values(1648791223002,12,14,13,11.1);
-sleep 100
+sleep 1000
 sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
 
 # row 1
@@ -741,7 +741,7 @@ if $data15 != 13 then
 endi
 
 sql insert into t1 values(1648791223003,12,14,13,11.1);
-sleep 100
+sleep 1000
 sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
 
 # row 1
@@ -773,7 +773,7 @@ endi
 sql insert into t1 values(1648791223001,1,1,1,1.1);
 sql insert into t1 values(1648791223002,2,2,2,2.1);
 sql insert into t1 values(1648791223003,3,3,3,3.1);
-sleep 100
+sleep 1000
 sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
 
 # row 1
@@ -805,7 +805,7 @@ endi
 sql insert into t1 values(1648791233003,3,2,3,2.1);
 sql insert into t1 values(1648791233002,5,6,7,8.1);
 sql insert into t1 values(1648791233002,3,2,3,2.1);
-sleep 100
+sleep 1000
 sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
 
 # row 2
@@ -835,7 +835,7 @@ if $data25 != 3 then
 endi
 
 sql insert into t1 values(1648791213004,4,2,3,4.1) (1648791213006,5,4,7,9.1) (1648791213004,40,20,30,40.1) (1648791213005,4,2,3,4.1);
-sleep 100
+sleep 1000
 sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
 
 # row 0
@@ -865,7 +865,7 @@ if $data05 != 3 then
 endi
 
 sql insert into t1 values(1648791223004,4,2,3,4.1) (1648791233006,5,4,7,9.1) (1648791223004,40,20,30,40.1) (1648791233005,4,2,3,4.1);
-sleep 100
+sleep 1000
 sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
 
 # row 1