Merge pull request #19999 from taosdata/feat/TS-2633

feat:add disable stream option

include/common/tglobal.h

@@ -159,6 +159,8 @@ extern int32_t tsUptimeInterval;
 extern int32_t tsRpcRetryLimit;
 extern int32_t tsRpcRetryInterval;
 
+extern bool    tsDisableStream;
+
 // #define NEEDTO_COMPRESSS_MSG(size) (tsCompressMsgSize != -1 && (size) > tsCompressMsgSize)
 
 int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDir, const char **envCmd,

source/common/src/tglobal.c

@@ -189,6 +189,7 @@ int32_t tsGrantHBInterval = 60;
 int32_t tsUptimeInterval = 300;    // seconds
 char    tsUdfdResFuncs[512] = "";  // udfd resident funcs that teardown when udfd exits
 char    tsUdfdLdLibPath[512] = "";
+bool    tsDisableStream = false;
 
 #ifndef _STORAGE
 int32_t taosSetTfsCfg(SConfig *pCfg) {
@@ -469,6 +470,8 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
   if (cfgAddString(pCfg, "udfdResFuncs", tsUdfdResFuncs, 0) != 0) return -1;
   if (cfgAddString(pCfg, "udfdLdLibPath", tsUdfdLdLibPath, 0) != 0) return -1;
 
+  if (cfgAddBool(pCfg, "disableStream", tsDisableStream, 0) != 0) return -1;
+
   GRANT_CFG_ADD;
   return 0;
 }
@@ -770,6 +773,9 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
   if (tsQueryBufferSize >= 0) {
     tsQueryBufferSizeBytes = tsQueryBufferSize * 1048576UL;
   }
+  
+  tsDisableStream = cfgGetItem(pCfg, "disableStream")->bval;
+
   GRANT_CFG_GET;
   return 0;
 }

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

@@ -307,7 +307,7 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver)
     taosWUnLockLatch(&pTq->pushLock);
   }
 
-  if (vnodeIsRoleLeader(pTq->pVnode)) {
+  if (!tsDisableStream && vnodeIsRoleLeader(pTq->pVnode)) {
     if (taosHashGetSize(pTq->pStreamMeta->pTasks) == 0) return 0;
     if (msgType == TDMT_VND_SUBMIT) {
       void* data = taosMemoryMalloc(msgLen);