refactor(tmq): subscribe and rebalance process

include/common/tmsg.h

@@ -329,7 +329,7 @@ typedef struct SEpSet {
 int32_t tEncodeSEpSet(SCoder* pEncoder, const SEpSet* pEp);
 int32_t tDecodeSEpSet(SCoder* pDecoder, SEpSet* pEp);
 int32_t taosEncodeSEpSet(void** buf, const SEpSet* pEp);
-void*   taosDecodeSEpSet(void* buf, SEpSet* pEp);
+void*   taosDecodeSEpSet(const void* buf, SEpSet* pEp);
 
 typedef struct {
   int8_t  connType;
@@ -1290,10 +1290,14 @@ typedef struct {
 int32_t tSerializeSCMCreateTopicRsp(void* buf, int32_t bufLen, const SCMCreateTopicRsp* pRsp);
 int32_t tDeserializeSCMCreateTopicRsp(void* buf, int32_t bufLen, SCMCreateTopicRsp* pRsp);
 
+typedef struct {
+  int64_t consumerId;
+} SMqConsumerLostMsg;
+
 typedef struct {
   int32_t topicNum;
   int64_t consumerId;
-  char*   consumerGroup;
+  char    cgroup[TSDB_CGROUP_LEN];
   SArray* topicNames;  // SArray<char*>
 } SCMSubscribeReq;
 
@@ -1301,7 +1305,7 @@ static FORCE_INLINE int32_t tSerializeSCMSubscribeReq(void** buf, const SCMSubsc
   int32_t tlen = 0;
   tlen += taosEncodeFixedI32(buf, pReq->topicNum);
   tlen += taosEncodeFixedI64(buf, pReq->consumerId);
-  tlen += taosEncodeString(buf, pReq->consumerGroup);
+  tlen += taosEncodeString(buf, pReq->cgroup);
 
   for (int32_t i = 0; i < pReq->topicNum; i++) {
     tlen += taosEncodeString(buf, (char*)taosArrayGetP(pReq->topicNames, i));
@@ -1312,7 +1316,7 @@ static FORCE_INLINE int32_t tSerializeSCMSubscribeReq(void** buf, const SCMSubsc
 static FORCE_INLINE void* tDeserializeSCMSubscribeReq(void* buf, SCMSubscribeReq* pReq) {
   buf = taosDecodeFixedI32(buf, &pReq->topicNum);
   buf = taosDecodeFixedI64(buf, &pReq->consumerId);
-  buf = taosDecodeString(buf, &pReq->consumerGroup);
+  buf = taosDecodeStringTo(buf, pReq->cgroup);
   pReq->topicNames = taosArrayInit(pReq->topicNum, sizeof(void*));
   for (int32_t i = 0; i < pReq->topicNum; i++) {
     char* name;
@@ -1388,10 +1392,10 @@ static FORCE_INLINE void* tDeserializeSMVSubscribeReq(void* buf, SMVSubscribeReq
 }
 
 typedef struct {
-  const char* key;
-  SArray*     lostConsumers;     // SArray<int64_t>
-  SArray*     removedConsumers;  // SArray<int64_t>
-  SArray*     newConsumers;      // SArray<int64_t>
+  char    key[TSDB_SUBSCRIBE_KEY_LEN];
+  SArray* lostConsumers;     // SArray<int64_t>
+  SArray* removedConsumers;  // SArray<int64_t>
+  SArray* newConsumers;      // SArray<int64_t>
 } SMqRebSubscribe;
 
 static FORCE_INLINE SMqRebSubscribe* tNewSMqRebSubscribe(const char* key) {
@@ -1399,7 +1403,7 @@ static FORCE_INLINE SMqRebSubscribe* tNewSMqRebSubscribe(const char* key) {
   if (pRebSub == NULL) {
     goto _err;
   }
-  pRebSub->key = strdup(key);
+  strcpy(pRebSub->key, key);
   pRebSub->lostConsumers = taosArrayInit(0, sizeof(int64_t));
   if (pRebSub->lostConsumers == NULL) {
     goto _err;
@@ -1424,6 +1428,7 @@ _err:
 // this message is sent from mnode to mnode(read thread to write thread), so there is no need for serialization or
 // deserialization
 typedef struct {
+  int8_t*   mqInReb;
   SHashObj* rebSubHash;  // SHashObj<key, SMqRebSubscribe>
 } SMqDoRebalanceMsg;
 
@@ -1876,6 +1881,40 @@ static FORCE_INLINE void* taosDecodeSMqMsg(void* buf, SMqHbMsg* pMsg) {
   return buf;
 }
 
+typedef struct {
+  int64_t leftForVer;
+  int32_t vgId;
+  int64_t oldConsumerId;
+  int64_t newConsumerId;
+  char    subKey[TSDB_SUBSCRIBE_KEY_LEN];
+  char*   qmsg;
+} SMqRebVgReq;
+
+static FORCE_INLINE int32_t tEncodeSMqRebVgReq(void** buf, const SMqRebVgReq* pReq) {
+  int32_t tlen = 0;
+  tlen += taosEncodeFixedI64(buf, pReq->leftForVer);
+  tlen += taosEncodeFixedI32(buf, pReq->vgId);
+  tlen += taosEncodeFixedI64(buf, pReq->oldConsumerId);
+  tlen += taosEncodeFixedI64(buf, pReq->newConsumerId);
+  tlen += taosEncodeString(buf, pReq->subKey);
+  tlen += taosEncodeString(buf, pReq->qmsg);
+  return tlen;
+}
+
+static FORCE_INLINE void* tDecodeSMqRebVgReq(const void* buf, SMqRebVgReq* pReq) {
+  buf = taosDecodeFixedI64(buf, &pReq->leftForVer);
+  buf = taosDecodeFixedI32(buf, &pReq->vgId);
+  buf = taosDecodeFixedI64(buf, &pReq->oldConsumerId);
+  buf = taosDecodeFixedI64(buf, &pReq->newConsumerId);
+  buf = taosDecodeStringTo(buf, pReq->subKey);
+  buf = taosDecodeString(buf, &pReq->qmsg);
+  return (void*)buf;
+}
+
+typedef struct {
+  int8_t reserved;
+} SMqRebVgRsp;
+
 typedef struct {
   int64_t leftForVer;
   int32_t vgId;
@@ -2441,7 +2480,10 @@ typedef struct {
   SArray*    topics;  // SArray<SMqSubTopicEp>
 } SMqCMGetSubEpRsp;
 
-static FORCE_INLINE void tDeleteSMqSubTopicEp(SMqSubTopicEp* pSubTopicEp) { taosArrayDestroy(pSubTopicEp->vgs); }
+static FORCE_INLINE void tDeleteSMqSubTopicEp(SMqSubTopicEp* pSubTopicEp) {
+  taosMemoryFree(pSubTopicEp->schema.pSchema);
+  taosArrayDestroy(pSubTopicEp->vgs);
+}
 
 static FORCE_INLINE int32_t tEncodeSMqSubVgEp(void** buf, const SMqSubVgEp* pVgEp) {
   int32_t tlen = 0;

include/common/tmsgdef.h

@@ -149,6 +149,7 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_MND_SUBSCRIBE, "mnode-subscribe", SCMSubscribeReq, SCMSubscribeRsp)
   TD_DEF_MSG_TYPE(TDMT_MND_GET_SUB_EP, "mnode-get-sub-ep", SMqCMGetSubEpReq, SMqCMGetSubEpRsp)
   TD_DEF_MSG_TYPE(TDMT_MND_MQ_TIMER, "mnode-mq-tmr", SMTimerReq, SMTimerReq)
+  TD_DEF_MSG_TYPE(TDMT_MND_MQ_CONSUMER_LOST, "mnode-mq-consumer-lost", SMTimerReq, SMTimerReq)
   TD_DEF_MSG_TYPE(TDMT_MND_MQ_DO_REBALANCE, "mnode-mq-do-rebalance", SMqDoRebalanceMsg, SMqDoRebalanceMsg)
   TD_DEF_MSG_TYPE(TDMT_MND_MQ_COMMIT_OFFSET, "mnode-mq-commit-offset", SMqCMCommitOffsetReq, SMqCMCommitOffsetRsp)
   TD_DEF_MSG_TYPE(TDMT_MND_CREATE_STREAM, "mnode-create-stream", SCMCreateStreamReq, SCMCreateStreamRsp)
@@ -180,6 +181,7 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_VND_MQ_SET_CONN, "vnode-mq-set-conn", SMqSetCVgReq, SMqSetCVgRsp)
   TD_DEF_MSG_TYPE(TDMT_VND_MQ_REB, "vnode-mq-mv-rebalance", SMqMVRebReq, SMqMVRebRsp)
   TD_DEF_MSG_TYPE(TDMT_VND_MQ_CANCEL_CONN, "vnode-mq-mv-cancel-conn", SMqCancelConnReq, SMqCancelConnRsp)
+  TD_DEF_MSG_TYPE(TDMT_VND_MQ_VG_CHANGE, "vnode-mq-vg-change", SMqRebVgReq, SMqRebVgRsp)
   TD_DEF_MSG_TYPE(TDMT_VND_MQ_SET_CUR, "vnode-mq-set-cur", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_RES_READY, "vnode-res-ready", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_TASKS_STATUS, "vnode-tasks-status", NULL, NULL)

include/dnode/mnode/sdb/sdb.h

@@ -92,7 +92,13 @@ extern "C" {
 typedef struct SMnode  SMnode;
 typedef struct SSdbRaw SSdbRaw;
 typedef struct SSdbRow SSdbRow;
-typedef enum { SDB_KEY_BINARY = 1, SDB_KEY_INT32 = 2, SDB_KEY_INT64 = 3 } EKeyType;
+
+typedef enum {
+  SDB_KEY_BINARY = 1,
+  SDB_KEY_INT32 = 2,
+  SDB_KEY_INT64 = 3,
+} EKeyType;
+
 typedef enum {
   SDB_STATUS_INIT = 0,
   SDB_STATUS_CREATING = 1,

include/util/talgo.h

@@ -27,6 +27,11 @@ extern "C" {
 typedef int32_t (*__compar_fn_t)(const void *, const void *);
 #endif
 
+typedef void *(*FCopy)(void *);
+typedef void (*FDelete)(void *);
+typedef int32_t (*FEncode)(void **buf, const void *dst);
+typedef void *(*FDecode)(const void *buf, void *dst);
+
 #define TD_EQ 0x1
 #define TD_GT 0x2
 #define TD_LT 0x4

include/util/taoserror.h

@@ -279,6 +279,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_MND_UNSUPPORTED_TOPIC         TAOS_DEF_ERROR_CODE(0, 0x03E8)
 #define TSDB_CODE_MND_SUBSCRIBE_NOT_EXIST       TAOS_DEF_ERROR_CODE(0, 0x03E9)
 #define TSDB_CODE_MND_OFFSET_NOT_EXIST          TAOS_DEF_ERROR_CODE(0, 0x03EA)
+#define TSDB_CODE_MND_CONSUMER_NOT_READY        TAOS_DEF_ERROR_CODE(0, 0x03EB)
 
 // mnode-stream
 #define TSDB_CODE_MND_STREAM_ALREADY_EXIST      TAOS_DEF_ERROR_CODE(0, 0x03F0)

include/util/tarray.h

@@ -41,10 +41,10 @@ extern "C" {
 #define TARRAY_GET_START(array)       ((array)->pData)
 
 typedef struct SArray {
-  size_t size;
+  size_t   size;
   uint32_t capacity;
   uint32_t elemSize;
-  void*  pData;
+  void*    pData;
 } SArray;
 
 /**
@@ -200,23 +200,20 @@ SArray* taosArrayFromList(const void* src, size_t size, size_t elemSize);
 SArray* taosArrayDup(const SArray* pSrc);
 
 /**
- * clear the array (remove all element)
- * @param pArray
+ * deep copy a new array
+ * @param pSrc
  */
-void taosArrayClear(SArray* pArray);
+SArray* taosArrayDeepCopy(const SArray* pSrc, FCopy deepCopy);
 
 /**
- * destroy array list
+ * clear the array (remove all element)
  * @param pArray
  */
-void* taosArrayDestroy(SArray* pArray);
+void taosArrayClear(SArray* pArray);
 
-/**
- *
- * @param pArray
- * @param fp
- */
-void taosArrayDestroyEx(SArray* pArray, void (*fp)(void*));
+void* taosArrayDestroy(SArray* pArray);
+void  taosArrayDestroyP(SArray* pArray, FDelete fp);
+void  taosArrayDestroyEx(SArray* pArray, FDelete fp);
 
 /**
  * sort the array
@@ -264,6 +261,9 @@ char* taosArraySearchString(const SArray* pArray, const char* key, __compar_fn_t
 
 void taosArraySortPWithExt(SArray* pArray, __ext_compar_fn_t fn, const void* param);
 
+int32_t taosEncodeArray(void** buf, const SArray* pArray, FEncode encode);
+void*   taosDecodeArray(const void* buf, SArray** pArray, FDecode decode, int32_t dataSz);
+
 #ifdef __cplusplus
 }
 #endif

source/client/src/tmq.c

@@ -255,7 +255,12 @@ int32_t tmq_list_append(tmq_list_t* list, const char* src) {
 void tmq_list_destroy(tmq_list_t* list) {
   SArray* container = &list->container;
   /*taosArrayDestroy(container);*/
-  taosArrayDestroyEx(container, (void (*)(void*))taosMemoryFree);
+  int32_t sz = taosArrayGetSize(container);
+  for (int32_t i = 0; i < sz; i++) {
+    char* str = taosArrayGetP(container, i);
+    taosMemoryFree(str);
+  }
+  taosArrayDestroy(container);
 }
 
 static int32_t tmqMakeTopicVgKey(char* dst, const char* topicName, int32_t vg) {
@@ -496,7 +501,7 @@ tmq_resp_err_t tmq_subscribe(tmq_t* tmq, tmq_list_t* topic_list) {
   SCMSubscribeReq req;
   req.topicNum = sz;
   req.consumerId = tmq->consumerId;
-  req.consumerGroup = strdup(tmq->groupId);
+  strcpy(req.cgroup, tmq->groupId);
   req.topicNames = taosArrayInit(sz, sizeof(void*));
 
   for (int i = 0; i < sz; i++) {

source/common/src/tmsg.c

@@ -121,14 +121,14 @@ int32_t taosEncodeSEpSet(void **buf, const SEpSet *pEp) {
   return tlen;
 }
 
-void *taosDecodeSEpSet(void *buf, SEpSet *pEp) {
+void *taosDecodeSEpSet(const void *buf, SEpSet *pEp) {
   buf = taosDecodeFixedI8(buf, &pEp->inUse);
   buf = taosDecodeFixedI8(buf, &pEp->numOfEps);
   for (int32_t i = 0; i < TSDB_MAX_REPLICA; i++) {
     buf = taosDecodeFixedU16(buf, &pEp->eps[i].port);
     buf = taosDecodeStringTo(buf, pEp->eps[i].fqdn);
   }
-  return buf;
+  return (void *)buf;
 }
 
 static int32_t tSerializeSClientHbReq(SCoder *pEncoder, const SClientHbReq *pReq) {
@@ -2184,7 +2184,7 @@ int32_t tSerializeSRetrieveTableReq(void *buf, int32_t bufLen, SRetrieveTableReq
   if (tStartEncode(&encoder) < 0) return -1;
   if (tEncodeI64(&encoder, pReq->showId) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->type) < 0) return -1;
-//  if (tEncodeI8(&encoder, pReq->free) < 0) return -1;
+  //  if (tEncodeI8(&encoder, pReq->free) < 0) return -1;
   if (tEncodeCStr(&encoder, pReq->db) < 0) return -1;
   if (tEncodeCStr(&encoder, pReq->tb) < 0) return -1;
   tEndEncode(&encoder);
@@ -2201,7 +2201,7 @@ int32_t tDeserializeSRetrieveTableReq(void *buf, int32_t bufLen, SRetrieveTableR
   if (tStartDecode(&decoder) < 0) return -1;
   if (tDecodeI64(&decoder, &pReq->showId) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->type) < 0) return -1;
-//  if (tDecodeI8(&decoder, &pReq->free) < 0) return -1;
+  //  if (tDecodeI8(&decoder, &pReq->free) < 0) return -1;
   if (tDecodeCStrTo(&decoder, pReq->db) < 0) return -1;
   if (tDecodeCStrTo(&decoder, pReq->tb) < 0) return -1;
   tEndDecode(&decoder);

source/dnode/mnode/impl/inc/mndConsumer.h

@@ -22,27 +22,30 @@
 extern "C" {
 #endif
 
-
 enum {
-  MQ_CONSUMER_STATUS__INIT = 1,
-  MQ_CONSUMER_STATUS__IDLE,
-  MQ_CONSUMER_STATUS__ACTIVE,
+  // MQ_CONSUMER_STATUS__INIT = 1,
+  MQ_CONSUMER_STATUS__MODIFY = 1,
+  MQ_CONSUMER_STATUS__MODIFY_IN_REB,
+  // MQ_CONSUMER_STATUS__IDLE,
+  MQ_CONSUMER_STATUS__READY,
   MQ_CONSUMER_STATUS__LOST,
-  MQ_CONSUMER_STATUS__MODIFY
+  MQ_CONSUMER_STATUS__LOST_IN_REB,
+  MQ_CONSUMER_STATUS__LOST_REBD,
 };
 
-
 int32_t mndInitConsumer(SMnode *pMnode);
 void    mndCleanupConsumer(SMnode *pMnode);
 
 SMqConsumerObj *mndAcquireConsumer(SMnode *pMnode, int64_t consumerId);
 void            mndReleaseConsumer(SMnode *pMnode, SMqConsumerObj *pConsumer);
 
-SMqConsumerObj* mndCreateConsumer(int64_t consumerId, const char* cgroup);
+SMqConsumerObj *mndCreateConsumer(int64_t consumerId, const char *cgroup);
 
 SSdbRaw *mndConsumerActionEncode(SMqConsumerObj *pConsumer);
 SSdbRow *mndConsumerActionDecode(SSdbRaw *pRaw);
 
+int32_t mndSetConsumerCommitLogs(SMnode *pMnode, STrans *pTrans, SMqConsumerObj *pConsumer);
+
 #ifdef __cplusplus
 }
 #endif

source/dnode/mnode/impl/inc/mndDef.h

@@ -88,6 +88,7 @@ typedef enum {
   TRN_TYPE_CREATE_STREAM = 1019,
   TRN_TYPE_DROP_STREAM = 1020,
   TRN_TYPE_ALTER_STREAM = 1021,
+  TRN_TYPE_CONSUMER_LOST = 1022,
   TRN_TYPE_BASIC_SCOPE_END,
   TRN_TYPE_GLOBAL_SCOPE = 2000,
   TRN_TYPE_CREATE_DNODE = 2001,
@@ -506,6 +507,7 @@ static FORCE_INLINE void* tDecodeSMqOffsetObj(void* buf, SMqOffsetObj* pOffset)
   return buf;
 }
 
+#if 0
 typedef struct {
   char    key[TSDB_SUBSCRIBE_KEY_LEN];
   int32_t status;
@@ -636,6 +638,7 @@ static FORCE_INLINE void tDeleteSMqSubscribeObj(SMqSubscribeObj* pSub) {
     pSub->unassignedVg = NULL;
   }
 }
+#endif
 
 typedef struct {
   char           name[TSDB_TOPIC_FNAME_LEN];
@@ -654,6 +657,7 @@ typedef struct {
   SSchemaWrapper schema;
 } SMqTopicObj;
 
+#if 0
 typedef struct {
   int64_t  consumerId;
   int64_t  connId;
@@ -708,7 +712,7 @@ static FORCE_INLINE void* tDecodeSMqConsumerObj(void* buf, SMqConsumerObj* pCons
   buf = taosDecodeStringTo(buf, pConsumer->cgroup);
 
   buf = taosDecodeFixedI32(buf, &sz);
-  pConsumer->currentTopics = taosArrayInit(sz, sizeof(SMqConsumerObj));
+  pConsumer->currentTopics = taosArrayInit(sz, sizeof(void*));
   for (int32_t i = 0; i < sz; i++) {
     char* topic;
     buf = taosDecodeString(buf, &topic);
@@ -716,7 +720,7 @@ static FORCE_INLINE void* tDecodeSMqConsumerObj(void* buf, SMqConsumerObj* pCons
   }
 
   buf = taosDecodeFixedI32(buf, &sz);
-  pConsumer->recentRemovedTopics = taosArrayInit(sz, sizeof(SMqConsumerObj));
+  pConsumer->recentRemovedTopics = taosArrayInit(sz, sizeof(void*));
   for (int32_t i = 0; i < sz; i++) {
     char* topic;
     buf = taosDecodeString(buf, &topic);
@@ -724,6 +728,132 @@ static FORCE_INLINE void* tDecodeSMqConsumerObj(void* buf, SMqConsumerObj* pCons
   }
   return buf;
 }
+#endif
+
+enum {
+  CONSUMER_UPDATE__TOUCH = 1,
+  CONSUMER_UPDATE__ADD,
+  CONSUMER_UPDATE__REMOVE,
+  CONSUMER_UPDATE__LOST,
+  CONSUMER_UPDATE__MODIFY,
+};
+
+typedef struct {
+  int64_t consumerId;
+  char    cgroup[TSDB_CGROUP_LEN];
+  int8_t  updateType;  // used only for update
+  int32_t epoch;
+  int32_t status;
+  // hbStatus is not applicable to serialization
+  int32_t hbStatus;
+  // lock is used for topics update
+  SRWLatch lock;
+  SArray*  currentTopics;  // SArray<char*>
+#if 0
+  SArray*  waitingRebTopics;  // SArray<char*>
+#endif
+  SArray* rebNewTopics;      // SArray<char*>
+  SArray* rebRemovedTopics;  // SArray<char*>
+} SMqConsumerObj;
+
+SMqConsumerObj* tNewSMqConsumerObj(int64_t consumerId, char cgroup[TSDB_CGROUP_LEN]);
+void            tDeleteSMqConsumerObj(SMqConsumerObj* pConsumer);
+int32_t         tEncodeSMqConsumerObj(void** buf, const SMqConsumerObj* pConsumer);
+void*           tDecodeSMqConsumerObj(const void* buf, SMqConsumerObj* pConsumer);
+
+typedef struct {
+  int32_t vgId;
+  char*   qmsg;
+  // char    topic[TSDB_TOPIC_FNAME_LEN];
+  SEpSet epSet;
+} SMqVgEp;
+
+SMqVgEp* tCloneSMqVgEp(const SMqVgEp* pVgEp);
+void     tDeleteSMqVgEp(SMqVgEp* pVgEp);
+int32_t  tEncodeSMqVgEp(void** buf, const SMqVgEp* pVgEp);
+void*    tDecodeSMqVgEp(const void* buf, SMqVgEp* pVgEp);
+
+typedef struct {
+  int64_t consumerId;  // -1 for unassigned
+  SArray* vgs;         // SArray<SMqVgEp*>
+} SMqConsumerEpInSub;
+
+SMqConsumerEpInSub* tCloneSMqConsumerEpInSub(const SMqConsumerEpInSub* pEpInSub);
+void                tDeleteSMqConsumerEpInSub(SMqConsumerEpInSub* pEpInSub);
+int32_t             tEncodeSMqConsumerEpInSub(void** buf, const SMqConsumerEpInSub* pEpInSub);
+void*               tDecodeSMqConsumerEpInSub(const void* buf, SMqConsumerEpInSub* pEpInSub);
+
+typedef struct {
+  char      key[TSDB_SUBSCRIBE_KEY_LEN];
+  SRWLatch  lock;
+  int32_t   vgNum;
+  SHashObj* consumerHash;  // consumerId -> SMqConsumerEpInSub
+} SMqSubscribeObj;
+
+SMqSubscribeObj* tNewSubscribeObj(const char key[TSDB_SUBSCRIBE_KEY_LEN]);
+SMqSubscribeObj* tCloneSubscribeObj(const SMqSubscribeObj* pSub);
+void             tDeleteSubscribeObj(SMqSubscribeObj* pSub);
+int32_t          tEncodeSubscribeObj(void** buf, const SMqSubscribeObj* pSub);
+void*            tDecodeSubscribeObj(const void* buf, SMqSubscribeObj* pSub);
+
+typedef struct {
+  int32_t epoch;
+  SArray* consumers;  // SArray<SMqConsumerEpInSub*>
+} SMqSubActionLogEntry;
+
+SMqSubActionLogEntry* tCloneSMqSubActionLogEntry(SMqSubActionLogEntry* pEntry);
+void                  tDeleteSMqSubActionLogEntry(SMqSubActionLogEntry* pEntry);
+int32_t               tEncodeSMqSubActionLogEntry(void** buf, const SMqSubActionLogEntry* pEntry);
+void*                 tDecodeSMqSubActionLogEntry(const void* buf, SMqSubActionLogEntry* pEntry);
+
+typedef struct {
+  char    key[TSDB_SUBSCRIBE_KEY_LEN];
+  SArray* logs;  // SArray<SMqSubActionLogEntry*>
+} SMqSubActionLogObj;
+
+SMqSubActionLogObj* tCloneSMqSubActionLogObj(SMqSubActionLogObj* pLog);
+void                tDeleteSMqSubActionLogObj(SMqSubActionLogObj* pLog);
+int32_t             tEncodeSMqSubActionLogObj(void** buf, const SMqSubActionLogObj* pLog);
+void*               tDecodeSMqSubActionLogObj(const void* buf, SMqSubActionLogObj* pLog);
+
+typedef struct {
+  int64_t  consumerId;
+  char     cgroup[TSDB_CGROUP_LEN];
+  SRWLatch lock;
+  SArray*  vgs;  // SArray<SMqVgEp*>
+} SMqConsumerEpObj;
+
+SMqConsumerEpObj* tCloneSMqConsumerEpObj(const SMqConsumerEpObj* pConsumerEp);
+void              tDeleteSMqConsumerEpObj(SMqConsumerEpObj* pConsumerEp);
+int32_t           tEncodeSMqConsumerEpObj(void** buf, const SMqConsumerEpObj* pConsumerEp);
+void*             tDecodeSMqConsumerEpObj(const void* buf, SMqConsumerEpObj* pConsumerEp);
+
+typedef struct {
+  const SMqSubscribeObj* pOldSub;
+  const SMqTopicObj*     pTopic;
+  const SMqRebSubscribe* pRebInfo;
+} SMqRebInputObj;
+
+typedef struct {
+  int64_t  oldConsumerId;
+  int64_t  newConsumerId;
+  SMqVgEp* pVgEp;
+} SMqRebOutputVg;
+
+#if 0
+typedef struct {
+  int64_t consumerId;
+} SMqRebOutputConsumer;
+#endif
+
+typedef struct {
+  SArray*               rebVgs;            // SArray<SMqRebOutputVg>
+  SArray*               newConsumers;      // SArray<int64_t>
+  SArray*               removedConsumers;  // SArray<int64_t>
+  SArray*               touchedConsumers;  // SArray<int64_t>
+  SMqSubscribeObj*      pSub;
+  SMqSubActionLogEntry* pLogEntry;
+} SMqRebOutputObj;
 
 typedef struct {
   char     name[TSDB_TOPIC_FNAME_LEN];

source/dnode/mnode/impl/inc/mndSubscribe.h

@@ -26,9 +26,11 @@ int32_t mndInitSubscribe(SMnode *pMnode);
 void    mndCleanupSubscribe(SMnode *pMnode);
 
 SMqSubscribeObj *mndAcquireSubscribe(SMnode *pMnode, const char *CGroup, const char *topicName);
-SMqSubscribeObj *mndAcquireSubscribeByKey(SMnode *pMnode, const char* key);
+SMqSubscribeObj *mndAcquireSubscribeByKey(SMnode *pMnode, const char *key);
 void             mndReleaseSubscribe(SMnode *pMnode, SMqSubscribeObj *pSub);
 
+int32_t mndMakeSubscribeKey(char *key, const char *cgroup, const char *topicName);
+
 #ifdef __cplusplus
 }
 #endif

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

@@ -14,6 +14,386 @@
  */
 
 #include "mndDef.h"
+#include "mndConsumer.h"
+
+SMqConsumerObj *tNewSMqConsumerObj(int64_t consumerId, char cgroup[TSDB_CGROUP_LEN]) {
+  SMqConsumerObj *pConsumer = taosMemoryCalloc(1, sizeof(SMqConsumerObj));
+  if (pConsumer == NULL) {
+    return NULL;
+  }
+
+  pConsumer->consumerId = consumerId;
+  memcpy(pConsumer->cgroup, cgroup, TSDB_CGROUP_LEN);
+
+  pConsumer->epoch = 0;
+  pConsumer->status = MQ_CONSUMER_STATUS__MODIFY;
+  pConsumer->hbStatus = 0;
+
+  taosInitRWLatch(&pConsumer->lock);
+
+  pConsumer->currentTopics = taosArrayInit(0, sizeof(void *));
+#if 0
+  pConsumer->waitingRebTopics = NULL;
+#endif
+  pConsumer->rebNewTopics = taosArrayInit(0, sizeof(void *));
+  pConsumer->rebRemovedTopics = taosArrayInit(0, sizeof(void *));
+
+  if (pConsumer->currentTopics == NULL || pConsumer->rebNewTopics == NULL || pConsumer->rebRemovedTopics == NULL) {
+    taosArrayDestroy(pConsumer->currentTopics);
+    taosArrayDestroy(pConsumer->rebNewTopics);
+    taosArrayDestroy(pConsumer->rebRemovedTopics);
+    taosMemoryFree(pConsumer);
+    return NULL;
+  }
+
+  return pConsumer;
+}
+
+void tDeleteSMqConsumerObj(SMqConsumerObj *pConsumer) {
+  if (pConsumer->currentTopics) {
+    taosArrayDestroyP(pConsumer->currentTopics, (FDelete)taosMemoryFree);
+  }
+#if 0
+  if (pConsumer->waitingRebTopics) {
+    taosArrayDestroyP(pConsumer->waitingRebTopics, taosMemoryFree);
+  }
+#endif
+  if (pConsumer->rebNewTopics) {
+    taosArrayDestroyP(pConsumer->rebNewTopics, (FDelete)taosMemoryFree);
+  }
+  if (pConsumer->rebRemovedTopics) {
+    taosArrayDestroyP(pConsumer->rebRemovedTopics, (FDelete)taosMemoryFree);
+  }
+}
+
+int32_t tEncodeSMqConsumerObj(void **buf, const SMqConsumerObj *pConsumer) {
+  int32_t tlen = 0;
+  int32_t sz;
+  tlen += taosEncodeFixedI64(buf, pConsumer->consumerId);
+  tlen += taosEncodeString(buf, pConsumer->cgroup);
+  tlen += taosEncodeFixedI8(buf, pConsumer->updateType);
+  tlen += taosEncodeFixedI32(buf, pConsumer->epoch);
+  tlen += taosEncodeFixedI32(buf, pConsumer->status);
+
+  // current topics
+  if (pConsumer->currentTopics) {
+    sz = taosArrayGetSize(pConsumer->currentTopics);
+    tlen += taosEncodeFixedI32(buf, sz);
+    for (int32_t i = 0; i < sz; i++) {
+      char *topic = taosArrayGetP(pConsumer->currentTopics, i);
+      tlen += taosEncodeString(buf, topic);
+    }
+  } else {
+    tlen += taosEncodeFixedI32(buf, 0);
+  }
+
+#if 0
+  // waiting reb topics
+  if (pConsumer->waitingRebTopics) {
+    sz = taosArrayGetSize(pConsumer->waitingRebTopics);
+    tlen += taosEncodeFixedI32(buf, sz);
+    for (int32_t i = 0; i < sz; i++) {
+      char *topic = taosArrayGetP(pConsumer->waitingRebTopics, i);
+      tlen += taosEncodeString(buf, topic);
+    }
+  } else {
+    tlen += taosEncodeFixedI32(buf, 0);
+  }
+#endif
+
+  // reb new topics
+  if (pConsumer->rebNewTopics) {
+    sz = taosArrayGetSize(pConsumer->rebNewTopics);
+    tlen += taosEncodeFixedI32(buf, sz);
+    for (int32_t i = 0; i < sz; i++) {
+      char *topic = taosArrayGetP(pConsumer->rebNewTopics, i);
+      tlen += taosEncodeString(buf, topic);
+    }
+  } else {
+    tlen += taosEncodeFixedI32(buf, 0);
+  }
+
+  // reb removed topics
+  if (pConsumer->rebRemovedTopics) {
+    sz = taosArrayGetSize(pConsumer->rebRemovedTopics);
+    tlen += taosEncodeFixedI32(buf, sz);
+    for (int32_t i = 0; i < sz; i++) {
+      char *topic = taosArrayGetP(pConsumer->rebRemovedTopics, i);
+      tlen += taosEncodeString(buf, topic);
+    }
+  } else {
+    tlen += taosEncodeFixedI32(buf, 0);
+  }
+
+  return tlen;
+}
+
+void *tDecodeSMqConsumerObj(const void *buf, SMqConsumerObj *pConsumer) {
+  int32_t sz;
+  buf = taosDecodeFixedI64(buf, &pConsumer->consumerId);
+  buf = taosDecodeStringTo(buf, pConsumer->cgroup);
+  buf = taosDecodeFixedI8(buf, &pConsumer->updateType);
+  buf = taosDecodeFixedI32(buf, &pConsumer->epoch);
+  buf = taosDecodeFixedI32(buf, &pConsumer->status);
+
+  // current topics
+  buf = taosDecodeFixedI32(buf, &sz);
+  pConsumer->currentTopics = taosArrayInit(sz, sizeof(void *));
+  for (int32_t i = 0; i < sz; i++) {
+    char *topic;
+    buf = taosDecodeString(buf, &topic);
+    taosArrayPush(pConsumer->currentTopics, &topic);
+  }
+
+#if 0
+  // waiting reb topics
+  buf = taosDecodeFixedI32(buf, &sz);
+  pConsumer->waitingRebTopics = taosArrayInit(sz, sizeof(void *));
+  for (int32_t i = 0; i < sz; i++) {
+    char *topic;
+    buf = taosDecodeString(buf, &topic);
+    taosArrayPush(pConsumer->waitingRebTopics, &topic);
+  }
+#endif
+
+  // reb new topics
+  buf = taosDecodeFixedI32(buf, &sz);
+  pConsumer->rebNewTopics = taosArrayInit(sz, sizeof(void *));
+  for (int32_t i = 0; i < sz; i++) {
+    char *topic;
+    buf = taosDecodeString(buf, &topic);
+    taosArrayPush(pConsumer->rebNewTopics, &topic);
+  }
+
+  // reb removed topics
+  buf = taosDecodeFixedI32(buf, &sz);
+  pConsumer->rebRemovedTopics = taosArrayInit(sz, sizeof(void *));
+  for (int32_t i = 0; i < sz; i++) {
+    char *topic;
+    buf = taosDecodeString(buf, &topic);
+    taosArrayPush(pConsumer->rebRemovedTopics, &topic);
+  }
+
+  return (void *)buf;
+}
+
+SMqVgEp *tCloneSMqVgEp(const SMqVgEp *pVgEp) {
+  SMqVgEp *pVgEpNew = taosMemoryMalloc(sizeof(SMqVgEp));
+  if (pVgEpNew == NULL) return NULL;
+  pVgEpNew->vgId = pVgEp->vgId;
+  pVgEpNew->qmsg = strdup(pVgEp->qmsg);
+  /*memcpy(pVgEpNew->topic, pVgEp->topic, TSDB_TOPIC_FNAME_LEN);*/
+  pVgEpNew->epSet = pVgEp->epSet;
+  return pVgEpNew;
+}
+
+void tDeleteSMqVgEp(SMqVgEp *pVgEp) { taosMemoryFree(pVgEp->qmsg); }
+
+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->topic);*/
+  tlen += taosEncodeSEpSet(buf, &pVgEp->epSet);
+  return tlen;
+}
+
+void *tDecodeSMqVgEp(const void *buf, SMqVgEp *pVgEp) {
+  buf = taosDecodeFixedI32(buf, &pVgEp->vgId);
+  buf = taosDecodeString(buf, &pVgEp->qmsg);
+  /*buf = taosDecodeStringTo(buf, pVgEp->topic);*/
+  buf = taosDecodeSEpSet(buf, &pVgEp->epSet);
+  return (void *)buf;
+}
+
+SMqConsumerEpObj *tCloneSMqConsumerEpObj(const SMqConsumerEpObj *pConsumerEp) {
+  SMqConsumerEpObj *pConsumerEpNew = taosMemoryMalloc(sizeof(SMqConsumerEpObj));
+  if (pConsumerEpNew == NULL) return NULL;
+  pConsumerEpNew->consumerId = pConsumerEp->consumerId;
+  memcpy(pConsumerEpNew->cgroup, pConsumerEp->cgroup, TSDB_CGROUP_LEN);
+  taosInitRWLatch(&pConsumerEpNew->lock);
+  pConsumerEpNew->vgs = taosArrayDeepCopy(pConsumerEpNew->vgs, (FCopy)tCloneSMqVgEp);
+  return pConsumerEpNew;
+}
+
+void tDeleteSMqConsumerEpObj(SMqConsumerEpObj *pConsumerEp) {
+  taosArrayDestroyEx(pConsumerEp->vgs, (FDelete)tDeleteSMqVgEp);
+}
+
+int32_t tEncodeSMqConsumerEpObj(void **buf, const SMqConsumerEpObj *pConsumerEp) {
+  int32_t tlen = 0;
+  tlen += taosEncodeFixedI64(buf, pConsumerEp->consumerId);
+  tlen += taosEncodeString(buf, pConsumerEp->cgroup);
+  tlen += taosEncodeArray(buf, pConsumerEp->vgs, (FEncode)tEncodeSMqVgEp);
+  return tlen;
+}
+
+void *tDecodeSMqConsumerEpObj(const void *buf, SMqConsumerEpObj *pConsumerEp) {
+  buf = taosDecodeFixedI64(buf, &pConsumerEp->consumerId);
+  buf = taosDecodeStringTo(buf, pConsumerEp->cgroup);
+  buf = taosDecodeArray(buf, &pConsumerEp->vgs, (FDecode)tDecodeSMqVgEp, sizeof(SMqSubVgEp));
+  return (void *)buf;
+}
+
+SMqConsumerEpInSub *tCloneSMqConsumerEpInSub(const SMqConsumerEpInSub *pEpInSub) {
+  SMqConsumerEpInSub *pEpInSubNew = taosMemoryMalloc(sizeof(SMqConsumerEpInSub));
+  if (pEpInSubNew == NULL) return NULL;
+  pEpInSubNew->consumerId = pEpInSub->consumerId;
+  pEpInSubNew->vgs = taosArrayDeepCopy(pEpInSub->vgs, (FCopy)tCloneSMqVgEp);
+  return pEpInSubNew;
+}
+
+void tDeleteSMqConsumerEpInSub(SMqConsumerEpInSub *pEpInSub) {
+  taosArrayDestroyEx(pEpInSub->vgs, (FDelete)tDeleteSMqVgEp);
+}
+
+int32_t tEncodeSMqConsumerEpInSub(void **buf, const SMqConsumerEpInSub *pEpInSub) {
+  int32_t tlen = 0;
+  tlen += taosEncodeFixedI64(buf, pEpInSub->consumerId);
+  tlen += taosEncodeArray(buf, pEpInSub->vgs, (FEncode)tEncodeSMqVgEp);
+  return tlen;
+}
+
+void *tDecodeSMqConsumerEpInSub(const void *buf, SMqConsumerEpInSub *pEpInSub) {
+  buf = taosDecodeFixedI64(buf, &pEpInSub->consumerId);
+  buf = taosDecodeArray(buf, &pEpInSub->vgs, (FDecode)tDecodeSMqVgEp, sizeof(SMqSubVgEp));
+  return (void *)buf;
+}
+
+SMqSubscribeObj *tNewSubscribeObj(const char key[TSDB_SUBSCRIBE_KEY_LEN]) {
+  SMqSubscribeObj *pSubNew = taosMemoryMalloc(sizeof(SMqSubscribeObj));
+  if (pSubNew == NULL) return NULL;
+  memcpy(pSubNew->key, key, TSDB_SUBSCRIBE_KEY_LEN);
+  taosInitRWLatch(&pSubNew->lock);
+  pSubNew->vgNum = -1;
+  pSubNew->consumerHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
+  // TODO set free fp
+  SMqConsumerEpInSub *pEpInSub = taosMemoryMalloc(sizeof(SMqConsumerEpInSub));
+  pEpInSub->vgs = taosArrayInit(0, sizeof(SMqVgEp));
+  int64_t unexistKey = -1;
+  taosHashPut(pSubNew->consumerHash, &unexistKey, sizeof(int64_t), pEpInSub, sizeof(SMqConsumerEpInSub));
+  return pSubNew;
+}
+
+SMqSubscribeObj *tCloneSubscribeObj(const SMqSubscribeObj *pSub) {
+  SMqSubscribeObj *pSubNew = taosMemoryMalloc(sizeof(SMqSubscribeObj));
+  if (pSubNew == NULL) return NULL;
+  memcpy(pSubNew->key, pSub->key, TSDB_SUBSCRIBE_KEY_LEN);
+  taosInitRWLatch(&pSubNew->lock);
+  pSubNew->vgNum = pSub->vgNum;
+  /*pSubNew->consumerEps = taosArrayDeepCopy(pSub->consumerEps, (FCopy)tCloneSMqConsumerEpInSub);*/
+  pSubNew->consumerHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
+  /*taosHashSetFreeFp(pSubNew->consumerHash, taosArrayDestroy);*/
+  void               *pIter = NULL;
+  SMqConsumerEpInSub *pEpInSub = NULL;
+  while (1) {
+    pIter = taosHashIterate(pSubNew->consumerHash, pIter);
+    if (pIter == NULL) break;
+    pEpInSub = (SMqConsumerEpInSub *)pIter;
+    SMqConsumerEpInSub newEp = {
+        .consumerId = pEpInSub->consumerId,
+        .vgs = taosArrayDeepCopy(pEpInSub->vgs, (FCopy)tCloneSMqVgEp),
+    };
+    taosHashPut(pSubNew->consumerHash, &newEp.consumerId, sizeof(int64_t), &newEp, sizeof(SMqConsumerEpInSub));
+  }
+  return pSubNew;
+}
+
+void tDeleteSubscribeObj(SMqSubscribeObj *pSub) {
+  /*taosArrayDestroyEx(pSub->consumerEps, (FDelete)tDeleteSMqConsumerEpInSub);*/
+  taosHashCleanup(pSub->consumerHash);
+}
+
+int32_t tEncodeSubscribeObj(void **buf, const SMqSubscribeObj *pSub) {
+  int32_t tlen = 0;
+  tlen += taosEncodeString(buf, pSub->key);
+  tlen += taosEncodeFixedI32(buf, pSub->vgNum);
+
+  void   *pIter = NULL;
+  int32_t sz = taosHashGetSize(pSub->consumerHash);
+  tlen += taosEncodeFixedI32(buf, sz);
+
+  int32_t cnt = 0;
+  while (1) {
+    pIter = taosHashIterate(pSub->consumerHash, pIter);
+    if (pIter == NULL) break;
+    SMqConsumerEpInSub *pEpInSub = (SMqConsumerEpInSub *)pIter;
+    tlen += tEncodeSMqConsumerEpInSub(buf, pEpInSub);
+    cnt++;
+  }
+  ASSERT(cnt == sz);
+  /*tlen += taosEncodeArray(buf, pSub->consumerEps, (FEncode)tEncodeSMqConsumerEpInSub);*/
+  return tlen;
+}
+
+void *tDecodeSubscribeObj(const void *buf, SMqSubscribeObj *pSub) {
+  //
+  buf = taosDecodeStringTo(buf, pSub->key);
+  buf = taosDecodeFixedI32(buf, &pSub->vgNum);
+
+  int32_t sz;
+  buf = taosDecodeFixedI32(buf, &sz);
+
+  pSub->consumerHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);
+  for (int32_t i = 0; i < sz; i++) {
+    /*SMqConsumerEpInSub* pEpInSub = taosMemoryMalloc(sizeof(SMqConsumerEpInSub));*/
+    SMqConsumerEpInSub epInSub = {0};
+    buf = tDecodeSMqConsumerEpInSub(buf, &epInSub);
+    taosHashPut(pSub->consumerHash, &epInSub.consumerId, sizeof(int64_t), &epInSub, sizeof(SMqConsumerEpInSub));
+  }
+
+  /*buf = taosDecodeArray(buf, &pSub->consumerEps, (FDecode)tDecodeSMqConsumerEpInSub, sizeof(SMqConsumerEpInSub));*/
+  return (void *)buf;
+}
+
+SMqSubActionLogEntry *tCloneSMqSubActionLogEntry(SMqSubActionLogEntry *pEntry) {
+  SMqSubActionLogEntry *pEntryNew = taosMemoryMalloc(sizeof(SMqSubActionLogEntry));
+  if (pEntryNew == NULL) return NULL;
+  pEntryNew->epoch = pEntry->epoch;
+  pEntryNew->consumers = taosArrayDeepCopy(pEntry->consumers, (FCopy)tCloneSMqConsumerEpInSub);
+  return pEntryNew;
+}
+
+void tDeleteSMqSubActionLogEntry(SMqSubActionLogEntry *pEntry) {
+  taosArrayDestroyEx(pEntry->consumers, (FDelete)tDeleteSMqConsumerEpInSub);
+}
+
+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;
+}
+
+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 = taosArrayDeepCopy(pLog->logs, (FCopy)tCloneSMqConsumerEpInSub);
+  return pLogNew;
+}
+
+void tDeleteSMqSubActionLogObj(SMqSubActionLogObj *pLog) {
+  taosArrayDestroyEx(pLog->logs, (FDelete)tDeleteSMqConsumerEpInSub);
+}
+
+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;
+}
 
 int32_t tEncodeSStreamObj(SCoder *pEncoder, const SStreamObj *pObj) {
   int32_t sz = 0;

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

@@ -453,6 +453,9 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
   }
   SSubplan* plan = nodesListGetNode(inner->pNodeList, 0);
 
+  int64_t             unexistKey = -1;
+  SMqConsumerEpInSub* pEpInSub = taosHashGet(pSub->consumerHash, &unexistKey, sizeof(int64_t));
+
   void* pIter = NULL;
   while (1) {
     pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void**)&pVgroup);
@@ -466,24 +469,33 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
     plan->execNode.nodeId = pVgroup->vgId;
     plan->execNode.epSet = mndGetVgroupEpset(pMnode, pVgroup);
 
+    SMqVgEp* pVgEp = taosMemoryMalloc(sizeof(SMqVgEp));
+    pVgEp->epSet = plan->execNode.epSet;
+    pVgEp->vgId = plan->execNode.nodeId;
+
+#if 0
     SMqConsumerEp consumerEp = {0};
     consumerEp.status = 0;
     consumerEp.consumerId = -1;
     consumerEp.epSet = plan->execNode.epSet;
     consumerEp.vgId = plan->execNode.nodeId;
-    
-    mDebug("init subscribption %s, assign vg: %d", pSub->key, consumerEp.vgId);
+#endif
+
+    mDebug("init subscribption %s, assign vg: %d", pSub->key, pVgEp->vgId);
 
     int32_t msgLen;
-    if (qSubPlanToString(plan, &consumerEp.qmsg, &msgLen) < 0) {
+    if (qSubPlanToString(plan, &pVgEp->qmsg, &msgLen) < 0) {
       sdbRelease(pSdb, pVgroup);
       qDestroyQueryPlan(pPlan);
       terrno = TSDB_CODE_QRY_INVALID_INPUT;
       return -1;
     }
-    taosArrayPush(pSub->unassignedVg, &consumerEp);
+    taosArrayPush(pEpInSub->vgs, &pVgEp);
+
+    /*taosArrayPush(pSub->unassignedVg, &consumerEp);*/
   }
 
+  taosHashRelease(pSub->consumerHash, pEpInSub);
   qDestroyQueryPlan(pPlan);
 
   return 0;

source/util/src/tarray.c

@@ -15,6 +15,7 @@
 
 #define _DEFAULT_SOURCE
 #include "tarray.h"
+#include "tcoding.h"
 
 SArray* taosArrayInit(size_t size, size_t elemSize) {
   assert(elemSize > 0);
@@ -312,7 +313,14 @@ void* taosArrayDestroy(SArray* pArray) {
   return NULL;
 }
 
-void taosArrayDestroyEx(SArray* pArray, void (*fp)(void*)) {
+void taosArrayDestroyP(SArray* pArray, FDelete fp) {
+  for (int32_t i = 0; i < pArray->size; i++) {
+    fp(*(void**)TARRAY_GET_ELEM(pArray, i));
+  }
+  taosArrayDestroy(pArray);
+}
+
+void taosArrayDestroyEx(SArray* pArray, FDelete fp) {
   if (pArray == NULL) {
     return;
   }
@@ -421,6 +429,37 @@ static void taosArrayInsertSort(SArray* pArray, __ext_compar_fn_t fn, const void
   return;
 }
 
+SArray* taosArrayDeepCopy(const SArray* pSrc, FCopy deepCopy) {
+  SArray* pArray = taosArrayInit(pSrc->size, pSrc->elemSize);
+  for (int32_t i = 0; i < pSrc->size; i++) {
+    void* clone = deepCopy(taosArrayGetP(pSrc, i));
+    taosArrayPush(pArray, &clone);
+  }
+  return pArray;
+}
+
+int32_t taosEncodeArray(void** buf, const SArray* pArray, FEncode encode) {
+  int32_t tlen = 0;
+  int32_t sz = pArray->size;
+  tlen += taosEncodeFixedI32(buf, sz);
+  for (int32_t i = 0; i < sz; i++) {
+    void* data = taosArrayGetP(pArray, i);
+    tlen += encode(buf, data);
+  }
+  return tlen;
+}
+
+void* taosDecodeArray(const void* buf, SArray** pArray, FDecode decode, int32_t dataSz) {
+  int32_t sz;
+  buf = taosDecodeFixedI32(buf, &sz);
+  *pArray = taosArrayInit(sz, sizeof(void*));
+  for (int32_t i = 0; i < sz; i++) {
+    void* data = taosMemoryCalloc(1, dataSz);
+    buf = decode(buf, data);
+  }
+  return (void*)buf;
+}
+
 // order array<type *>
 void taosArraySortPWithExt(SArray* pArray, __ext_compar_fn_t fn, const void* param) {
   taosArrayGetSize(pArray) > 8 ? taosArrayQuickSort(pArray, fn, param) : taosArrayInsertSort(pArray, fn, param);