diff --git a/include/util/taoserror.h b/include/util/taoserror.h
index 1911c48d26..19cdd72cc7 100644
--- a/include/util/taoserror.h
+++ b/include/util/taoserror.h
@@ -958,6 +958,7 @@ int32_t  taosGetErrSize();
 #define TSDB_CODE_STREAM_INVALID_STATETRANS      TAOS_DEF_ERROR_CODE(0, 0x4103)
 #define TSDB_CODE_STREAM_TASK_IVLD_STATUS        TAOS_DEF_ERROR_CODE(0, 0x4104)
 #define TSDB_CODE_STREAM_NOT_LEADER              TAOS_DEF_ERROR_CODE(0, 0x4105)
+#define TSDB_CODE_STREAM_INTERNAL_ERROR          TAOS_DEF_ERROR_CODE(0, 0x41FF)
 
 // TDLite
 #define TSDB_CODE_TDLITE_IVLD_OPEN_FLAGS         TAOS_DEF_ERROR_CODE(0, 0x5100)
diff --git a/source/client/test/clientTests.cpp b/source/client/test/clientTests.cpp
index 3e9f16a2a1..307ef7e06f 100644
--- a/source/client/test/clientTests.cpp
+++ b/source/client/test/clientTests.cpp
@@ -44,7 +44,10 @@ void printSubResults(void* pRes, int32_t* totalRows) {
     }
 
     TAOS_FIELD* fields = taos_fetch_fields(pRes);
-    assert(fields != NULL);
+    if(fields == NULL) {
+      std::cout << "fields is null" << std::endl;
+      break;
+    }
     int32_t numOfFields = taos_field_count(pRes);
     int32_t precision = taos_result_precision(pRes);
     (void)taos_print_row(buf, row, fields, numOfFields);
