From da550029d893d53649d0b25b08b0ebf7dfa85d5b Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Mon, 6 Mar 2023 11:52:52 +0800
Subject: [PATCH] fix(tmq): fix memory error and adjust some logs.

---
 source/client/src/clientTmq.c              |  2 +-
 source/dnode/mnode/impl/src/mndConsumer.c  |  5 ++---
 source/dnode/mnode/impl/src/mndSubscribe.c |  8 +++++---
 source/dnode/mnode/impl/src/mndTrans.c     |  1 +
 tests/script/tsim/tmq/consume.sh           | 10 +++++-----
 5 files changed, 14 insertions(+), 12 deletions(-)

diff --git a/source/client/src/clientTmq.c b/source/client/src/clientTmq.c
index f6a2c5fdc1..9f24deff94 100644
--- a/source/client/src/clientTmq.c
+++ b/source/client/src/clientTmq.c
@@ -1291,7 +1291,7 @@ int32_t tmqPollCb(void* param, SDataBuf* pMsg, int32_t code) {
 
     tscDebug("consumer:0x%" PRIx64 " recv poll rsp, vgId:%d, req offset:%" PRId64 ", rsp offset:%" PRId64 " type %d, reqId:0x%"PRIx64,
              tmq->consumerId, pVg->vgId, pRspWrapper->dataRsp.reqOffset.version, pRspWrapper->dataRsp.rspOffset.version,
-             rspType, pParam->requestId);
+             rspType, requestId);
   } else if (rspType == TMQ_MSG_TYPE__POLL_META_RSP) {
     SDecoder decoder;
     tDecoderInit(&decoder, POINTER_SHIFT(pMsg->pData, sizeof(SMqRspHead)), pMsg->len - sizeof(SMqRspHead));
diff --git a/source/dnode/mnode/impl/src/mndConsumer.c b/source/dnode/mnode/impl/src/mndConsumer.c
index 24974a1973..f1ef83aca5 100644
--- a/source/dnode/mnode/impl/src/mndConsumer.c
+++ b/source/dnode/mnode/impl/src/mndConsumer.c
@@ -35,9 +35,9 @@ static const char *mndConsumerStatusName(int status);
 
 static int32_t mndConsumerActionInsert(SSdb *pSdb, SMqConsumerObj *pConsumer);
 static int32_t mndConsumerActionDelete(SSdb *pSdb, SMqConsumerObj *pConsumer);
-static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pConsumer, SMqConsumerObj *pNewConsumer);
+static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pOldConsumer, SMqConsumerObj *pNewConsumer);
 static int32_t mndProcessConsumerMetaMsg(SRpcMsg *pMsg);
-static int32_t mndRetrieveConsumer(SRpcMsg *pMsg, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
+static int32_t mndRetrieveConsumer(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
 static void    mndCancelGetNextConsumer(SMnode *pMnode, void *pIter);
 
 static int32_t mndProcessSubscribeReq(SRpcMsg *pMsg);
@@ -712,7 +712,6 @@ int32_t mndProcessSubscribeReq(SRpcMsg *pMsg) {
 
     // no topics need to be rebalanced
     if (taosArrayGetSize(pConsumerNew->rebNewTopics) == 0 && taosArrayGetSize(pConsumerNew->rebRemovedTopics) == 0) {
-//      mInfo();
       goto _over;
     }
 
diff --git a/source/dnode/mnode/impl/src/mndSubscribe.c b/source/dnode/mnode/impl/src/mndSubscribe.c
index 86f6976398..21539a6313 100644
--- a/source/dnode/mnode/impl/src/mndSubscribe.c
+++ b/source/dnode/mnode/impl/src/mndSubscribe.c
@@ -444,7 +444,9 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
 
 static int32_t mndPersistRebResult(SMnode *pMnode, SRpcMsg *pMsg, const SMqRebOutputObj *pOutput) {
   STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_DB_INSIDE, pMsg, "tmq-reb");
-  if (pTrans == NULL) return -1;
+  if (pTrans == NULL) {
+    return -1;
+  }
 
   mndTransSetDbName(pTrans, pOutput->pSub->dbName, NULL);
   if (mndTrancCheckConflict(pMnode, pTrans) != 0) {
@@ -616,9 +618,9 @@ static int32_t mndProcessRebalanceReq(SRpcMsg *pMsg) {
 
     // if add more consumer to balanced subscribe,
     // possibly no vg is changed
-
+    // when each topic is re-balanced, issue an trans to save the results in sdb.
     if (mndPersistRebResult(pMnode, pMsg, &rebOutput) < 0) {
-      mError("mq re-balance persist re-balance output error, possibly vnode splitted or dropped");
+      mError("mq re-balance persist output error, possibly vnode splitted or dropped");
     }
 
     taosArrayDestroy(pRebInfo->lostConsumers);
diff --git a/source/dnode/mnode/impl/src/mndTrans.c b/source/dnode/mnode/impl/src/mndTrans.c
index ff8b4dd9ce..39b4252618 100644
--- a/source/dnode/mnode/impl/src/mndTrans.c
+++ b/source/dnode/mnode/impl/src/mndTrans.c
@@ -875,6 +875,7 @@ int32_t mndTrancCheckConflict(SMnode *pMnode, STrans *pTrans) {
     }
   }
 
+
   if (mndCheckTransConflict(pMnode, pTrans)) {
     terrno = TSDB_CODE_MND_TRANS_CONFLICT;
     mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
diff --git a/tests/script/tsim/tmq/consume.sh b/tests/script/tsim/tmq/consume.sh
index 001ce6ae49..f0c9197a59 100755
--- a/tests/script/tsim/tmq/consume.sh
+++ b/tests/script/tsim/tmq/consume.sh
@@ -79,15 +79,15 @@ CFG_DIR=$PRG_DIR/cfg
 LOG_DIR=$PRG_DIR/log
 
 echo "------------------------------------------------------------------------"
+echo "TOP_DIR: $TOP_DIR"
 echo "BUILD_DIR: $BUILD_DIR"
 echo "SIM_DIR  : $SIM_DIR"
 echo "CFG_DIR  : $CFG_DIR"
 
-
-echo "PROGRAM: $PROGRAM
-echo "CFG_DIR: $CFG_DIR 
-echo "POLL_DELAY: $POLL_DELAY 
-echo "DB_NAME: $DB_NAME
+echo "PROGRAM: $PROGRAM"
+echo "CFG_DIR: $CFG_DIR"
+echo "POLL_DELAY: $POLL_DELAY"
+echo "DB_NAME: $DB_NAME"
 
 echo "------------------------------------------------------------------------"
 if [ "$EXEC_OPTON" = "start" ]; then 
-- 
GitLab