diff --git a/source/dnode/mnode/impl/inc/mndDef.h b/source/dnode/mnode/impl/inc/mndDef.h
index 62beee0303..4d1fccf690 100644
--- a/source/dnode/mnode/impl/inc/mndDef.h
+++ b/source/dnode/mnode/impl/inc/mndDef.h
@@ -137,48 +137,48 @@ typedef enum {
 } EDndReason;
 
 typedef enum {
-  CONSUMER_UPDATE__TOUCH = 1,   // rebalance req do not need change consume topic
+  CONSUMER_UPDATE__TOUCH = 1,  // rebalance req do not need change consume topic
   CONSUMER_UPDATE__ADD,
   CONSUMER_UPDATE__REMOVE,
   CONSUMER_UPDATE__LOST,
   CONSUMER_UPDATE__RECOVER,
-  CONSUMER_UPDATE__REBALANCE,      // subscribe req need change consume topic
+  CONSUMER_UPDATE__REBALANCE,  // subscribe req need change consume topic
 } ECsmUpdateType;
 
 typedef struct {
-  int32_t     id;
-  ETrnStage   stage;
-  ETrnPolicy  policy;
-  ETrnConflct conflict;
-  ETrnExec    exec;
-  EOperType   oper;
-  int32_t     code;
-  int32_t     failedTimes;
-  void*       rpcRsp;
-  int32_t     rpcRspLen;
-  int32_t     redoActionPos;
-  SArray*     prepareActions;
-  SArray*     redoActions;
-  SArray*     undoActions;
-  SArray*     commitActions;
-  int64_t     createdTime;
-  int64_t     lastExecTime;
-  int32_t     lastAction;
-  int32_t     lastErrorNo;
-  SEpSet      lastEpset;
-  tmsg_t      lastMsgType;
-  tmsg_t      originRpcType;
-  char        dbname[TSDB_TABLE_FNAME_LEN];
-  char        stbname[TSDB_TABLE_FNAME_LEN];
-  int32_t     startFunc;
-  int32_t     stopFunc;
-  int32_t     paramLen;
-  void*       param;
-  char        opername[TSDB_TRANS_OPER_LEN];
-  SArray*     pRpcArray;
-  SRWLatch    lockRpcArray;
-  int64_t     mTraceId;
-  TdThreadMutex    mutex;
+  int32_t       id;
+  ETrnStage     stage;
+  ETrnPolicy    policy;
+  ETrnConflct   conflict;
+  ETrnExec      exec;
+  EOperType     oper;
+  int32_t       code;
+  int32_t       failedTimes;
+  void*         rpcRsp;
+  int32_t       rpcRspLen;
+  int32_t       redoActionPos;
+  SArray*       prepareActions;
+  SArray*       redoActions;
+  SArray*       undoActions;
+  SArray*       commitActions;
+  int64_t       createdTime;
+  int64_t       lastExecTime;
+  int32_t       lastAction;
+  int32_t       lastErrorNo;
+  SEpSet        lastEpset;
+  tmsg_t        lastMsgType;
+  tmsg_t        originRpcType;
+  char          dbname[TSDB_TABLE_FNAME_LEN];
+  char          stbname[TSDB_TABLE_FNAME_LEN];
+  int32_t       startFunc;
+  int32_t       stopFunc;
+  int32_t       paramLen;
+  void*         param;
+  char          opername[TSDB_TRANS_OPER_LEN];
+  SArray*       pRpcArray;
+  SRWLatch      lockRpcArray;
+  int64_t       mTraceId;
+  TdThreadMutex mutex;
 } STrans;
 
 typedef struct {
@@ -445,20 +445,20 @@ typedef struct {
 } SStbObj;
 
 typedef struct {
-  char    name[TSDB_FUNC_NAME_LEN];
-  int64_t createdTime;
-  int8_t  funcType;
-  int8_t  scriptType;
-  int8_t  align;
-  int8_t  outputType;
-  int32_t outputLen;
-  int32_t bufSize;
-  int64_t signature;
-  int32_t commentSize;
-  int32_t codeSize;
-  char*   pComment;
-  char*   pCode;
-  int32_t funcVersion;
+  char     name[TSDB_FUNC_NAME_LEN];
+  int64_t  createdTime;
+  int8_t   funcType;
+  int8_t   scriptType;
+  int8_t   align;
+  int8_t   outputType;
+  int32_t  outputLen;
+  int32_t  bufSize;
+  int64_t  signature;
+  int32_t  commentSize;
+  int32_t  codeSize;
+  char*    pComment;
+  char*    pCode;
+  int32_t  funcVersion;
   SRWLatch lock;
 } SFuncObj;
 
@@ -552,11 +552,11 @@ typedef struct {
   int64_t subscribeTime;
   int64_t rebalanceTime;
 
-  int8_t         withTbName;
-  int8_t         useSnapshot;
-  int8_t         autoCommit;
-  int32_t        autoCommitInterval;
-  int32_t        resetOffsetCfg;
+  int8_t  withTbName;
+  int8_t  useSnapshot;
+  int8_t  autoCommit;
+  int32_t autoCommitInterval;
+  int32_t resetOffsetCfg;
 } SMqConsumerObj;
 
 SMqConsumerObj* tNewSMqConsumerObj(int64_t consumerId, char cgroup[TSDB_CGROUP_LEN]);
@@ -566,8 +566,8 @@ void*           tDecodeSMqConsumerObj(const void* buf, SMqConsumerObj* pConsumer
 
 typedef struct {
   int32_t vgId;
-//  char*   qmsg;  // SubPlanToString
-  SEpSet  epSet;
+  //  char*   qmsg;  // SubPlanToString
+  SEpSet epSet;
 } SMqVgEp;
 
 SMqVgEp* tCloneSMqVgEp(const SMqVgEp* pVgEp);
@@ -581,10 +581,10 @@ typedef struct {
   SArray* offsetRows;  // SArray<OffsetRows*>
 } SMqConsumerEp;
 
-//SMqConsumerEp* tCloneSMqConsumerEp(const SMqConsumerEp* pEp);
-//void           tDeleteSMqConsumerEp(void* pEp);
-int32_t        tEncodeSMqConsumerEp(void** buf, const SMqConsumerEp* pEp);
-void*          tDecodeSMqConsumerEp(const void* buf, SMqConsumerEp* pEp, int8_t sver);
+// SMqConsumerEp* tCloneSMqConsumerEp(const SMqConsumerEp* pEp);
+// void           tDeleteSMqConsumerEp(void* pEp);
+int32_t tEncodeSMqConsumerEp(void** buf, const SMqConsumerEp* pEp);
+void*   tDecodeSMqConsumerEp(const void* buf, SMqConsumerEp* pEp, int8_t sver);
 
 typedef struct {
   char      key[TSDB_SUBSCRIBE_KEY_LEN];
@@ -598,7 +598,7 @@ typedef struct {
   SArray*   unassignedVgs;  // SArray<SMqVgEp*>
   SArray*   offsetRows;
   char      dbName[TSDB_DB_FNAME_LEN];
-  char*     qmsg;           // SubPlanToString
+  char*     qmsg;  // SubPlanToString
 } SMqSubscribeObj;
 
 SMqSubscribeObj* tNewSubscribeObj(const char key[TSDB_SUBSCRIBE_KEY_LEN]);
@@ -675,7 +675,7 @@ typedef struct {
   int64_t targetStbUid;
 
   // fixedSinkVg is not applicable for encode and decode
-  SVgObj fixedSinkVg;
+  SVgObj  fixedSinkVg;
   int32_t fixedSinkVgId;  // 0 for shuffle
 
   // transformation
@@ -691,18 +691,21 @@ typedef struct {
   int64_t currentTick;     // do not serialize
   int64_t deleteMark;
   int8_t  igCheckUpdate;
+
+  // 3.0.5.
+  int64_t checkpointId;
 } SStreamObj;
 
 int32_t tEncodeSStreamObj(SEncoder* pEncoder, const SStreamObj* pObj);
 int32_t tDecodeSStreamObj(SDecoder* pDecoder, SStreamObj* pObj, int32_t sver);
 void    tFreeStreamObj(SStreamObj* pObj);
 
-//typedef struct {
-//  char    streamName[TSDB_STREAM_FNAME_LEN];
-//  int64_t uid;
-//  int64_t streamUid;
-//  SArray* childInfo;  // SArray<SStreamChildEpInfo>
-//} SStreamCheckpointObj;
+// typedef struct {
+//   char    streamName[TSDB_STREAM_FNAME_LEN];
+//   int64_t uid;
+//   int64_t streamUid;
+//   SArray* childInfo;  // SArray<SStreamChildEpInfo>
+// } SStreamCheckpointObj;
 
 #ifdef __cplusplus
 }
diff --git a/source/dnode/mnode/impl/src/mndDef.c b/source/dnode/mnode/impl/src/mndDef.c
index f0fa40cacf..c7b1fec5f2 100644
--- a/source/dnode/mnode/impl/src/mndDef.c
+++ b/source/dnode/mnode/impl/src/mndDef.c
@@ -80,6 +80,9 @@ int32_t tEncodeSStreamObj(SEncoder *pEncoder, const SStreamObj *pObj) {
   if (tEncodeI64(pEncoder, pObj->checkpointFreq) < 0) return -1;
   if (tEncodeI8(pEncoder, pObj->igCheckUpdate) < 0) return -1;
 
+  // 3.0.50
+  if (tEncodeI64(pEncoder, pObj->checkpointId) < 0) return -1;
+
   tEndEncode(pEncoder);
   return pEncoder->pos;
 }
@@ -150,6 +153,9 @@ int32_t tDecodeSStreamObj(SDecoder *pDecoder, SStreamObj *pObj, int32_t sver) {
       if (tDecodeI8(pDecoder, &pObj->igCheckUpdate) < 0) return -1;
     }
   }
+  if (sver >= 3) {
+    if (tDecodeI64(pDecoder, &pObj->checkpointId) < 0) return -1;
+  }
   tEndDecode(pDecoder);
   return 0;
 }
@@ -187,14 +193,14 @@ SMqVgEp *tCloneSMqVgEp(const SMqVgEp *pVgEp) {
   SMqVgEp *pVgEpNew = taosMemoryMalloc(sizeof(SMqVgEp));
   if (pVgEpNew == NULL) return NULL;
   pVgEpNew->vgId = pVgEp->vgId;
-//  pVgEpNew->qmsg = taosStrdup(pVgEp->qmsg);
+  //  pVgEpNew->qmsg = taosStrdup(pVgEp->qmsg);
   pVgEpNew->epSet = pVgEp->epSet;
   return pVgEpNew;
 }
 
 void tDeleteSMqVgEp(SMqVgEp *pVgEp) {
   if (pVgEp) {
-//    taosMemoryFreeClear(pVgEp->qmsg);
+    //    taosMemoryFreeClear(pVgEp->qmsg);
     taosMemoryFree(pVgEp);
   }
 }
@@ -202,14 +208,14 @@ void tDeleteSMqVgEp(SMqVgEp *pVgEp) {
 int32_t tEncodeSMqVgEp(void **buf, const SMqVgEp *pVgEp) {
   int32_t tlen = 0;
   tlen += taosEncodeFixedI32(buf, pVgEp->vgId);
-//  tlen += taosEncodeString(buf, pVgEp->qmsg);
+  //  tlen += taosEncodeString(buf, pVgEp->qmsg);
   tlen += taosEncodeSEpSet(buf, &pVgEp->epSet);
   return tlen;
 }
 
 void *tDecodeSMqVgEp(const void *buf, SMqVgEp *pVgEp, int8_t sver) {
   buf = taosDecodeFixedI32(buf, &pVgEp->vgId);
-  if(sver == 1){
+  if (sver == 1) {
     uint64_t size = 0;
     buf = taosDecodeVariantU64(buf, &size);
     buf = POINTER_SHIFT(buf, size);
@@ -384,7 +390,7 @@ void *tDecodeSMqConsumerObj(const void *buf, SMqConsumerObj *pConsumer, int8_t s
     taosArrayPush(pConsumer->assignedTopics, &topic);
   }
 
-  if(sver > 1){
+  if (sver > 1) {
     buf = taosDecodeFixedI8(buf, &pConsumer->withTbName);
     buf = taosDecodeFixedI8(buf, &pConsumer->useSnapshot);
     buf = taosDecodeFixedI8(buf, &pConsumer->autoCommit);
@@ -394,18 +400,18 @@ void *tDecodeSMqConsumerObj(const void *buf, SMqConsumerObj *pConsumer, int8_t s
   return (void *)buf;
 }
 
-//SMqConsumerEp *tCloneSMqConsumerEp(const SMqConsumerEp *pConsumerEpOld) {
-//  SMqConsumerEp *pConsumerEpNew = taosMemoryMalloc(sizeof(SMqConsumerEp));
-//  if (pConsumerEpNew == NULL) return NULL;
-//  pConsumerEpNew->consumerId = pConsumerEpOld->consumerId;
-//  pConsumerEpNew->vgs = taosArrayDup(pConsumerEpOld->vgs, NULL);
-//  return pConsumerEpNew;
-//}
+// SMqConsumerEp *tCloneSMqConsumerEp(const SMqConsumerEp *pConsumerEpOld) {
+//   SMqConsumerEp *pConsumerEpNew = taosMemoryMalloc(sizeof(SMqConsumerEp));
+//   if (pConsumerEpNew == NULL) return NULL;
+//   pConsumerEpNew->consumerId = pConsumerEpOld->consumerId;
+//   pConsumerEpNew->vgs = taosArrayDup(pConsumerEpOld->vgs, NULL);
+//   return pConsumerEpNew;
+// }
 //
-//void tDeleteSMqConsumerEp(void *data) {
-//  SMqConsumerEp *pConsumerEp = (SMqConsumerEp *)data;
-//  taosArrayDestroy(pConsumerEp->vgs);
-//}
+// void tDeleteSMqConsumerEp(void *data) {
+//   SMqConsumerEp *pConsumerEp = (SMqConsumerEp *)data;
+//   taosArrayDestroy(pConsumerEp->vgs);
+// }
 
 int32_t tEncodeSMqConsumerEp(void **buf, const SMqConsumerEp *pConsumerEp) {
   int32_t tlen = 0;
@@ -413,7 +419,7 @@ int32_t tEncodeSMqConsumerEp(void **buf, const SMqConsumerEp *pConsumerEp) {
   tlen += taosEncodeArray(buf, pConsumerEp->vgs, (FEncode)tEncodeSMqVgEp);
   int32_t szVgs = taosArrayGetSize(pConsumerEp->offsetRows);
   tlen += taosEncodeFixedI32(buf, szVgs);
-  for (int32_t j= 0; j < szVgs; ++j) {
+  for (int32_t j = 0; j < szVgs; ++j) {
     OffsetRows *offRows = taosArrayGet(pConsumerEp->offsetRows, j);
     tlen += taosEncodeFixedI32(buf, offRows->vgId);
     tlen += taosEncodeFixedI64(buf, offRows->rows);
@@ -427,28 +433,28 @@ int32_t tEncodeSMqConsumerEp(void **buf, const SMqConsumerEp *pConsumerEp) {
       // do nothing
     }
   }
-//#if 0
-//  int32_t sz = taosArrayGetSize(pConsumerEp->vgs);
-//  tlen += taosEncodeFixedI32(buf, sz);
-//  for (int32_t i = 0; i < sz; i++) {
-//    SMqVgEp *pVgEp = taosArrayGetP(pConsumerEp->vgs, i);
-//    tlen += tEncodeSMqVgEp(buf, pVgEp);
-//  }
-//#endif
+  // #if 0
+  //   int32_t sz = taosArrayGetSize(pConsumerEp->vgs);
+  //   tlen += taosEncodeFixedI32(buf, sz);
+  //   for (int32_t i = 0; i < sz; i++) {
+  //     SMqVgEp *pVgEp = taosArrayGetP(pConsumerEp->vgs, i);
+  //     tlen += tEncodeSMqVgEp(buf, pVgEp);
+  //   }
+  // #endif
   return tlen;
 }
 
 void *tDecodeSMqConsumerEp(const void *buf, SMqConsumerEp *pConsumerEp, int8_t sver) {
   buf = taosDecodeFixedI64(buf, &pConsumerEp->consumerId);
   buf = taosDecodeArray(buf, &pConsumerEp->vgs, (FDecode)tDecodeSMqVgEp, sizeof(SMqVgEp), sver);
-  if (sver > 1){
+  if (sver > 1) {
     int32_t szVgs = 0;
     buf = taosDecodeFixedI32(buf, &szVgs);
-    if(szVgs > 0){
+    if (szVgs > 0) {
       pConsumerEp->offsetRows = taosArrayInit(szVgs, sizeof(OffsetRows));
       if (NULL == pConsumerEp->offsetRows) return NULL;
-      for (int32_t j= 0; j < szVgs; ++j) {
-        OffsetRows* offRows = taosArrayReserve(pConsumerEp->offsetRows, 1);
+      for (int32_t j = 0; j < szVgs; ++j) {
+        OffsetRows *offRows = taosArrayReserve(pConsumerEp->offsetRows, 1);
         buf = taosDecodeFixedI32(buf, &offRows->vgId);
         buf = taosDecodeFixedI64(buf, &offRows->rows);
         buf = taosDecodeFixedI8(buf, &offRows->offset.type);
@@ -463,21 +469,21 @@ void *tDecodeSMqConsumerEp(const void *buf, SMqConsumerEp *pConsumerEp, int8_t s
       }
     }
   }
-//#if 0
-//  int32_t sz;
-//  buf = taosDecodeFixedI32(buf, &sz);
-//  pConsumerEp->vgs = taosArrayInit(sz, sizeof(void *));
-//  for (int32_t i = 0; i < sz; i++) {
-//    SMqVgEp *pVgEp = taosMemoryMalloc(sizeof(SMqVgEp));
-//    buf = tDecodeSMqVgEp(buf, pVgEp);
-//    taosArrayPush(pConsumerEp->vgs, &pVgEp);
-//  }
-//#endif
+  // #if 0
+  //   int32_t sz;
+  //   buf = taosDecodeFixedI32(buf, &sz);
+  //   pConsumerEp->vgs = taosArrayInit(sz, sizeof(void *));
+  //   for (int32_t i = 0; i < sz; i++) {
+  //     SMqVgEp *pVgEp = taosMemoryMalloc(sizeof(SMqVgEp));
+  //     buf = tDecodeSMqVgEp(buf, pVgEp);
+  //     taosArrayPush(pConsumerEp->vgs, &pVgEp);
+  //   }
+  // #endif
 
   return (void *)buf;
 }
 
-SMqSubscribeObj *tNewSubscribeObj(const char* key) {
+SMqSubscribeObj *tNewSubscribeObj(const char *key) {
   SMqSubscribeObj *pSubObj = taosMemoryCalloc(1, sizeof(SMqSubscribeObj));
   if (pSubObj == NULL) {
     return NULL;
@@ -570,7 +576,7 @@ int32_t tEncodeSubscribeObj(void **buf, const SMqSubscribeObj *pSub) {
 
   int32_t szVgs = taosArrayGetSize(pSub->offsetRows);
   tlen += taosEncodeFixedI32(buf, szVgs);
-  for (int32_t j= 0; j < szVgs; ++j) {
+  for (int32_t j = 0; j < szVgs; ++j) {
     OffsetRows *offRows = taosArrayGet(pSub->offsetRows, j);
     tlen += taosEncodeFixedI32(buf, offRows->vgId);
     tlen += taosEncodeFixedI64(buf, offRows->rows);
@@ -610,14 +616,14 @@ void *tDecodeSubscribeObj(const void *buf, SMqSubscribeObj *pSub, int8_t sver) {
   buf = taosDecodeArray(buf, &pSub->unassignedVgs, (FDecode)tDecodeSMqVgEp, sizeof(SMqVgEp), sver);
   buf = taosDecodeStringTo(buf, pSub->dbName);
 
-  if (sver > 1){
+  if (sver > 1) {
     int32_t szVgs = 0;
     buf = taosDecodeFixedI32(buf, &szVgs);
-    if(szVgs > 0){
+    if (szVgs > 0) {
       pSub->offsetRows = taosArrayInit(szVgs, sizeof(OffsetRows));
       if (NULL == pSub->offsetRows) return NULL;
-      for (int32_t j= 0; j < szVgs; ++j) {
-        OffsetRows* offRows = taosArrayReserve(pSub->offsetRows, 1);
+      for (int32_t j = 0; j < szVgs; ++j) {
+        OffsetRows *offRows = taosArrayReserve(pSub->offsetRows, 1);
         buf = taosDecodeFixedI32(buf, &offRows->vgId);
         buf = taosDecodeFixedI64(buf, &offRows->rows);
         buf = taosDecodeFixedI8(buf, &offRows->offset.type);
@@ -636,65 +642,65 @@ void *tDecodeSubscribeObj(const void *buf, SMqSubscribeObj *pSub, int8_t sver) {
   return (void *)buf;
 }
 
-//SMqSubActionLogEntry *tCloneSMqSubActionLogEntry(SMqSubActionLogEntry *pEntry) {
-//  SMqSubActionLogEntry *pEntryNew = taosMemoryMalloc(sizeof(SMqSubActionLogEntry));
-//  if (pEntryNew == NULL) return NULL;
-//  pEntryNew->epoch = pEntry->epoch;
-//  pEntryNew->consumers = taosArrayDup(pEntry->consumers, (__array_item_dup_fn_t)tCloneSMqConsumerEp);
-//  return pEntryNew;
-//}
+// SMqSubActionLogEntry *tCloneSMqSubActionLogEntry(SMqSubActionLogEntry *pEntry) {
+//   SMqSubActionLogEntry *pEntryNew = taosMemoryMalloc(sizeof(SMqSubActionLogEntry));
+//   if (pEntryNew == NULL) return NULL;
+//   pEntryNew->epoch = pEntry->epoch;
+//   pEntryNew->consumers = taosArrayDup(pEntry->consumers, (__array_item_dup_fn_t)tCloneSMqConsumerEp);
+//   return pEntryNew;
+// }
 //
-//void tDeleteSMqSubActionLogEntry(SMqSubActionLogEntry *pEntry) {
-//  taosArrayDestroyEx(pEntry->consumers, (FDelete)tDeleteSMqConsumerEp);
-//}
+// void tDeleteSMqSubActionLogEntry(SMqSubActionLogEntry *pEntry) {
+//   taosArrayDestroyEx(pEntry->consumers, (FDelete)tDeleteSMqConsumerEp);
+// }
 
-//int32_t tEncodeSMqSubActionLogEntry(void **buf, const SMqSubActionLogEntry *pEntry) {
-//  int32_t tlen = 0;
-//  tlen += taosEncodeFixedI32(buf, pEntry->epoch);
-//  tlen += taosEncodeArray(buf, pEntry->consumers, (FEncode)tEncodeSMqSubActionLogEntry);
-//  return tlen;
-//}
+// int32_t tEncodeSMqSubActionLogEntry(void **buf, const SMqSubActionLogEntry *pEntry) {
+//   int32_t tlen = 0;
+//   tlen += taosEncodeFixedI32(buf, pEntry->epoch);
+//   tlen += taosEncodeArray(buf, pEntry->consumers, (FEncode)tEncodeSMqSubActionLogEntry);
+//   return tlen;
+// }
 //
-//void *tDecodeSMqSubActionLogEntry(const void *buf, SMqSubActionLogEntry *pEntry) {
-//  buf = taosDecodeFixedI32(buf, &pEntry->epoch);
-//  buf = taosDecodeArray(buf, &pEntry->consumers, (FDecode)tDecodeSMqSubActionLogEntry, sizeof(SMqSubActionLogEntry));
-//  return (void *)buf;
-//}
+// void *tDecodeSMqSubActionLogEntry(const void *buf, SMqSubActionLogEntry *pEntry) {
+//   buf = taosDecodeFixedI32(buf, &pEntry->epoch);
+//   buf = taosDecodeArray(buf, &pEntry->consumers, (FDecode)tDecodeSMqSubActionLogEntry, sizeof(SMqSubActionLogEntry));
+//   return (void *)buf;
+// }
 
-//SMqSubActionLogObj *tCloneSMqSubActionLogObj(SMqSubActionLogObj *pLog) {
-//  SMqSubActionLogObj *pLogNew = taosMemoryMalloc(sizeof(SMqSubActionLogObj));
-//  if (pLogNew == NULL) return pLogNew;
-//  memcpy(pLogNew->key, pLog->key, TSDB_SUBSCRIBE_KEY_LEN);
-//  pLogNew->logs = taosArrayDup(pLog->logs, (__array_item_dup_fn_t)tCloneSMqConsumerEp);
-//  return pLogNew;
-//}
+// SMqSubActionLogObj *tCloneSMqSubActionLogObj(SMqSubActionLogObj *pLog) {
+//   SMqSubActionLogObj *pLogNew = taosMemoryMalloc(sizeof(SMqSubActionLogObj));
+//   if (pLogNew == NULL) return pLogNew;
+//   memcpy(pLogNew->key, pLog->key, TSDB_SUBSCRIBE_KEY_LEN);
+//   pLogNew->logs = taosArrayDup(pLog->logs, (__array_item_dup_fn_t)tCloneSMqConsumerEp);
+//   return pLogNew;
+// }
 //
-//void tDeleteSMqSubActionLogObj(SMqSubActionLogObj *pLog) {
-//  taosArrayDestroyEx(pLog->logs, (FDelete)tDeleteSMqConsumerEp);
-//}
+// void tDeleteSMqSubActionLogObj(SMqSubActionLogObj *pLog) {
+//   taosArrayDestroyEx(pLog->logs, (FDelete)tDeleteSMqConsumerEp);
+// }
 
-//int32_t tEncodeSMqSubActionLogObj(void **buf, const SMqSubActionLogObj *pLog) {
-//  int32_t tlen = 0;
-//  tlen += taosEncodeString(buf, pLog->key);
-//  tlen += taosEncodeArray(buf, pLog->logs, (FEncode)tEncodeSMqSubActionLogEntry);
-//  return tlen;
-//}
+// int32_t tEncodeSMqSubActionLogObj(void **buf, const SMqSubActionLogObj *pLog) {
+//   int32_t tlen = 0;
+//   tlen += taosEncodeString(buf, pLog->key);
+//   tlen += taosEncodeArray(buf, pLog->logs, (FEncode)tEncodeSMqSubActionLogEntry);
+//   return tlen;
+// }
 //
-//void *tDecodeSMqSubActionLogObj(const void *buf, SMqSubActionLogObj *pLog) {
-//  buf = taosDecodeStringTo(buf, pLog->key);
-//  buf = taosDecodeArray(buf, &pLog->logs, (FDecode)tDecodeSMqSubActionLogEntry, sizeof(SMqSubActionLogEntry));
-//  return (void *)buf;
-//}
+// void *tDecodeSMqSubActionLogObj(const void *buf, SMqSubActionLogObj *pLog) {
+//   buf = taosDecodeStringTo(buf, pLog->key);
+//   buf = taosDecodeArray(buf, &pLog->logs, (FDecode)tDecodeSMqSubActionLogEntry, sizeof(SMqSubActionLogEntry));
+//   return (void *)buf;
+// }
 //
-//int32_t tEncodeSMqOffsetObj(void **buf, const SMqOffsetObj *pOffset) {
-//  int32_t tlen = 0;
-//  tlen += taosEncodeString(buf, pOffset->key);
-//  tlen += taosEncodeFixedI64(buf, pOffset->offset);
-//  return tlen;
-//}
+// int32_t tEncodeSMqOffsetObj(void **buf, const SMqOffsetObj *pOffset) {
+//   int32_t tlen = 0;
+//   tlen += taosEncodeString(buf, pOffset->key);
+//   tlen += taosEncodeFixedI64(buf, pOffset->offset);
+//   return tlen;
+// }
 //
-//void *tDecodeSMqOffsetObj(void *buf, SMqOffsetObj *pOffset) {
-//  buf = taosDecodeStringTo(buf, pOffset->key);
-//  buf = taosDecodeFixedI64(buf, &pOffset->offset);
-//  return buf;
-//}
+// void *tDecodeSMqOffsetObj(void *buf, SMqOffsetObj *pOffset) {
+//   buf = taosDecodeStringTo(buf, pOffset->key);
+//   buf = taosDecodeFixedI64(buf, &pOffset->offset);
+//   return buf;
+// }
diff --git a/source/dnode/mnode/impl/src/mndStream.c b/source/dnode/mnode/impl/src/mndStream.c
index fc12fe8f25..2b0bc98a7d 100644
--- a/source/dnode/mnode/impl/src/mndStream.c
+++ b/source/dnode/mnode/impl/src/mndStream.c
@@ -28,7 +28,7 @@
 #include "parser.h"
 #include "tname.h"
 
-#define MND_STREAM_VER_NUMBER   2
+#define MND_STREAM_VER_NUMBER   3
 #define MND_STREAM_RESERVE_SIZE 64
 
 #define MND_STREAM_MAX_NUM 60
@@ -147,7 +147,7 @@ SSdbRow *mndStreamActionDecode(SSdbRaw *pRaw) {
   int8_t sver = 0;
   if (sdbGetRawSoftVer(pRaw, &sver) != 0) goto STREAM_DECODE_OVER;
 
-  if (sver != 1 && sver != 2) {
+  if (sver != 1 && sver != 2 && sver != 3) {
     terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
     goto STREAM_DECODE_OVER;
   }
@@ -946,6 +946,9 @@ static int32_t mndBuildStreamCheckpointSourceReq2(void **pBuf, int32_t *pLen, in
 }
 static int32_t mndProcessStreamCheckpointTrans(SMnode *pMnode, SStreamObj *pStream, SHashObj *vgIds,
                                                int64_t checkpointId) {
+  if (checkpointId == pStream->checkpointId) {
+    return -1;
+  }
   STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, NULL, "stream-checkpoint");
   if (pTrans == NULL) return -1;
   mndTransSetDbName(pTrans, pStream->sourceDb, pStream->targetDb);
@@ -955,7 +958,8 @@ static int32_t mndProcessStreamCheckpointTrans(SMnode *pMnode, SStreamObj *pStre
     mndTransDrop(pTrans);
     return -1;
   }
-  taosRLockLatch(&pStream->lock);
+  atomic_store_64(&pStream->currentTick, 1);
+  taosWLockLatch(&pStream->lock);
   // 1. redo action: broadcast checkpoint source msg for all source vg
   int32_t totLevel = taosArrayGetSize(pStream->tasks);
   for (int32_t i = 0; i < totLevel; i++) {
@@ -1003,10 +1007,11 @@ static int32_t mndProcessStreamCheckpointTrans(SMnode *pMnode, SStreamObj *pStre
   }
   // 2. reset tick
   pStream->checkpointFreq = checkpointId;
+  pStream->checkpointId = checkpointId;
   atomic_store_64(&pStream->currentTick, 0);
   // 3. commit log: stream checkpoint info
-
-  taosRUnLockLatch(&pStream->lock);
+  pStream->version = pStream->version + 1;
+  taosWUnLockLatch(&pStream->lock);
 
   //   // code condtion
 
@@ -1051,6 +1056,9 @@ static int32_t mndProcessStreamDoCheckpoint(SRpcMsg *pReq) {
     pIter = sdbFetch(pSdb, SDB_STREAM, pIter, (void **)&pStream);
     if (pIter == NULL) break;
     code = mndProcessStreamCheckpointTrans(pMnode, pStream, vgIds, checkpointId);
+    if (code == -1) {
+      mInfo("stream:%s failed to do checkpoint, reason: last checkpoint not finished", pStream->name);
+    }
     sdbRelease(pSdb, pStream);
   }
   taosHashCleanup(vgIds);