fix:error

include/libs/wal/wal.h

@@ -214,7 +214,7 @@ int32_t walSkipFetchBody(SWalReader *pRead, const SWalCkHead *pHead);
 
 void walRefFirstVer(SWal *, SWalRef *);
 void walRefLastVer(SWal *, SWalRef *);
-void walRefCommitVer(SWal *, SWalRef *);
+//void walRefCommitVer(SWal *, SWalRef *);
 
 SWalRef *walOpenRef(SWal *);
 void     walCloseRef(SWal *pWal, int64_t refId);

source/dnode/mnode/impl/src/mndDef.c

@@ -630,6 +630,8 @@ void *tDecodeSubscribeObj(const void *buf, SMqSubscribeObj *pSub, int8_t sver) {
       }
     }
     buf = taosDecodeString(buf, &pSub->qmsg);
+  }else{
+    pSub->qmsg = taosStrdup("");
   }
   return (void *)buf;
 }

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

@@ -379,7 +379,7 @@ int32_t tqCreateHandle(STQ* pTq, SMqRebVgReq* req, STqHandle* handle){
     handle->execHandle.execTb.qmsg = taosStrdup(req->qmsg);
   }
 
-  handle->snapshotVer = walGetLastVer(pTq->pVnode->pWal);
+  handle->snapshotVer = walGetCommittedVer(pTq->pVnode->pWal);
 
   if(buildHandle(pTq, handle) < 0){
     return -1;

source/libs/wal/src/walRef.c

@@ -81,11 +81,11 @@ void walRefLastVer(SWal *pWal, SWalRef *pRef) {
   wDebug("vgId:%d, wal ref version %" PRId64 " for last", pWal->cfg.vgId, ver);
 }
 
-void walRefCommitVer(SWal *pWal, SWalRef *pRef) {
-  taosThreadMutexLock(&pWal->mutex);
-  int64_t ver = walGetCommittedVer(pWal);
-  pRef->refVer = ver;
-
-  taosThreadMutexUnlock(&pWal->mutex);
-  wDebug("vgId:%d, wal ref version %" PRId64 " for committed", pWal->cfg.vgId, ver);
-}
+//void walRefCommitVer(SWal *pWal, SWalRef *pRef) {
+//  taosThreadMutexLock(&pWal->mutex);
+//  int64_t ver = walGetCommittedVer(pWal);
+//  pRef->refVer = ver;
+//
+//  taosThreadMutexUnlock(&pWal->mutex);
+//  wDebug("vgId:%d, wal ref version %" PRId64 " for committed", pWal->cfg.vgId, ver);
+//}