From d158ca74d841b0cf373725fed96010b8a7872f4a Mon Sep 17 00:00:00 2001
From: wangmm0220 <wangmm0220@gmail.com>
Date: Fri, 29 Dec 2023 17:22:27 +0800
Subject: [PATCH] fix:[TS-4391] rebalance cnt always 1 if msg lost

---
 source/dnode/mnode/impl/src/mndConsumer.c  | 15 ---------------
 source/dnode/mnode/impl/src/mndSubscribe.c | 14 ++++++++------
 2 files changed, 8 insertions(+), 21 deletions(-)

diff --git a/source/dnode/mnode/impl/src/mndConsumer.c b/source/dnode/mnode/impl/src/mndConsumer.c
index cf8b9e019a..9d9473d883 100644
--- a/source/dnode/mnode/impl/src/mndConsumer.c
+++ b/source/dnode/mnode/impl/src/mndConsumer.c
@@ -220,21 +220,6 @@ FAIL:
   return -1;
 }
 
-static SMqRebInfo *mndGetOrCreateRebSub(SHashObj *pHash, const char *key) {
-  SMqRebInfo *pRebInfo = taosHashGet(pHash, key, strlen(key) + 1);
-  if (pRebInfo == NULL) {
-    pRebInfo = tNewSMqRebSubscribe(key);
-    if (pRebInfo == NULL) {
-      terrno = TSDB_CODE_OUT_OF_MEMORY;
-      return NULL;
-    }
-    taosHashPut(pHash, key, strlen(key) + 1, pRebInfo, sizeof(SMqRebInfo));
-    taosMemoryFree(pRebInfo);
-    pRebInfo = taosHashGet(pHash, key, strlen(key) + 1);
-  }
-  return pRebInfo;
-}
-
 static int32_t mndProcessMqHbReq(SRpcMsg *pMsg) {
   int32_t code = 0;
   SMnode  *pMnode = pMsg->info.node;
diff --git a/source/dnode/mnode/impl/src/mndSubscribe.c b/source/dnode/mnode/impl/src/mndSubscribe.c
index b7958c1484..16ae7f2548 100644
--- a/source/dnode/mnode/impl/src/mndSubscribe.c
+++ b/source/dnode/mnode/impl/src/mndSubscribe.c
@@ -209,16 +209,18 @@ static int32_t mndSplitSubscribeKey(const char *key, char *topic, char *cgroup,
 }
 
 static SMqRebInfo *mndGetOrCreateRebSub(SHashObj *pHash, const char *key) {
-  SMqRebInfo *pRebSub = taosHashGet(pHash, key, strlen(key) + 1);
-  if (pRebSub == NULL) {
-    pRebSub = tNewSMqRebSubscribe(key);
-    if (pRebSub == NULL) {
+  SMqRebInfo *pRebInfo = taosHashGet(pHash, key, strlen(key) + 1);
+  if (pRebInfo == NULL) {
+    pRebInfo = tNewSMqRebSubscribe(key);
+    if (pRebInfo == NULL) {
       terrno = TSDB_CODE_OUT_OF_MEMORY;
       return NULL;
     }
-    taosHashPut(pHash, key, strlen(key) + 1, pRebSub, sizeof(SMqRebInfo));
+    taosHashPut(pHash, key, strlen(key) + 1, pRebInfo, sizeof(SMqRebInfo));
+    taosMemoryFree(pRebInfo);
+    pRebInfo = taosHashGet(pHash, key, strlen(key) + 1);
   }
-  return pRebSub;
+  return pRebInfo;
 }
 
 static void doRemoveLostConsumers(SMqRebOutputObj *pOutput, SHashObj *pHash, const SMqRebInputObj *pInput) {