Merge pull request #9799 from taosdata/feature/tq

add mq definition

include/client/taos.h

@@ -193,8 +193,7 @@ DLL_EXPORT void taos_close_stream(TAOS_STREAM *tstr);
 DLL_EXPORT int taos_load_table_info(TAOS *taos, const char* tableNameList);
 DLL_EXPORT TAOS_RES* taos_schemaless_insert(TAOS* taos, char* lines[], int numLines, int protocol, int precision);
 
-
-DLL_EXPORT TAOS_RES *tmq_create_topic(TAOS* taos, const char* name, const char* sql, int sqlLen);
+DLL_EXPORT TAOS_RES* tmq_create_topic(TAOS* taos, const char* name, const char* sql, int sqlLen);
 
 #ifdef __cplusplus
 }

include/common/tmsg.h

@@ -68,6 +68,14 @@ typedef uint16_t tmsg_t;
 #define TSDB_IE_TYPE_DNODE_EXT 6
 #define TSDB_IE_TYPE_DNODE_STATE 7
 
+typedef enum {
+  HEARTBEAT_TYPE_MQ    = 0,
+  HEARTBEAT_TYPE_QUERY = 1,
+  // types can be added here
+  //
+  HEARTBEAT_TYPE_MAX
+} EHbType;
+
 typedef enum _mgmt_table {
   TSDB_MGMT_TABLE_START,
   TSDB_MGMT_TABLE_ACCT,
@@ -147,7 +155,7 @@ typedef struct {
 
 typedef struct {
   SClientHbKey connKey;
-  SHashObj*    info;  // hash<Slv.key, Sklv>
+  SHashObj*    info;  // hash<Skv.key, Skv>
 } SClientHbReq;
 
 typedef struct {
@@ -173,7 +181,10 @@ static FORCE_INLINE uint32_t hbKeyHashFunc(const char* key, uint32_t keyLen) {
 }
 
 int   tSerializeSClientHbReq(void** buf, const SClientHbReq* pReq);
-void* tDeserializeClientHbReq(void* buf, SClientHbReq* pReq);
+void* tDeserializeSClientHbReq(void* buf, SClientHbReq* pReq);
+
+int   tSerializeSClientHbRsp(void** buf, const SClientHbRsp* pRsp);
+void* tDeserializeSClientHbRsp(void* buf, SClientHbRsp* pRsp);
 
 static FORCE_INLINE void  tFreeClientHbReq(void *pReq) {
   SClientHbReq* req = (SClientHbReq*)pReq;
@@ -182,14 +193,17 @@ static FORCE_INLINE void  tFreeClientHbReq(void *pReq) {
 }
 
 int   tSerializeSClientHbBatchReq(void** buf, const SClientHbBatchReq* pReq);
-void* tDeserializeClientHbBatchReq(void* buf, SClientHbBatchReq* pReq);
+void* tDeserializeSClientHbBatchReq(void* buf, SClientHbBatchReq* pReq);
 
 static FORCE_INLINE void tFreeClientHbBatchReq(void* pReq) {
   SClientHbBatchReq *req = (SClientHbBatchReq*)pReq;
-  taosArrayDestroyEx(req->reqs, tFreeClientHbReq);
+  //taosArrayDestroyEx(req->reqs, tFreeClientHbReq);
   free(pReq);
 }
 
+int   tSerializeSClientHbBatchRsp(void** buf, const SClientHbBatchRsp* pBatchRsp);
+void* tDeserializeSClientHbBatchRsp(void* buf, SClientHbBatchRsp* pBatchRsp);
+
 static FORCE_INLINE int taosEncodeSKv(void** buf, const SKv* pKv) {
   int tlen = 0;
   tlen += taosEncodeFixedI32(buf, pKv->keyLen);
@@ -220,6 +234,7 @@ static FORCE_INLINE void* taosDecodeSClientHbKey(void* buf, SClientHbKey* pKey)
   return buf;
 }
 
+
 typedef struct {
   int32_t vgId;
   char*   dbName;
@@ -359,6 +374,31 @@ static FORCE_INLINE void* taosDecodeSEpSet(void* buf, SEpSet* pEp) {
   return buf;
 }
 
+typedef struct SMqHbRsp {
+  int8_t status;    //idle or not
+  int8_t vnodeChanged;
+  int8_t epChanged; // should use new epset
+  int8_t reserved;
+  SEpSet epSet;
+} SMqHbRsp;
+
+static FORCE_INLINE int taosEncodeSMqHbRsp(void** buf, const SMqHbRsp* pRsp) {
+  int tlen = 0;
+  tlen += taosEncodeFixedI8(buf, pRsp->status);
+  tlen += taosEncodeFixedI8(buf, pRsp->vnodeChanged);
+  tlen += taosEncodeFixedI8(buf, pRsp->epChanged);
+  tlen += taosEncodeSEpSet(buf, &pRsp->epSet);
+  return tlen;
+}
+
+static FORCE_INLINE void* taosDecodeSMqHbRsp(void* buf, SMqHbRsp* pRsp) {
+  buf = taosDecodeFixedI8(buf, &pRsp->status);
+  buf = taosDecodeFixedI8(buf, &pRsp->vnodeChanged);
+  buf = taosDecodeFixedI8(buf, &pRsp->epChanged);
+  buf = taosDecodeSEpSet(buf, &pRsp->epSet);
+  return buf;
+}
+
 typedef struct {
   int32_t acctId;
   int64_t clusterId;

include/common/tmsgdef.h

@@ -129,7 +129,7 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_MND_VGROUP_LIST, "mnode-vgroup-list", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_KILL_QUERY, "mnode-kill-query", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_KILL_CONN, "mnode-kill-conn", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_HEARTBEAT, "mnode-heartbeat", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_HEARTBEAT, "mnode-heartbeat", SClientHbBatchReq, SClientHbBatchRsp)
   TD_DEF_MSG_TYPE(TDMT_MND_SHOW, "mnode-show", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_SHOW_RETRIEVE, "mnode-retrieve", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_STATUS, "mnode-status", NULL, NULL)

source/client/inc/clientHb.h

-/*
- * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
- *
- * This program is free software: you can use, redistribute, and/or modify
- * it under the terms of the GNU Affero General Public License, version 3
- * or later ("AGPL"), as published by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
-
-#include "os.h"
-#include "tarray.h"
-#include "thash.h"
-#include "tmsg.h"
-
-#define HEARTBEAT_INTERVAL 1500  // ms
-
-typedef enum {
-  HEARTBEAT_TYPE_MQ = 0,
-  // types can be added here
-  //
-  HEARTBEAT_TYPE_MAX
-} EHbType;
-
-typedef int32_t (*FHbRspHandle)(SClientHbRsp* pReq);
-
-typedef struct SAppHbMgr {
-  // statistics
-  int32_t reportCnt;
-  int32_t connKeyCnt;
-  int64_t reportBytes;  // not implemented
-  int64_t startTime;
-  // ctl
-  SRWLatch lock;  // lock is used in serialization
-  // connection
-  void*  transporter;
-  SEpSet epSet;
-  // info
-  SHashObj* activeInfo;    // hash<SClientHbKey, SClientHbReq>
-  SHashObj* getInfoFuncs;  // hash<SClientHbKey, FGetConnInfo>
-} SAppHbMgr;
-
-typedef struct SClientHbMgr {
-  int8_t inited;
-  // ctl
-  int8_t          threadStop;
-  pthread_t       thread;
-  pthread_mutex_t lock;       // used when app init and cleanup
-  SArray*         appHbMgrs;  // SArray<SAppHbMgr*> one for each cluster
-  FHbRspHandle    handle[HEARTBEAT_TYPE_MAX];
-} SClientHbMgr;
-
-// TODO: embed param into function
-// return type: SArray<Skv>
-typedef SArray* (*FGetConnInfo)(SClientHbKey connKey, void* param);
-
-// global, called by mgmt
-int  hbMgrInit();
-void hbMgrCleanUp();
-int  hbHandleRsp(SClientHbBatchRsp* hbRsp);
-
-// cluster level
-SAppHbMgr* appHbMgrInit(void* transporter, SEpSet epSet);
-void appHbMgrCleanup(SAppHbMgr* pAppHbMgr);
-
-// conn level
-int  hbRegisterConn(SAppHbMgr* pAppHbMgr, SClientHbKey connKey, FGetConnInfo func);
-void hbDeregisterConn(SAppHbMgr* pAppHbMgr, SClientHbKey connKey);
-
-int hbAddConnInfo(SAppHbMgr* pAppHbMgr, SClientHbKey connKey, void* key, void* value, int32_t keyLen, int32_t valueLen);
-
-// mq
-void hbMgrInitMqHbRspHandle();

source/client/inc/clientInt.h

@@ -31,6 +31,41 @@ extern "C" {
 #include "trpc.h"
 #include "query.h"
 
+#define HEARTBEAT_INTERVAL 1500  // ms
+
+typedef struct SAppInstInfo SAppInstInfo;
+
+typedef int32_t (*FHbRspHandle)(SClientHbRsp* pReq);
+
+typedef struct SAppHbMgr {
+  // statistics
+  int32_t reportCnt;
+  int32_t connKeyCnt;
+  int64_t reportBytes;  // not implemented
+  int64_t startTime;
+  // ctl
+  SRWLatch lock;  // lock is used in serialization
+  // connection
+  SAppInstInfo* pAppInstInfo;
+  // info
+  SHashObj* activeInfo;    // hash<SClientHbKey, SClientHbReq>
+  SHashObj* getInfoFuncs;  // hash<SClientHbKey, FGetConnInfo>
+} SAppHbMgr;
+
+typedef struct SClientHbMgr {
+  int8_t inited;
+  // ctl
+  int8_t          threadStop;
+  pthread_t       thread;
+  pthread_mutex_t lock;       // used when app init and cleanup
+  SArray*         appHbMgrs;  // SArray<SAppHbMgr*> one for each cluster
+  FHbRspHandle    handle[HEARTBEAT_TYPE_MAX];
+} SClientHbMgr;
+
+// TODO: embed param into function
+// return type: SArray<Skv>
+typedef SArray* (*FGetConnInfo)(SClientHbKey connKey, void* param);
+
 typedef struct SQueryExecMetric {
   int64_t      start;    // start timestamp
   int64_t      parsed;   // start to parse
@@ -55,15 +90,15 @@ typedef struct SHeartBeatInfo {
   void  *pTimer;   // timer, used to send request msg to mnode
 } SHeartBeatInfo;
 
-typedef struct SAppInstInfo {
-  int64_t           numOfConns;
-  SCorEpSet         mgmtEp;
-  SInstanceSummary  summary;
+struct SAppInstInfo {
+  int64_t          numOfConns;
+  SCorEpSet        mgmtEp;
+  SInstanceSummary summary;
   SList            *pConnList;  // STscObj linked list
-  int64_t           clusterId;
+  int64_t          clusterId;
   void             *pTransporter;
-  SHeartBeatInfo hb;
-} SAppInstInfo;
+  struct SAppHbMgr *pAppHbMgr;
+};
 
 typedef struct SAppInfo {
   int64_t        startTime;
@@ -81,6 +116,7 @@ typedef struct STscObj {
   char             db[TSDB_DB_FNAME_LEN];
   int32_t          acctId;
   uint32_t         connId;
+  int32_t          connType;
   uint64_t         id;       // ref ID returned by taosAddRef
   void            *pTransporter;
   pthread_mutex_t  mutex;     // used to protect the operation on db
@@ -88,6 +124,10 @@ typedef struct STscObj {
   SAppInstInfo    *pAppInfo;
 } STscObj;
 
+typedef struct SMqConsumer {
+  STscObj* pTscObj;
+} SMqConsumer;
+
 typedef struct SReqResultInfo {
   const char  *pRspMsg;
   const char  *pData;
@@ -169,6 +209,26 @@ void *doFetchRow(SRequestObj* pRequest);
 
 void  setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32_t numOfCols, int32_t numOfRows);
 
+// --- heartbeat 
+// global, called by mgmt
+int  hbMgrInit();
+void hbMgrCleanUp();
+int  hbHandleRsp(SClientHbBatchRsp* hbRsp);
+
+// cluster level
+SAppHbMgr* appHbMgrInit(SAppInstInfo* pAppInstInfo);
+void appHbMgrCleanup(SAppHbMgr* pAppHbMgr);
+
+// conn level
+int  hbRegisterConn(SAppHbMgr* pAppHbMgr, SClientHbKey connKey, FGetConnInfo func);
+void hbDeregisterConn(SAppHbMgr* pAppHbMgr, SClientHbKey connKey);
+
+int hbAddConnInfo(SAppHbMgr* pAppHbMgr, SClientHbKey connKey, void* key, void* value, int32_t keyLen, int32_t valueLen);
+
+// --- mq
+void hbMgrInitMqHbRspHandle();
+
+
 #ifdef __cplusplus
 }
 #endif

source/client/src/clientHb.c

@@ -13,7 +13,7 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "clientHb.h"
+#include "clientInt.h"
 #include "trpc.h"
 
 static SClientHbMgr clientHbMgr = {0};
@@ -21,10 +21,18 @@ static SClientHbMgr clientHbMgr = {0};
 static int32_t hbCreateThread();
 static void    hbStopThread();
 
-static int32_t hbMqHbRspHandle(SClientHbRsp* pReq) {
+static int32_t hbMqHbRspHandle(SClientHbRsp* pRsp) {
   return 0;
 }
 
+static int32_t hbMqAsyncCallBack(void* param, const SDataBuf* pMsg, int32_t code) {
+  if (code != 0) {
+    return -1;
+  }
+  SClientHbRsp* pRsp = (SClientHbRsp*) pMsg->pData;
+  return hbMqHbRspHandle(pRsp);
+}
+
 void hbMgrInitMqHbRspHandle() {
   clientHbMgr.handle[HEARTBEAT_TYPE_MQ] = hbMqHbRspHandle;
 }
@@ -35,18 +43,18 @@ static FORCE_INLINE void hbMgrInitHandle() {
 }
 
 SClientHbBatchReq* hbGatherAllInfo(SAppHbMgr *pAppHbMgr) {
-  SClientHbBatchReq* pReq = malloc(sizeof(SClientHbBatchReq));
-  if (pReq == NULL) {
+  SClientHbBatchReq* pBatchReq = malloc(sizeof(SClientHbBatchReq));
+  if (pBatchReq == NULL) {
     terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
     return NULL;
   }
   int32_t connKeyCnt = atomic_load_32(&pAppHbMgr->connKeyCnt);
-  pReq->reqs = taosArrayInit(connKeyCnt, sizeof(SClientHbReq));
+  pBatchReq->reqs = taosArrayInit(connKeyCnt, sizeof(SClientHbReq));
 
   void *pIter = taosHashIterate(pAppHbMgr->activeInfo, NULL);
   while (pIter != NULL) {
-    taosArrayPush(pReq->reqs, pIter);
     SClientHbReq* pOneReq = pIter;
+    taosArrayPush(pBatchReq->reqs, pOneReq);
     taosHashClear(pOneReq->info);
 
     pIter = taosHashIterate(pAppHbMgr->activeInfo, pIter);
@@ -59,10 +67,10 @@ SClientHbBatchReq* hbGatherAllInfo(SAppHbMgr *pAppHbMgr) {
     taosHashCopyKey(pIter, &connKey);
     getConnInfoFp(connKey, NULL);
 
-    pIter = taosHashIterate(pAppHbMgr->activeInfo, pIter);
+    pIter = taosHashIterate(pAppHbMgr->getInfoFuncs, pIter);
   }
 
-  return pReq;
+  return pBatchReq;
 }
 
 static void* hbThreadFunc(void* param) {
@@ -75,20 +83,48 @@ static void* hbThreadFunc(void* param) {
 
     int sz = taosArrayGetSize(clientHbMgr.appHbMgrs);
     for(int i = 0; i < sz; i++) {
-      SAppHbMgr* pAppHbMgr = taosArrayGet(clientHbMgr.appHbMgrs, i);
-      SClientHbBatchReq* pReq = hbGatherAllInfo(pAppHbMgr);
-      void* reqStr = NULL;
-      int tlen = tSerializeSClientHbBatchReq(&reqStr, pReq);
-      SMsgSendInfo info;
-      /*info.fp = hbHandleRsp;*/
+      SAppHbMgr* pAppHbMgr = taosArrayGetP(clientHbMgr.appHbMgrs, i);
 
+      int32_t connCnt = atomic_load_32(&pAppHbMgr->connKeyCnt);
+      if (connCnt == 0) {
+        continue;
+      }
+      SClientHbBatchReq* pReq = hbGatherAllInfo(pAppHbMgr);
+      if (pReq == NULL) {
+        continue;
+      }
+      int tlen = tSerializeSClientHbBatchReq(NULL, pReq);
+      void *buf = malloc(tlen);
+      if (buf == NULL) {
+        //TODO: error handling
+        break;
+      }
+      void *bufCopy = buf;
+      tSerializeSClientHbBatchReq(&bufCopy, pReq);
+      SMsgSendInfo *pInfo = malloc(sizeof(SMsgSendInfo));
+      if (pInfo == NULL) {
+        terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
+        tFreeClientHbBatchReq(pReq);
+        free(buf);
+        break;
+      }
+      pInfo->fp = hbMqAsyncCallBack;
+      pInfo->msgInfo.pData = buf;
+      pInfo->msgInfo.len = tlen;
+      pInfo->msgType = TDMT_MND_HEARTBEAT;
+      pInfo->param = NULL;
+      pInfo->requestId = generateRequestId();
+      pInfo->requestObjRefId = 0;
+
+      SAppInstInfo *pAppInstInfo = pAppHbMgr->pAppInstInfo;
       int64_t transporterId = 0;
-      asyncSendMsgToServer(pAppHbMgr->transporter, &pAppHbMgr->epSet, &transporterId, &info);
+      SEpSet epSet = getEpSet_s(&pAppInstInfo->mgmtEp);
+      asyncSendMsgToServer(pAppInstInfo->pTransporter, &epSet, &transporterId, pInfo);
       tFreeClientHbBatchReq(pReq);
 
       atomic_add_fetch_32(&pAppHbMgr->reportCnt, 1);
-      taosMsleep(HEARTBEAT_INTERVAL);
     }
+    taosMsleep(HEARTBEAT_INTERVAL);
   }
   return NULL;
 }
@@ -110,7 +146,8 @@ static void hbStopThread() {
   atomic_store_8(&clientHbMgr.threadStop, 1);
 }
 
-SAppHbMgr* appHbMgrInit(void* transporter, SEpSet epSet) {
+SAppHbMgr* appHbMgrInit(SAppInstInfo* pAppInstInfo) {
+  hbMgrInit();
   SAppHbMgr* pAppHbMgr = malloc(sizeof(SAppHbMgr)); 
   if (pAppHbMgr == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
@@ -119,16 +156,27 @@ SAppHbMgr* appHbMgrInit(void* transporter, SEpSet epSet) {
   // init stat
   pAppHbMgr->startTime = taosGetTimestampMs();
 
-  // init connection info
-  pAppHbMgr->transporter = transporter;
-  pAppHbMgr->epSet = epSet;
+  // init app info
+  pAppHbMgr->pAppInstInfo = pAppInstInfo;
 
   // init hash info
   pAppHbMgr->activeInfo = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK);
+
+  if (pAppHbMgr->activeInfo == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    free(pAppHbMgr);
+    return NULL;
+  }
   pAppHbMgr->activeInfo->freeFp = tFreeClientHbReq;
   // init getInfoFunc
   pAppHbMgr->getInfoFuncs = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK);
 
+  if (pAppHbMgr->getInfoFuncs == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    free(pAppHbMgr);
+    return NULL;
+  }
+
   taosArrayPush(clientHbMgr.appHbMgrs, &pAppHbMgr);
   return pAppHbMgr;
 }
@@ -138,7 +186,7 @@ void appHbMgrCleanup(SAppHbMgr* pAppHbMgr) {
 
   int sz = taosArrayGetSize(clientHbMgr.appHbMgrs);
   for (int i = 0; i < sz; i++) {
-    SAppHbMgr* pTarget = taosArrayGet(clientHbMgr.appHbMgrs, i);
+    SAppHbMgr* pTarget = taosArrayGetP(clientHbMgr.appHbMgrs, i);
     if (pAppHbMgr == pTarget) {
       taosHashCleanup(pTarget->activeInfo);
       taosHashCleanup(pTarget->getInfoFuncs);
@@ -171,7 +219,6 @@ void hbMgrCleanUp() {
   if (old == 0) return;
 
   taosArrayDestroy(clientHbMgr.appHbMgrs);
-
 }
 
 int hbHandleRsp(SClientHbBatchRsp* hbRsp) {

source/client/src/clientImpl.c

@@ -116,6 +116,7 @@ TAOS *taos_connect_internal(const char *ip, const char *user, const char *pass,
     SAppInstInfo* p = calloc(1, sizeof(struct SAppInstInfo));
     p->mgmtEp       = epSet;
     p->pTransporter = openTransporter(user, secretEncrypt, tsNumOfCores);
+    p->pAppHbMgr = appHbMgrInit(p);
     taosHashPut(appInfo.pInstMap, key, strlen(key), &p, POINTER_BYTES);
 
     pInst = &p;
@@ -258,6 +259,101 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryDag* pDag) {
   return scheduleAsyncExecJob(pRequest->pTscObj->pTransporter, NULL, pDag, &pRequest->body.pQueryJob);
 }
 
+typedef struct tmq_t tmq_t;
+
+typedef struct SMqClientTopic {
+  // subscribe info
+  int32_t sqlLen;
+  char*   sql;
+  char*   topicName;
+  int64_t topicId;
+  // statistics
+  int64_t consumeCnt;
+  // offset
+  int64_t committedOffset;
+  int64_t currentOffset;
+  //connection info
+  int32_t vgId;
+  SEpSet  epSet;
+} SMqClientTopic;
+
+typedef struct tmq_resp_err_t {
+  int32_t code;
+} tmq_resp_err_t;
+
+typedef struct tmq_topic_vgroup_list_t {
+  char* topicName;
+  int32_t vgId;
+  int64_t committedOffset;
+} tmq_topic_vgroup_list_t;
+
+typedef void (tmq_commit_cb(tmq_t*, tmq_resp_err_t, tmq_topic_vgroup_list_t*, void* param));
+
+typedef struct tmq_conf_t{
+  char*          clientId;
+  char*          groupId;
+  char*          ip;
+  uint16_t       port;
+  tmq_commit_cb* commit_cb;
+} tmq_conf_t;
+
+struct tmq_t {
+  char           groupId[256];
+  char           clientId[256];
+  STscObj*       pTscObj;
+  tmq_commit_cb* commit_cb;
+  SArray*        clientTopics;  // SArray<SMqClientTopic>
+};
+
+void tmq_conf_set_offset_commit_cb(tmq_conf_t* conf, tmq_commit_cb* cb) {
+  conf->commit_cb = cb;
+}
+
+SArray* tmqGetConnInfo(SClientHbKey connKey, void* param) {
+  tmq_t* pTmq = (void*)param;
+  SArray* pArray = taosArrayInit(0, sizeof(SKv));
+  if (pArray == NULL) {
+    return NULL;
+  }
+  SKv kv = {0};
+  kv.key = malloc(256);
+  if (kv.key == NULL) {
+    taosArrayDestroy(pArray);
+    return NULL;
+  }
+  strcpy(kv.key, "groupId");
+  kv.keyLen = strlen("groupId") + 1;
+  kv.value = malloc(256);
+  if (kv.value == NULL) {
+    free(kv.key);
+    taosArrayDestroy(pArray);
+    return NULL;
+  }
+  strcpy(kv.value, pTmq->groupId);
+  kv.valueLen = strlen(pTmq->groupId) + 1;
+
+  taosArrayPush(pArray, &kv);
+  strcpy(kv.key, "clientUid");
+  kv.keyLen = strlen("clientUid") + 1;
+  *(uint32_t*)kv.value = pTmq->pTscObj->connId;
+  kv.valueLen = sizeof(uint32_t); 
+  
+  return NULL;
+}
+
+tmq_t* tmqCreateConsumerImpl(TAOS* conn, tmq_conf_t* conf) {
+  tmq_t* pTmq = malloc(sizeof(tmq_t));
+  if (pTmq == NULL) {
+    return NULL;
+  }
+  strcpy(pTmq->groupId, conf->groupId);
+  strcpy(pTmq->clientId, conf->clientId);
+  pTmq->pTscObj = (STscObj*)conn;
+  pTmq->pTscObj->connType = HEARTBEAT_TYPE_MQ;
+
+  return pTmq;
+}
+
 TAOS_RES *tmq_create_topic(TAOS* taos, const char* name, const char* sql, int sqlLen) {
   STscObj* pTscObj = (STscObj*)taos;
   SRequestObj* pRequest = NULL;
@@ -319,6 +415,25 @@ _return:
   return pRequest;
 }
 
+typedef struct tmq_message_t {
+  int32_t  numOfRows; 
+  char*    topicName;
+  TAOS_ROW row[];
+} tmq_message_t;
+
+tmq_message_t* tmq_consume_poll(tmq_t* mq, int64_t blocking_time) {
+  return NULL;
+}
+
+tmq_resp_err_t* tmq_commit(tmq_t* mq, void* callback, int32_t async) {
+  return NULL;
+}
+
+void tmq_message_destroy(tmq_message_t* mq_message) {
+  
+}
+
+
 TAOS_RES *taos_query_l(TAOS *taos, const char *sql, int sqlLen) {
   STscObj *pTscObj = (STscObj *)taos;
   if (sqlLen > (size_t) tsMaxSQLStringLen) {
@@ -673,4 +788,4 @@ void setQueryResultByRsp(SReqResultInfo* pResultInfo, const SRetrieveTableRsp* p
   pResultInfo->completed = (pRsp->completed == 1);
 
   setResultDataPtr(pResultInfo, pResultInfo->fields, pResultInfo->numOfCols, pResultInfo->numOfRows);
-}
\ No newline at end of file
+}

source/client/src/clientMsgHandler.c

@@ -71,6 +71,9 @@ int processConnectRsp(void* param, const SDataBuf* pMsg, int32_t code) {
   pTscObj->pAppInfo->clusterId = pConnect->clusterId;
   atomic_add_fetch_64(&pTscObj->pAppInfo->numOfConns, 1);
 
+  SClientHbKey connKey = {.connId = pConnect->connId, .hbType = HEARTBEAT_TYPE_QUERY};
+  hbRegisterConn(pTscObj->pAppInfo->pAppHbMgr, connKey, NULL);
+
   //  pRequest->body.resInfo.pRspMsg = pMsg->pData;
   tscDebug("0x%" PRIx64 " clusterId:%" PRId64 ", totalConn:%" PRId64, pRequest->requestId, pConnect->clusterId,
            pTscObj->pAppInfo->numOfConns);
@@ -379,4 +382,4 @@ void initMsgHandleFp() {
 
   handleRequestRspFp[TMSG_INDEX(TDMT_VND_SHOW_TABLES)]   = processShowRsp;
   handleRequestRspFp[TMSG_INDEX(TDMT_VND_SHOW_TABLES_FETCH)]   = processRetrieveVndRsp;
-}
\ No newline at end of file
+}

source/client/test/clientTests.cpp

@@ -53,6 +53,7 @@ TEST(testCase, connect_Test) {
   if (pConn == NULL) {
     printf("failed to connect to server, reason:%s\n", taos_errstr(NULL));
   }
+  sleep(3);
   taos_close(pConn);
 }
 
@@ -148,27 +149,27 @@ TEST(testCase, connect_Test) {
 //}
 //
 //TEST(testCase, create_db_Test) {
-//  TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
-//  assert(pConn != NULL);
-//
-//  TAOS_RES* pRes = taos_query(pConn, "create database abc1 vgroups 2");
-//  if (taos_errno(pRes) != 0) {
-//    printf("error in create db, reason:%s\n", taos_errstr(pRes));
-//  }
-//
-//  TAOS_FIELD* pFields = taos_fetch_fields(pRes);
-//  ASSERT_TRUE(pFields == NULL);
-//
-//  int32_t numOfFields = taos_num_fields(pRes);
-//  ASSERT_EQ(numOfFields, 0);
-//
-//  taos_free_result(pRes);
-//
-//  pRes = taos_query(pConn, "create database abc1 vgroups 4");
-//  if (taos_errno(pRes) != 0) {
-//    printf("error in create db, reason:%s\n", taos_errstr(pRes));
-//  }
-//  taos_close(pConn);
+  //TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
+  //assert(pConn != NULL);
+
+  //TAOS_RES* pRes = taos_query(pConn, "create database abc1 vgroups 2");
+  //if (taos_errno(pRes) != 0) {
+    //printf("error in create db, reason:%s\n", taos_errstr(pRes));
+  //}
+
+  //TAOS_FIELD* pFields = taos_fetch_fields(pRes);
+  //ASSERT_TRUE(pFields == NULL);
+
+  //int32_t numOfFields = taos_num_fields(pRes);
+  //ASSERT_EQ(numOfFields, 0);
+
+  //taos_free_result(pRes);
+
+  //pRes = taos_query(pConn, "create database abc1 vgroups 4");
+  //if (taos_errno(pRes) != 0) {
+    //printf("error in create db, reason:%s\n", taos_errstr(pRes));
+  //}
+  //taos_close(pConn);
 //}
 //
 //TEST(testCase, create_dnode_Test) {
@@ -293,24 +294,24 @@ TEST(testCase, connect_Test) {
 //  taos_close(pConn);
 //}
 
-TEST(testCase, create_ctable_Test) {
-  TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
-  assert(pConn != NULL);
+//TEST(testCase, create_ctable_Test) {
+  //TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
+  //assert(pConn != NULL);
 
-  TAOS_RES* pRes = taos_query(pConn, "use abc1");
-  if (taos_errno(pRes) != 0) {
-    printf("failed to use db, reason:%s\n", taos_errstr(pRes));
-  }
-  taos_free_result(pRes);
+  //TAOS_RES* pRes = taos_query(pConn, "use abc1");
+  //if (taos_errno(pRes) != 0) {
+    //printf("failed to use db, reason:%s\n", taos_errstr(pRes));
+  //}
+  //taos_free_result(pRes);
 
-  pRes = taos_query(pConn, "create table tm0 using st1 tags(1)");
-  if (taos_errno(pRes) != 0) {
-    printf("failed to create child table tm0, reason:%s\n", taos_errstr(pRes));
-  }
+  //pRes = taos_query(pConn, "create table tm0 using st1 tags(1)");
+  //if (taos_errno(pRes) != 0) {
+    //printf("failed to create child table tm0, reason:%s\n", taos_errstr(pRes));
+  //}
 
-  taos_free_result(pRes);
-  taos_close(pConn);
-}
+  //taos_free_result(pRes);
+  //taos_close(pConn);
+//}
 
 //TEST(testCase, show_stable_Test) {
 //  TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);

source/common/src/tmsg.c

@@ -89,7 +89,7 @@ int tSerializeSClientHbReq(void **buf, const SClientHbReq *pReq) {
   int tlen = 0;
   tlen += taosEncodeSClientHbKey(buf, &pReq->connKey);
 
-  int kvNum = taosHashGetSize(pReq->info);
+  int32_t kvNum = taosHashGetSize(pReq->info);
   tlen += taosEncodeFixedI32(buf, kvNum);
   SKv kv;
   void* pIter = taosHashIterate(pReq->info, pIter);
@@ -104,14 +104,15 @@ int tSerializeSClientHbReq(void **buf, const SClientHbReq *pReq) {
   return tlen;
 }
 
-void *tDeserializeClientHbReq(void *buf, SClientHbReq *pReq) {
-  ASSERT(pReq->info != NULL);
+void *tDeserializeSClientHbReq(void *buf, SClientHbReq *pReq) {
   buf = taosDecodeSClientHbKey(buf, &pReq->connKey);
 
   // TODO: error handling
-  int kvNum;
-  taosDecodeFixedI32(buf, &kvNum);
-  pReq->info = taosHashInit(kvNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
+  int32_t kvNum;
+  buf = taosDecodeFixedI32(buf, &kvNum);
+  if (pReq->info == NULL) {
+    pReq->info = taosHashInit(kvNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
+  }
   for(int i = 0; i < kvNum; i++) {
     SKv kv;
     buf = taosDecodeSKv(buf, &kv);
@@ -121,12 +122,69 @@ void *tDeserializeClientHbReq(void *buf, SClientHbReq *pReq) {
   return buf;
 }
 
-int   tSerializeSClientHbBatchReq(void** buf, const SClientHbBatchReq* pReq) {
+int   tSerializeSClientHbRsp(void** buf, const SClientHbRsp* pRsp) {
+  int tlen = 0;
+  tlen += taosEncodeSClientHbKey(buf, &pRsp->connKey);
+  tlen += taosEncodeFixedI32(buf, pRsp->status);
+  tlen += taosEncodeFixedI32(buf, pRsp->bodyLen);
+  tlen += taosEncodeBinary(buf, pRsp->body, pRsp->bodyLen);
+  return tlen;
+}
+void* tDeserializeSClientHbRsp(void* buf, SClientHbRsp* pRsp) {
+  buf = taosDecodeSClientHbKey(buf, &pRsp->connKey);
+  buf = taosDecodeFixedI32(buf, &pRsp->status);
+  buf = taosDecodeFixedI32(buf, &pRsp->bodyLen);
+  buf = taosDecodeBinary(buf, &pRsp->body, pRsp->bodyLen);
+  return buf;
+}
+
+int   tSerializeSClientHbBatchReq(void** buf, const SClientHbBatchReq* pBatchReq) {
   int tlen = 0;
+  tlen += taosEncodeFixedI64(buf, pBatchReq->reqId);
+  int32_t reqNum = taosArrayGetSize(pBatchReq->reqs);
+  tlen += taosEncodeFixedI32(buf, reqNum); 
+  for (int i = 0; i < reqNum; i++) {
+    SClientHbReq* pReq = taosArrayGet(pBatchReq->reqs, i);
+    tlen += tSerializeSClientHbReq(buf, pReq);
+  }
   return tlen;
 }
 
-void* tDeserializeClientHbBatchReq(void* buf, SClientHbBatchReq* pReq) {
+void* tDeserializeSClientHbBatchReq(void* buf, SClientHbBatchReq* pBatchReq) {
+  buf = taosDecodeFixedI64(buf, &pBatchReq->reqId);
+  if (pBatchReq->reqs == NULL) {
+    pBatchReq->reqs = taosArrayInit(0, sizeof(SClientHbReq));
+  }
+  int32_t reqNum;
+  buf = taosDecodeFixedI32(buf, &reqNum);
+  for (int i = 0; i < reqNum; i++) {
+    SClientHbReq req = {0};
+    buf = tDeserializeSClientHbReq(buf, &req);
+    taosArrayPush(pBatchReq->reqs, &req);
+  }
+  return buf;
+}
+
+int   tSerializeSClientHbBatchRsp(void** buf, const SClientHbBatchRsp* pBatchRsp) {
+  int tlen = 0;
+  int32_t sz = taosArrayGetSize(pBatchRsp->rsps);
+  tlen += taosEncodeFixedI32(buf, sz);
+  for (int i = 0; i < sz; i++) {
+    SClientHbRsp* pRsp = taosArrayGet(pBatchRsp->rsps, i);
+    tlen += tSerializeSClientHbRsp(buf, pRsp);
+  }
+  return tlen;
+}
+
+void* tDeserializeSClientHbBatchRsp(void* buf, SClientHbBatchRsp* pBatchRsp) {
+  int32_t sz;
+  buf = taosDecodeFixedI32(buf, &sz);
+  pBatchRsp->rsps = taosArrayInit(sz, sizeof(SClientHbRsp));
+  for (int i = 0; i < sz; i++) {
+    SClientHbRsp rsp = {0};
+    buf = tDeserializeSClientHbRsp(buf, &rsp); 
+    taosArrayPush(pBatchRsp->rsps, &rsp);
+  }
   return buf;
 }
 

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

@@ -258,6 +258,39 @@ static int32_t mndSaveQueryStreamList(SConnObj *pConn, SHeartBeatReq *pReq) {
 }
 
 static int32_t mndProcessHeartBeatReq(SMnodeMsg *pReq) {
+  SMnode *pMnode = pReq->pMnode;
+  char *batchReqStr = pReq->rpcMsg.pCont;
+  SClientHbBatchReq batchReq = {0};
+  tDeserializeSClientHbBatchReq(batchReqStr, &batchReq);
+  SArray *pArray = batchReq.reqs;
+  int sz = taosArrayGetSize(pArray);
+
+  SClientHbBatchRsp batchRsp = {0};
+  batchRsp.rsps = taosArrayInit(0, sizeof(SClientHbRsp));
+
+  for (int i = 0; i < sz; i++) {
+    SClientHbReq* pHbReq = taosArrayGet(pArray, i);
+    if (pHbReq->connKey.hbType == HEARTBEAT_TYPE_QUERY) {
+
+    } else if (pHbReq->connKey.hbType == HEARTBEAT_TYPE_MQ) {
+      SClientHbRsp rsp = {
+        .status = 0,
+        .connKey = pHbReq->connKey,
+        .bodyLen = 0,
+        .body = NULL
+      };
+      taosArrayPush(batchRsp.rsps, &rsp);
+    }
+  }
+  int32_t tlen = tSerializeSClientHbBatchRsp(NULL, &batchRsp);
+  void* buf = rpcMallocCont(tlen);
+  void* bufCopy = buf;
+  tSerializeSClientHbBatchRsp(&bufCopy, &batchRsp);
+  pReq->contLen = tlen;
+  pReq->pCont = buf;
+  return 0;
+
+#if 0
   SMnode       *pMnode = pReq->pMnode;
   SProfileMgmt *pMgmt = &pMnode->profileMgmt;
 
@@ -327,6 +360,7 @@ static int32_t mndProcessHeartBeatReq(SMnodeMsg *pReq) {
   pReq->contLen = sizeof(SConnectRsp);
   pReq->pCont = pRsp;
   return 0;
+#endif
 }
 
 static int32_t mndProcessKillQueryReq(SMnodeMsg *pReq) {

source/dnode/mnode/impl/test/profile/profile.cpp

@@ -96,6 +96,38 @@ TEST_F(MndTestProfile, 03_ConnectMsg_Show) {
 }
 
 TEST_F(MndTestProfile, 04_HeartBeatMsg) {
+  SClientHbBatchReq batchReq;
+  batchReq.reqs = taosArrayInit(0, sizeof(SClientHbReq));
+  SClientHbReq req = {0};
+  req.connKey = {.connId = 123, .hbType = HEARTBEAT_TYPE_MQ};
+  req.info = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK);
+  SKv kv;
+  kv.key = (void*)"abc";
+  kv.keyLen = 4;
+  kv.value = (void*)"bcd";
+  kv.valueLen = 4;
+  taosHashPut(req.info, kv.key, kv.keyLen, kv.value, kv.valueLen);
+  taosArrayPush(batchReq.reqs, &req);
+
+  int32_t tlen = tSerializeSClientHbBatchReq(NULL, &batchReq);
+  
+  void* buf = (SClientHbBatchReq*)rpcMallocCont(tlen);
+  void* bufCopy = buf;
+  tSerializeSClientHbBatchReq(&bufCopy, &batchReq);
+  SRpcMsg* pMsg = test.SendReq(TDMT_MND_HEARTBEAT, buf, tlen);
+  ASSERT_NE(pMsg, nullptr);
+  ASSERT_EQ(pMsg->code, 0);
+  char* pRspChar = (char*)pMsg->pCont;
+  SClientHbBatchRsp rsp = {0};
+  tDeserializeSClientHbBatchRsp(pRspChar, &rsp);
+  int sz = taosArrayGetSize(rsp.rsps);
+  ASSERT_EQ(sz, 1);
+  SClientHbRsp* pRsp = (SClientHbRsp*) taosArrayGet(rsp.rsps, 0);
+  EXPECT_EQ(pRsp->connKey.connId, 123);
+  EXPECT_EQ(pRsp->connKey.hbType, HEARTBEAT_TYPE_MQ);
+  EXPECT_EQ(pRsp->status, 0);
+
+#if 0
   int32_t contLen = sizeof(SHeartBeatReq);
 
   SHeartBeatReq* pReq = (SHeartBeatReq*)rpcMallocCont(contLen);
@@ -129,9 +161,12 @@ TEST_F(MndTestProfile, 04_HeartBeatMsg) {
   EXPECT_EQ(pRsp->epSet.numOfEps, 1);
   EXPECT_EQ(pRsp->epSet.port[0], 9031);
   EXPECT_STREQ(pRsp->epSet.fqdn[0], "localhost");
+#endif
 }
 
 TEST_F(MndTestProfile, 05_KillConnMsg) {
+  // temporary remove since kill will use new heartbeat msg
+#if 0
   {
     int32_t contLen = sizeof(SKillConnReq);
 
@@ -190,6 +225,7 @@ TEST_F(MndTestProfile, 05_KillConnMsg) {
 
     connId = pRsp->connId;
   }
+#endif
 }
 
 TEST_F(MndTestProfile, 06_KillConnMsg_InvalidConn) {
@@ -204,6 +240,8 @@ TEST_F(MndTestProfile, 06_KillConnMsg_InvalidConn) {
 }
 
 TEST_F(MndTestProfile, 07_KillQueryMsg) {
+  // temporary remove since kill will use new heartbeat msg
+#if 0
   {
     int32_t contLen = sizeof(SKillQueryReq);
 
@@ -252,6 +290,7 @@ TEST_F(MndTestProfile, 07_KillQueryMsg) {
     EXPECT_EQ(pRsp->epSet.port[0], 9031);
     EXPECT_STREQ(pRsp->epSet.fqdn[0], "localhost");
   }
+#endif
 }
 
 TEST_F(MndTestProfile, 08_KillQueryMsg_InvalidConn) {

source/libs/wal/src/walMeta.c

@@ -36,7 +36,7 @@ void* tmemmem(char* haystack, int hlen, char* needle, int nlen) {
   char* limit;
 
   if (nlen == 0 || hlen < nlen) {
-    return false;
+    return NULL;
   }
 
   limit = haystack + hlen - nlen + 1;
@@ -54,10 +54,12 @@ static inline int64_t walScanLogGetLastVer(SWal* pWal) {
   ASSERT(pWal->fileInfoSet != NULL);
   int sz = taosArrayGetSize(pWal->fileInfoSet);
   ASSERT(sz > 0);
+#if 0
   for (int i = 0; i < sz; i++) {
     SWalFileInfo* pFileInfo = taosArrayGet(pWal->fileInfoSet, i);
-
   }
+#endif
+
   SWalFileInfo *pLastFileInfo = taosArrayGet(pWal->fileInfoSet, sz-1);
   char fnameStr[WAL_FILE_LEN];
   walBuildLogName(pWal, pLastFileInfo->firstVer, fnameStr);
@@ -143,8 +145,6 @@ int walCheckAndRepairMeta(SWal* pWal) {
       SWalFileInfo fileInfo;
       memset(&fileInfo, -1, sizeof(SWalFileInfo));
       sscanf(name, "%" PRId64 ".log", &fileInfo.firstVer);
-      //get lastVer
-      //get size
       taosArrayPush(pLogInfoArray, &fileInfo);
     }
   }
@@ -158,60 +158,51 @@ int walCheckAndRepairMeta(SWal* pWal) {
     oldSz = taosArrayGetSize(pWal->fileInfoSet);
   }
   int newSz = taosArrayGetSize(pLogInfoArray);
-  // case 1. meta file not exist / cannot be parsed
-  if (oldSz < newSz) {
+
+  if (oldSz > newSz) {
+    taosArrayPopFrontBatch(pWal->fileInfoSet, oldSz - newSz); 
+  } else if (oldSz < newSz) {
     for (int i = oldSz; i < newSz; i++) {
       SWalFileInfo *pFileInfo = taosArrayGet(pLogInfoArray, i);
       taosArrayPush(pWal->fileInfoSet, pFileInfo);
     }
-
-    pWal->writeCur = newSz - 1;
-    pWal->vers.firstVer = ((SWalFileInfo*)taosArrayGet(pLogInfoArray, 0))->firstVer;
-    pWal->vers.lastVer = walScanLogGetLastVer(pWal);
-    ((SWalFileInfo*)taosArrayGetLast(pWal->fileInfoSet))->lastVer = pWal->vers.lastVer;
-    ASSERT(pWal->vers.lastVer != -1);
-
-    int code = walSaveMeta(pWal);
-    if (code < 0) {
-      taosArrayDestroy(pLogInfoArray);
-      return -1;
-    }
   }
-  
-  // case 2. versions in meta not match log 
-  //         or some log not included in meta
-  // (e.g. program killed)
-  //
-  // case 3. other corrupt cases
-  //
-#if 0
-  int sz = taosArrayGetSize(pLogInfoArray);
-  for (int i = 0; i < sz; i++) {
-    SWalFileInfo* pFileInfo = taosArrayGet(pLogInfoArray, i);
-    if (i == 0 && pFileInfo->firstVer != walGetFirstVer(pWal)) {
-      //repair
-    }
-
-    if (i > 0) {
-      SWalFileInfo* pLastFileInfo = taosArrayGet(pLogInfoArray, i-1);
-      if (pLastFileInfo->lastVer != pFileInfo->firstVer) {
+  taosArrayDestroy(pLogInfoArray);
 
+  pWal->writeCur = newSz - 1;
+  if (newSz > 0) {
+    pWal->vers.firstVer = ((SWalFileInfo*)taosArrayGet(pWal->fileInfoSet, 0))->firstVer;
+
+    SWalFileInfo *pLastFileInfo = taosArrayGet(pWal->fileInfoSet, newSz-1);
+    char fnameStr[WAL_FILE_LEN];
+    walBuildLogName(pWal, pLastFileInfo->firstVer, fnameStr);
+    struct stat statbuf;
+    stat(fnameStr, &statbuf);
+
+    if (oldSz != newSz || pLastFileInfo->fileSize != statbuf.st_size) {
+      pLastFileInfo->fileSize = statbuf.st_size;
+      pWal->vers.lastVer = walScanLogGetLastVer(pWal);
+      ((SWalFileInfo*)taosArrayGetLast(pWal->fileInfoSet))->lastVer = pWal->vers.lastVer;
+      ASSERT(pWal->vers.lastVer != -1);
+
+      int code = walSaveMeta(pWal);
+      if (code < 0) {
+        taosArrayDestroy(pLogInfoArray);
+        return -1;
       }
     }
   }
-#endif
-
 
-  // get last version of this file
-  //
-  // rebuild meta
-  taosArrayDestroy(pLogInfoArray);
+  //TODO: set fileSize and lastVer if necessary
+  
   return 0;
 }
 
 int walCheckAndRepairIdx(SWal* pWal) {
-  // iterate all idx files
-  // check first and last entry of each idx file valid
+  // TODO: iterate all log files
+  // if idx not found, scan log and write idx
+  // if found, check complete by first and last entry of each idx file
+  // if idx incomplete, binary search last valid entry, and then build other part
   return 0;
 }