@@ -61,7 +64,6 @@ void showDB(TAOS* pConn) {
   TAOS_ROW  pRow = NULL;
 
   TAOS_FIELD* pFields = taos_fetch_fields(pRes);
-  assert(pFields != NULL);
   int32_t     numOfFields = taos_num_fields(pRes);
 
   char str[512] = {0};
@@ -173,7 +175,6 @@ void createNewTable(TAOS* pConn, int32_t index, int32_t numOfRows, int64_t start
               i + 14, pVarchar, startTs + 15, i + 15, pVarchar, startTs + 16, i + 16, pVarchar, startTs + 17, i + 17, pVarchar, startTs + 18, i + 18,
               pVarchar, startTs + 19, i + 19, pVarchar);
       TAOS_RES* p = taos_query(pConn, sql);
-      assert(p != NULL);
       if (taos_errno(p) != 0) {
         (void)printf("failed to insert data, reason:%s\n", taos_errstr(p));
       }
@@ -223,9 +224,7 @@ void tmq_commit_cb_print(tmq_t* pTmq, int32_t code, void* param) {
 
 void* doConsumeData(void* param) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
-  assert(pConn != NULL);
   tmq_conf_t* conf = tmq_conf_new();
-  assert(conf != NULL);
   (void)tmq_conf_set(conf, "enable.auto.commit", "true");
   (void)tmq_conf_set(conf, "auto.commit.interval.ms", "1000");
   (void)tmq_conf_set(conf, "group.id", "cgrpName41");
@@ -237,12 +236,10 @@ void* doConsumeData(void* param) {
   (void)tmq_conf_set_auto_commit_cb(conf, tmq_commit_cb_print, NULL);
 
   tmq_t* tmq = tmq_consumer_new(conf, NULL, 0);
-  assert(tmq != NULL);
   tmq_conf_destroy(conf);
 
   // 创建订阅 topics 列表
   tmq_list_t* topicList = tmq_list_new();
-  assert(topicList != NULL);
   (void)tmq_list_append(topicList, "topic_t2");
 
   // 启动订阅
@@ -279,7 +276,6 @@ void* doConsumeData(void* param) {
         }
 
         fields = taos_fetch_fields(pRes);
-        assert(fields != NULL);
         numOfFields = taos_field_count(pRes);
         precision = taos_result_precision(pRes);
         (void)taos_print_row(buf, row, fields, numOfFields);
@@ -336,7 +332,7 @@ TEST(clientCase, connect_Test) {
 
 TEST(clientCase, create_user_Test) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
-  assert(pConn != NULL);
+  ASSERT_NE(pConn, nullptr);
 
   TAOS_RES* pRes = taos_query(pConn, "create user abc pass 'abc'");
   if (taos_errno(pRes) != TSDB_CODE_SUCCESS) {
@@ -349,7 +345,7 @@ TEST(clientCase, create_user_Test) {
 
 TEST(clientCase, create_account_Test) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
-  assert(pConn != NULL);
+  ASSERT_NE(pConn, nullptr);
 
   TAOS_RES* pRes = taos_query(pConn, "create account aabc pass 'abc'");
   if (taos_errno(pRes) != TSDB_CODE_SUCCESS) {
@@ -362,7 +358,7 @@ TEST(clientCase, create_account_Test) {
 
 TEST(clientCase, drop_account_Test) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
-  assert(pConn != NULL);
+  ASSERT_NE(pConn, nullptr);
 
   TAOS_RES* pRes = taos_query(pConn, "drop account aabc");
   if (taos_errno(pRes) != TSDB_CODE_SUCCESS) {
@@ -375,13 +371,13 @@ TEST(clientCase, drop_account_Test) {
 
 TEST(clientCase, show_user_Test) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
-  assert(pConn != NULL);
+  ASSERT_NE(pConn, nullptr);
 
   TAOS_RES* pRes = taos_query(pConn, "show users");
   TAOS_ROW  pRow = NULL;
 
   TAOS_FIELD* pFields = taos_fetch_fields(pRes);
-  assert(pFields != NULL);
+  ASSERT_NE(pFields, nullptr);
   int32_t     numOfFields = taos_num_fields(pRes);
 
   char str[512] = {0};
@@ -396,7 +392,7 @@ TEST(clientCase, show_user_Test) {
 
 TEST(clientCase, drop_user_Test) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
-  assert(pConn != NULL);
+  ASSERT_NE(pConn, nullptr);
 
   TAOS_RES* pRes = taos_query(pConn, "drop user abc");
   if (taos_errno(pRes) != TSDB_CODE_SUCCESS) {
@@ -409,13 +405,13 @@ TEST(clientCase, drop_user_Test) {
 
 TEST(clientCase, show_db_Test) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
-  assert(pConn != NULL);
+  ASSERT_NE(pConn, nullptr);
 
   TAOS_RES* pRes = taos_query(pConn, "show databases");
   TAOS_ROW  pRow = NULL;
 
   TAOS_FIELD* pFields = taos_fetch_fields(pRes);
-  assert(pFields != NULL);
+  ASSERT_NE(pFields,  nullptr);
   int32_t     numOfFields = taos_num_fields(pRes);
 
   char str[512] = {0};
@@ -429,7 +425,7 @@ TEST(clientCase, show_db_Test) {
 
 TEST(clientCase, create_db_Test) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
-  assert(pConn != NULL);
+  ASSERT_NE(pConn, nullptr);
 
   TAOS_RES* pRes = taos_query(pConn, "create database abc1 vgroups 2");
   if (taos_errno(pRes) != 0) {
@@ -454,7 +450,7 @@ TEST(clientCase, create_db_Test) {
 
 TEST(clientCase, create_dnode_Test) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
-  assert(pConn != NULL);
+  ASSERT_NE(pConn, nullptr);
 
   TAOS_RES* pRes = taos_query(pConn, "create dnode abc1 port 7000");
   if (taos_errno(pRes) != 0) {
@@ -473,7 +469,7 @@ TEST(clientCase, create_dnode_Test) {
 
 TEST(clientCase, drop_dnode_Test) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
-  assert(pConn != NULL);
+  ASSERT_NE(pConn, nullptr);
 
   TAOS_RES* pRes = taos_query(pConn, "drop dnode 3");
   if (taos_errno(pRes) != 0) {
@@ -498,7 +494,7 @@ TEST(clientCase, drop_dnode_Test) {
 
 TEST(clientCase, use_db_test) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
-  assert(pConn != NULL);
+  ASSERT_NE(pConn, nullptr);
 
   TAOS_RES* pRes = taos_query(pConn, "use abc1");
   if (taos_errno(pRes) != 0) {
@@ -517,7 +513,7 @@ TEST(clientCase, use_db_test) {
 
 TEST(clientCase, create_stable_Test) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
-  assert(pConn != NULL);
+  ASSERT_NE(pConn, nullptr);
 
   TAOS_RES* pRes = taos_query(pConn, "create database if not exists abc1 vgroups 2");
   if (taos_errno(pRes) != 0) {
@@ -545,7 +541,7 @@ TEST(clientCase, create_stable_Test) {
 
 TEST(clientCase, create_table_Test) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
-  assert(pConn != NULL);
+  ASSERT_NE(pConn, nullptr);
 
   TAOS_RES* pRes = taos_query(pConn, "use abc1");
   taos_free_result(pRes);
@@ -564,7 +560,7 @@ TEST(clientCase, create_table_Test) {
 
 TEST(clientCase, create_ctable_Test) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
-  assert(pConn != NULL);
+  ASSERT_NE(pConn, nullptr);
 
   TAOS_RES* pRes = taos_query(pConn, "use abc1");
   if (taos_errno(pRes) != 0) {
@@ -589,7 +585,7 @@ TEST(clientCase, create_ctable_Test) {
 
 TEST(clientCase, show_stable_Test) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
-  assert(pConn != nullptr);
+  ASSERT_NE(pConn, nullptr);
 
   TAOS_RES* pRes = taos_query(pConn, "show abc1.stables");
   if (taos_errno(pRes) != 0) {
@@ -614,7 +610,7 @@ TEST(clientCase, show_stable_Test) {
 
 TEST(clientCase, show_vgroup_Test) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
-  assert(pConn != NULL);
+  ASSERT_NE(pConn, nullptr);
 
   TAOS_RES* pRes = taos_query(pConn, "use abc1");
   if (taos_errno(pRes) != 0) {
@@ -716,7 +712,7 @@ TEST(clientCase, create_multiple_tables) {
 
 TEST(clientCase, show_table_Test) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
-  assert(pConn != NULL);
+  ASSERT_NE(pConn, nullptr);
 
   TAOS_RES* pRes = taos_query(pConn, "show tables");
   if (taos_errno(pRes) != 0) {
@@ -760,7 +756,7 @@ TEST(clientCase, generated_request_id_test) {
     if (result != nullptr) {
       //      (void)printf("0x%llx, index:%d\n", v, i);
     }
-    assert(result == nullptr);
+    ASSERT_EQ(result, nullptr);
     (void)taosHashPut(phash, &v, sizeof(v), NULL, 0);
   }
 
@@ -1090,7 +1086,7 @@ TEST(clientCase, sub_db_test) {
         if (row == NULL) break;
 
         fields = taos_fetch_fields(pRes);
-        assert(fields != NULL);
+        ASSERT_NE(fields, nullptr);
         numOfFields = taos_field_count(pRes);
         precision = taos_result_precision(pRes);
         rows++;
@@ -1226,7 +1222,7 @@ TEST(clientCase, td_25129) {
   (void)tmq_conf_set(conf, "msg.with.table.name", "true");
 
   tmq_t* tmq = tmq_consumer_new(conf, NULL, 0);
-  assert(tmq != NULL);
+  ASSERT_NE(tmq, nullptr);
   tmq_conf_destroy(conf);
 
   char topicName[128] = "tp";
@@ -1393,7 +1389,7 @@ TEST(clientCase, sub_tb_test) {
   (void)tmq_conf_set_auto_commit_cb(conf, tmq_commit_cb_print, NULL);
 
   tmq_t* tmq = tmq_consumer_new(conf, NULL, 0);
-  assert(tmq != NULL);
+  ASSERT_NE(tmq, nullptr);
   tmq_conf_destroy(conf);
 
   // 创建订阅 topics 列表
diff --git a/source/client/test/tmqTest.cpp b/source/client/test/tmqTest.cpp
index 00ed16944c..c7a75e352a 100644
--- a/source/client/test/tmqTest.cpp
+++ b/source/client/test/tmqTest.cpp
@@ -40,7 +40,7 @@ TEST(testCase, driverInit_Test) {
 
 TEST(testCase, create_topic_ctb_Test) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
-  assert(pConn != NULL);
+  ASSERT_NE(pConn, nullptr);
 
   TAOS_RES* pRes = taos_query(pConn, "use abc1");
   if (taos_errno(pRes) != 0) {
@@ -64,7 +64,7 @@ TEST(testCase, create_topic_ctb_Test) {
 
 TEST(testCase, create_topic_stb_Test) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
-  assert(pConn != NULL);
+  ASSERT_NE(pConn, nullptr);
 
   TAOS_RES* pRes = taos_query(pConn, "use abc1");
   if (taos_errno(pRes) != 0) {
@@ -90,7 +90,7 @@ TEST(testCase, create_topic_stb_Test) {
 #if 0
 TEST(testCase, tmq_subscribe_ctb_Test) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
-  assert(pConn != NULL);
+  ASSERT_NE(pConn, NULL);
 
   TAOS_RES* pRes = taos_query(pConn, "use abc1");
   if (taos_errno(pRes) != 0) {
@@ -116,7 +116,7 @@ TEST(testCase, tmq_subscribe_ctb_Test) {
 
 TEST(testCase, tmq_subscribe_stb_Test) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
-  assert(pConn != NULL);
+  ASSERT_NE(pConn, NULL);
 
   TAOS_RES* pRes = taos_query(pConn, "use abc1");
   if (taos_errno(pRes) != 0) {
diff --git a/source/dnode/mnode/impl/test/topic/topic.cpp b/source/dnode/mnode/impl/test/topic/topic.cpp
index 4c2213b700..6808fdd04d 100644
--- a/source/dnode/mnode/impl/test/topic/topic.cpp
+++ b/source/dnode/mnode/impl/test/topic/topic.cpp
@@ -53,7 +53,6 @@ void* MndTestTopic::BuildCreateDbReq(const char* dbname, int32_t* pContLen) {
 
   int32_t contLen = tSerializeSCreateDbReq(NULL, 0, &createReq);
   void*   pReq = rpcMallocCont(contLen);
-  assert(pReq != NULL);
   (void)tSerializeSCreateDbReq(pReq, contLen, &createReq);
 
   *pContLen = contLen;
@@ -69,7 +68,6 @@ void* MndTestTopic::BuildCreateTopicReq(const char* topicName, const char* sql,
 
   int32_t contLen = tSerializeSCMCreateTopicReq(NULL, 0, &createReq);
   void*   pReq = rpcMallocCont(contLen);
-  assert(pReq != NULL);
   (void)tSerializeSCMCreateTopicReq(pReq, contLen, &createReq);
 
   *pContLen = contLen;
@@ -82,7 +80,6 @@ void* MndTestTopic::BuildDropTopicReq(const char* topicName, int32_t* pContLen)
 
   int32_t contLen = tSerializeSMDropTopicReq(NULL, 0, &dropReq);
   void*   pReq = rpcMallocCont(contLen);
-  assert(pReq != NULL);
   (void)tSerializeSMDropTopicReq(pReq, contLen, &dropReq);
 
   *pContLen = contLen;
diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c
index a70a04f23d..7b422b75fc 100644
--- a/source/dnode/vnode/src/tq/tq.c
+++ b/source/dnode/vnode/src/tq/tq.c
@@ -335,7 +335,7 @@ int32_t tqProcessPollPush(STQ* pTq, SRpcMsg* pMsg) {
       STqHandle* pHandle = *(STqHandle**)pIter;
       tqDebug("vgId:%d start set submit for pHandle:%p, consumer:0x%" PRIx64, vgId, pHandle, pHandle->consumerId);
 
-      if (ASSERT(pHandle->msg != NULL)) {
+      if (pHandle->msg == NULL) {
         tqError("pHandle->msg should not be null");
         taosHashCancelIterate(pTq->pPushMgr, pIter);
         break;
@@ -777,7 +777,11 @@ int32_t tqBuildStreamTask(void* pTqObj, SStreamTask* pTask, int64_t nextProcessV
            pTask->info.selfChildId, pTask->info.taskLevel, p, pNext, pTask->info.fillHistory,
            (int32_t)pTask->hTaskInfo.id.taskId, pTask->info.delaySchedParam, nextProcessVer);
 
-    ASSERT(pChkInfo->checkpointVer <= pChkInfo->nextProcessVer);
+    if(pChkInfo->checkpointVer > pChkInfo->nextProcessVer) {
+      tqError("vgId:%d build stream task, s-task:%s, checkpointVer:%" PRId64 " > nextProcessVer:%" PRId64,
+              vgId, pTask->id.idStr, pChkInfo->checkpointVer, pChkInfo->nextProcessVer);
+      return TSDB_CODE_STREAM_INTERNAL_ERROR;
+    }
   }
 
   return 0;
@@ -817,7 +821,9 @@ static void doStartFillhistoryStep2(SStreamTask* pTask, SStreamTask* pStreamTask
             ", do secondary scan-history from WAL after halt the related stream task:%s",
             id, pTask->info.taskLevel, pStep2Range->minVer, pStep2Range->maxVer, pWindow->skey, pWindow->ekey,
             pStreamTask->id.idStr);
-    ASSERT(pTask->status.schedStatus == TASK_SCHED_STATUS__WAITING);
+    if (pTask->status.schedStatus != TASK_SCHED_STATUS__WAITING) {
+      tqError("s-task:%s level:%d unexpected sched-status:%d", id, pTask->info.taskLevel, pTask->status.schedStatus);
+    }
 
     int32_t code = streamSetParamForStreamScannerStep2(pTask, pStep2Range, pWindow);
     if (code) {
@@ -950,7 +956,10 @@ int32_t tqProcessTaskScanHistory(STQ* pTq, SRpcMsg* pMsg) {
   // the following procedure should be executed, no matter status is stop/pause or not
   tqDebug("s-task:%s scan-history(step 1) ended, elapsed time:%.2fs", id, pTask->execInfo.step1El);
 
-  ASSERT(pTask->info.fillHistory == 1);
+  if(pTask->info.fillHistory != 1) {
+    tqError("s-task:%s fill-history is disabled, unexpected", id);
+    return  TSDB_CODE_STREAM_INTERNAL_ERROR;
+  }
 
   // 1. get the related stream task
   SStreamTask* pStreamTask = NULL;
@@ -967,7 +976,10 @@ int32_t tqProcessTaskScanHistory(STQ* pTq, SRpcMsg* pMsg) {
     return code;  // todo: handle failure
   }
 
-  ASSERT(pStreamTask->info.taskLevel == TASK_LEVEL__SOURCE);
+  if(pStreamTask->info.taskLevel != TASK_LEVEL__SOURCE) {
+    tqError("s-task:%s fill-history task related stream task level:%d, unexpected", id, pStreamTask->info.taskLevel);
+    return TSDB_CODE_STREAM_INTERNAL_ERROR;
+  }
   code = streamTaskHandleEventAsync(pStreamTask->status.pSM, TASK_EVENT_HALT, handleStep2Async, pTq);
 
   streamMetaReleaseTask(pMeta, pStreamTask);
diff --git a/source/dnode/vnode/src/tq/tqRead.c b/source/dnode/vnode/src/tq/tqRead.c
index c779a17301..902680f76a 100644
--- a/source/dnode/vnode/src/tq/tqRead.c
+++ b/source/dnode/vnode/src/tq/tqRead.c
@@ -386,7 +386,8 @@ int32_t extractMsgFromWal(SWalReader* pReader, void** pItem, int64_t maxVer, con
       }
 
     } else {
-      ASSERT(0);
+      tqError("s-task:%s invalid msg type:%d, ver:%" PRId64, id, pCont->msgType, ver);
+      return TSDB_CODE_STREAM_INTERNAL_ERROR;
     }
 
     return code;
@@ -675,7 +676,11 @@ int32_t tqRetrieveDataBlock(STqReader* pReader, SSDataBlock** pRes, const char*
     pReader->cachedSchemaSuid = suid;
     pReader->cachedSchemaVer = sversion;
 
-    ASSERT(pReader->cachedSchemaVer == pReader->pSchemaWrapper->version);
+    if(pReader->cachedSchemaVer != pReader->pSchemaWrapper->version) {
+      tqError("vgId:%d, schema version mismatch, suid:%" PRId64 ", uid:%" PRId64 ", version:%d, cached version:%d",
+              vgId, suid, uid, sversion, pReader->pSchemaWrapper->version);
+      return TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND;
+    }
     if (blockDataGetNumOfCols(pBlock) == 0) {
       code = buildResSDataBlock(pReader->pResBlock, pReader->pSchemaWrapper, pReader->pColIdList);
       TSDB_CHECK_CODE(code, line, END);