@@ -85,21 +85,17 @@ void freeSMqConsumerEp(void* data) {
8585 pConsumerEp -> offsetRows = NULL ;
8686}
8787
88- int32_t tEncodeSMqVgEp (void * * buf , const SMqVgEp * pVgEp ) {
89- int32_t tlen = 0 ;
90- tlen += taosEncodeFixedI32 (buf , pVgEp -> vgId );
91- tlen += taosEncodeSEpSet (buf , & pVgEp -> epSet );
92- return tlen ;
93- }
94-
95- void * tDecodeSMqVgEp (const void * buf , SMqVgEp * pVgEp , int8_t sver ) {
96- buf = taosDecodeFixedI32 (buf , & pVgEp -> vgId );
88+ static void * tDecodeSMqVgEp (const void * buf , int32_t * vgId , int8_t sver ) {
89+ buf = taosDecodeFixedI32 (buf , vgId );
9790 if (sver == 1 ) {
9891 uint64_t size = 0 ;
9992 buf = taosDecodeVariantU64 (buf , & size );
10093 buf = POINTER_SHIFT (buf , size );
10194 }
102- buf = taosDecodeSEpSet (buf , & pVgEp -> epSet );
95+ if (sver <= 3 ) {
96+ SEpSet epSet = {0 };
97+ buf = taosDecodeSEpSet (buf , & epSet );
98+ }
10399 return (void * )buf ;
104100}
105101
@@ -204,7 +200,6 @@ int32_t tEncodeSMqConsumerObj(void **buf, const SMqConsumerObj *pConsumer) {
204200 tlen += taosEncodeFixedI32 (buf , pConsumer -> status );
205201
206202 tlen += taosEncodeFixedI32 (buf , pConsumer -> pid );
207- tlen += taosEncodeSEpSet (buf , & pConsumer -> ep );
208203 tlen += taosEncodeFixedI64 (buf , pConsumer -> createTime );
209204 tlen += taosEncodeFixedI64 (buf , pConsumer -> subscribeTime );
210205 tlen += taosEncodeFixedI64 (buf , pConsumer -> rebalanceTime );
@@ -278,7 +273,10 @@ void *tDecodeSMqConsumerObj(const void *buf, SMqConsumerObj *pConsumer, int8_t s
278273 buf = taosDecodeFixedI32 (buf , & pConsumer -> status );
279274
280275 buf = taosDecodeFixedI32 (buf , & pConsumer -> pid );
281- buf = taosDecodeSEpSet (buf , & pConsumer -> ep );
276+ if (sver <= 3 ) {
277+ SEpSet ep = {0 };
278+ buf = taosDecodeSEpSet (buf , & ep );
279+ }
282280 buf = taosDecodeFixedI64 (buf , & pConsumer -> createTime );
283281 buf = taosDecodeFixedI64 (buf , & pConsumer -> subscribeTime );
284282 buf = taosDecodeFixedI64 (buf , & pConsumer -> rebalanceTime );
@@ -378,8 +376,9 @@ int32_t tEncodeSMqConsumerEp(void **buf, const SMqConsumerEp *pConsumerEp) {
378376 int32_t sz = taosArrayGetSize (pConsumerEp -> vgs );
379377 tlen += taosEncodeFixedI32 (buf , sz );
380378 for (int32_t i = 0 ; i < sz ; i ++ ) {
381- void * data = taosArrayGet (pConsumerEp -> vgs , i );
382- tlen += tEncodeSMqVgEp (buf , data );
379+ int32_t * vgId = (int32_t * )taosArrayGet (pConsumerEp -> vgs , i );
380+ if (vgId == NULL ) continue ;
381+ tlen += taosEncodeFixedI32 (buf , * vgId );
383382 }
384383
385384 return tlen + tEncodeOffRows (buf , pConsumerEp -> offsetRows );
@@ -416,14 +415,20 @@ void *tDecodeSMqConsumerEp(const void *buf, SMqConsumerEp *pConsumerEp, int8_t s
416415 buf = taosDecodeFixedI64 (buf , & pConsumerEp -> consumerId );
417416 int32_t sz = 0 ;
418417 buf = taosDecodeFixedI32 (buf , & sz );
419- pConsumerEp -> vgs = taosArrayInit (sz , sizeof (SMqVgEp ));
418+ pConsumerEp -> vgs = taosArrayInit (sz , sizeof (int32_t ));
420419 if (pConsumerEp -> vgs == NULL ) {
421420 return NULL ;
422421 }
423422 for (int32_t i = 0 ; i < sz ; i ++ ) {
424- SMqVgEp * vgEp = taosArrayReserve (pConsumerEp -> vgs , 1 );
425- if (vgEp != NULL )
426- buf = tDecodeSMqVgEp (buf , vgEp , sver );
423+ int32_t * vgId = taosArrayReserve (pConsumerEp -> vgs , 1 );
424+ if (vgId == NULL ) {
425+ return NULL ;
426+ }
427+ if (sver <= 3 ) {
428+ buf = tDecodeSMqVgEp (buf , vgId , sver );
429+ } else {
430+ buf = taosDecodeFixedI32 (buf , vgId );
431+ }
427432 }
428433 if (sver > 1 ) {
429434 buf = tDecodeOffRows (buf , & pConsumerEp -> offsetRows , sver );
@@ -446,7 +451,7 @@ int32_t tNewSubscribeObj(const char *key, SMqSubscribeObj **ppSub) {
446451 MND_TMQ_NULL_CHECK (pSubObj -> consumerHash );
447452 taosHashSetFreeFp (pSubObj -> consumerHash , freeSMqConsumerEp );
448453
449- pSubObj -> unassignedVgs = taosArrayInit (0 , sizeof (SMqVgEp ));
454+ pSubObj -> unassignedVgs = taosArrayInit (0 , sizeof (int32_t ));
450455 MND_TMQ_NULL_CHECK (pSubObj -> unassignedVgs );
451456 * ppSub = pSubObj ;
452457 pSubObj = NULL ;
@@ -539,8 +544,9 @@ int32_t tEncodeSubscribeObj(void **buf, const SMqSubscribeObj *pSub) {
539544 int32_t len = taosArrayGetSize (pSub -> unassignedVgs );
540545 tlen += taosEncodeFixedI32 (buf , len );
541546 for (int32_t i = 0 ; i < len ; i ++ ) {
542- void * data = taosArrayGet (pSub -> unassignedVgs , i );
543- tlen += tEncodeSMqVgEp (buf , data );
547+ int32_t * vgId = (int32_t * )taosArrayGet (pSub -> unassignedVgs , i );
548+ if (vgId == NULL ) continue ;
549+ tlen += taosEncodeFixedI32 (buf , * vgId );
544550 }
545551 tlen += taosEncodeString (buf , pSub -> dbName );
546552
@@ -576,14 +582,20 @@ void *tDecodeSubscribeObj(const void *buf, SMqSubscribeObj *pSub, int8_t sver) {
576582
577583 int32_t len = 0 ;
578584 buf = taosDecodeFixedI32 (buf , & len );
579- pSub -> unassignedVgs = taosArrayInit (len , sizeof (SMqVgEp ));
585+ pSub -> unassignedVgs = taosArrayInit (len , sizeof (int32_t ));
580586 if (pSub -> unassignedVgs == NULL ) {
581587 return NULL ;
582588 }
583589 for (int32_t i = 0 ; i < len ; i ++ ) {
584- SMqVgEp * vgEp = taosArrayReserve (pSub -> unassignedVgs , 1 );
585- if (vgEp != NULL )
586- buf = tDecodeSMqVgEp (buf , vgEp , sver );
590+ int32_t * vgId = taosArrayReserve (pSub -> unassignedVgs , 1 );
591+ if (vgId == NULL ) {
592+ return NULL ;
593+ }
594+ if (sver <= 3 ) {
595+ buf = tDecodeSMqVgEp (buf , vgId , sver );
596+ } else {
597+ buf = taosDecodeFixedI32 (buf , vgId );
598+ }
587599 }
588600
589601 buf = taosDecodeStringTo (buf , pSub -> dbName );
0 commit comments