mndConsumer.c 40.5 KB
Newer Older
L
Liu Jicong 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
/*
 * 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/>.
 */

#define _DEFAULT_SOURCE
#include "mndConsumer.h"
L
Liu Jicong 已提交
18
#include "mndPrivilege.h"
L
Liu Jicong 已提交
19
#include "mndShow.h"
20
#include "mndSubscribe.h"
L
Liu Jicong 已提交
21 22
#include "mndTopic.h"
#include "mndTrans.h"
L
Liu Jicong 已提交
23
#include "tcompare.h"
L
Liu Jicong 已提交
24 25
#include "tname.h"

26
#define MND_CONSUMER_VER_NUMBER   1
L
Liu Jicong 已提交
27 28
#define MND_CONSUMER_RESERVE_SIZE 64

H
Haojun Liao 已提交
29
#define MND_CONSUMER_LOST_HB_CNT          6
L
Liu Jicong 已提交
30
#define MND_CONSUMER_LOST_CLEAR_THRESHOLD 43200
31

H
Haojun Liao 已提交
32
static int32_t mqRebInExecCnt = 0;
33

L
Liu Jicong 已提交
34 35
static const char *mndConsumerStatusName(int status);

L
Liu Jicong 已提交
36 37 38
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);
S
Shengliang Guan 已提交
39 40
static int32_t mndProcessConsumerMetaMsg(SRpcMsg *pMsg);
static int32_t mndRetrieveConsumer(SRpcMsg *pMsg, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
L
Liu Jicong 已提交
41
static void    mndCancelGetNextConsumer(SMnode *pMnode, void *pIter);
L
Liu Jicong 已提交
42

S
Shengliang Guan 已提交
43 44
static int32_t mndProcessSubscribeReq(SRpcMsg *pMsg);
static int32_t mndProcessAskEpReq(SRpcMsg *pMsg);
45
static int32_t mndProcessMqHbReq(SRpcMsg *pMsg);
S
Shengliang Guan 已提交
46 47
static int32_t mndProcessMqTimerMsg(SRpcMsg *pMsg);
static int32_t mndProcessConsumerLostMsg(SRpcMsg *pMsg);
L
Liu Jicong 已提交
48
static int32_t mndProcessConsumerClearMsg(SRpcMsg *pMsg);
S
Shengliang Guan 已提交
49
static int32_t mndProcessConsumerRecoverMsg(SRpcMsg *pMsg);
50

L
Liu Jicong 已提交
51
int32_t mndInitConsumer(SMnode *pMnode) {
S
Shengliang Guan 已提交
52 53 54 55 56 57 58 59 60
  SSdbTable table = {
      .sdbType = SDB_CONSUMER,
      .keyType = SDB_KEY_INT64,
      .encodeFp = (SdbEncodeFp)mndConsumerActionEncode,
      .decodeFp = (SdbDecodeFp)mndConsumerActionDecode,
      .insertFp = (SdbInsertFp)mndConsumerActionInsert,
      .updateFp = (SdbUpdateFp)mndConsumerActionUpdate,
      .deleteFp = (SdbDeleteFp)mndConsumerActionDelete,
  };
L
Liu Jicong 已提交
61

L
Liu Jicong 已提交
62 63 64
  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_SUBSCRIBE, mndProcessSubscribeReq);
  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_HB, mndProcessMqHbReq);
  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_ASK_EP, mndProcessAskEpReq);
L
Liu Jicong 已提交
65
  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_TIMER, mndProcessMqTimerMsg);
L
Liu Jicong 已提交
66 67
  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_CONSUMER_LOST, mndProcessConsumerLostMsg);
  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_CONSUMER_RECOVER, mndProcessConsumerRecoverMsg);
L
Liu Jicong 已提交
68
  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_LOST_CONSUMER_CLEAR, mndProcessConsumerClearMsg);
L
Liu Jicong 已提交
69 70 71 72

  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_CONSUMERS, mndRetrieveConsumer);
  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_CONSUMERS, mndCancelGetNextConsumer);

L
Liu Jicong 已提交
73 74 75 76 77
  return sdbSetTable(pMnode->pSdb, table);
}

void mndCleanupConsumer(SMnode *pMnode) {}

L
Liu Jicong 已提交
78
bool mndRebTryStart() {
H
Haojun Liao 已提交
79 80
  int32_t old = atomic_val_compare_exchange_32(&mqRebInExecCnt, 0, 1);
  mInfo("tq timer, rebalance counter old val:%d", old);
L
Liu Jicong 已提交
81 82 83
  return old == 0;
}

S
Shengliang Guan 已提交
84
void mndRebEnd() {
85
  mndRebCntDec();
S
Shengliang Guan 已提交
86
}
L
Liu Jicong 已提交
87

S
Shengliang Guan 已提交
88 89
void mndRebCntInc() {
  int32_t val = atomic_add_fetch_32(&mqRebInExecCnt, 1);
90
  mInfo("rebalance trans start, rebalance counter:%d", val);
S
Shengliang Guan 已提交
91
}
L
Liu Jicong 已提交
92

S
Shengliang Guan 已提交
93
void mndRebCntDec() {
94 95 96 97 98 99 100 101 102 103 104 105 106 107
  while (1) {
    int32_t val = atomic_load_32(&mqRebInExecCnt);
    if (val <= 0) {
      mError("rebalance trans end, rebalance counter:%d should not be less equalled than 0, ignore counter desc", val);
      break;
    }

    int32_t newVal = val - 1;
    int32_t oldVal = atomic_val_compare_exchange_32(&mqRebInExecCnt, val, newVal);
    if (oldVal == val) {
      mInfo("rebalance trans end, rebalance counter:%d", newVal);
      break;
    }
  }
S
Shengliang Guan 已提交
108
}
L
Liu Jicong 已提交
109

S
Shengliang Guan 已提交
110 111 112
static int32_t mndProcessConsumerLostMsg(SRpcMsg *pMsg) {
  SMnode             *pMnode = pMsg->info.node;
  SMqConsumerLostMsg *pLostMsg = pMsg->pCont;
113
  SMqConsumerObj     *pConsumer = mndAcquireConsumer(pMnode, pLostMsg->consumerId);
L
Liu Jicong 已提交
114 115 116
  if (pConsumer == NULL) {
    return 0;
  }
117

X
Xiaoyu Wang 已提交
118
  mInfo("process consumer lost msg, consumer:0x%" PRIx64 " status:%d(%s)", pLostMsg->consumerId, pConsumer->status,
L
Liu Jicong 已提交
119 120
        mndConsumerStatusName(pConsumer->status));

D
dapan1121 已提交
121 122 123 124 125
  if (pConsumer->status != MQ_CONSUMER_STATUS__READY) {
    mndReleaseConsumer(pMnode, pConsumer);
    return -1;
  }

126 127 128 129 130
  SMqConsumerObj *pConsumerNew = tNewSMqConsumerObj(pConsumer->consumerId, pConsumer->cgroup);
  pConsumerNew->updateType = CONSUMER_UPDATE__LOST;

  mndReleaseConsumer(pMnode, pConsumer);

131
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pMsg, "lost-csm");
H
Haojun Liao 已提交
132 133 134 135 136 137 138 139 140 141 142
  if (pTrans == NULL) {
    goto FAIL;
  }

  if (mndSetConsumerCommitLogs(pMnode, pTrans, pConsumerNew) != 0) {
    goto FAIL;
  }

  if (mndTransPrepare(pMnode, pTrans) != 0) {
    goto FAIL;
  }
143

L
Liu Jicong 已提交
144
  tDeleteSMqConsumerObj(pConsumerNew);
145 146
  taosMemoryFree(pConsumerNew);
  mndTransDrop(pTrans);
147 148
  return 0;
FAIL:
L
Liu Jicong 已提交
149
  tDeleteSMqConsumerObj(pConsumerNew);
150
  taosMemoryFree(pConsumerNew);
151 152 153 154
  mndTransDrop(pTrans);
  return -1;
}

S
Shengliang Guan 已提交
155 156 157
static int32_t mndProcessConsumerRecoverMsg(SRpcMsg *pMsg) {
  SMnode                *pMnode = pMsg->info.node;
  SMqConsumerRecoverMsg *pRecoverMsg = pMsg->pCont;
L
Liu Jicong 已提交
158
  SMqConsumerObj        *pConsumer = mndAcquireConsumer(pMnode, pRecoverMsg->consumerId);
L
Liu Jicong 已提交
159 160 161 162
  if (pConsumer == NULL) {
    mError("cannot find consumer %" PRId64 " when processing consumer recover msg", pRecoverMsg->consumerId);
    return -1;
  }
L
Liu Jicong 已提交
163

H
Haojun Liao 已提交
164 165
  mInfo("receive consumer recover msg, consumer:0x%" PRIx64 " status:%d(%s)", pRecoverMsg->consumerId,
        pConsumer->status, mndConsumerStatusName(pConsumer->status));
L
Liu Jicong 已提交
166

L
Liu Jicong 已提交
167
  if (pConsumer->status != MQ_CONSUMER_STATUS__LOST_REBD) {
L
Liu Jicong 已提交
168
    mndReleaseConsumer(pMnode, pConsumer);
L
Liu Jicong 已提交
169
    terrno = TSDB_CODE_MND_CONSUMER_NOT_READY;
L
Liu Jicong 已提交
170 171 172
    return -1;
  }

L
Liu Jicong 已提交
173 174 175 176 177
  SMqConsumerObj *pConsumerNew = tNewSMqConsumerObj(pConsumer->consumerId, pConsumer->cgroup);
  pConsumerNew->updateType = CONSUMER_UPDATE__RECOVER;

  mndReleaseConsumer(pMnode, pConsumer);

178
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_NOTHING, pMsg, "recover-csm");
H
Haojun Liao 已提交
179 180 181 182
  if (pTrans == NULL) {
    goto FAIL;
  }

L
Liu Jicong 已提交
183 184 185
  if (mndSetConsumerCommitLogs(pMnode, pTrans, pConsumerNew) != 0) goto FAIL;
  if (mndTransPrepare(pMnode, pTrans) != 0) goto FAIL;

L
Liu Jicong 已提交
186 187
  tDeleteSMqConsumerObj(pConsumerNew);
  taosMemoryFree(pConsumerNew);
L
Liu Jicong 已提交
188 189 190 191
  mndTransDrop(pTrans);
  return 0;
FAIL:
  tDeleteSMqConsumerObj(pConsumerNew);
L
Liu Jicong 已提交
192
  taosMemoryFree(pConsumerNew);
L
Liu Jicong 已提交
193 194 195 196
  mndTransDrop(pTrans);
  return -1;
}

L
Liu Jicong 已提交
197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226
static int32_t mndProcessConsumerClearMsg(SRpcMsg *pMsg) {
  SMnode              *pMnode = pMsg->info.node;
  SMqConsumerClearMsg *pClearMsg = pMsg->pCont;
  SMqConsumerObj      *pConsumer = mndAcquireConsumer(pMnode, pClearMsg->consumerId);
  if (pConsumer == NULL) {
    return 0;
  }

  mInfo("receive consumer clear msg, consumer id %" PRId64 ", status %s", pClearMsg->consumerId,
        mndConsumerStatusName(pConsumer->status));

  if (pConsumer->status != MQ_CONSUMER_STATUS__LOST_REBD) {
    mndReleaseConsumer(pMnode, pConsumer);
    return -1;
  }

  SMqConsumerObj *pConsumerNew = tNewSMqConsumerObj(pConsumer->consumerId, pConsumer->cgroup);
  pConsumerNew->updateType = CONSUMER_UPDATE__LOST;

  mndReleaseConsumer(pMnode, pConsumer);

  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pMsg, "clear-csm");
  if (pTrans == NULL) goto FAIL;
  if (mndSetConsumerDropLogs(pMnode, pTrans, pConsumerNew) != 0) goto FAIL;
  if (mndTransPrepare(pMnode, pTrans) != 0) goto FAIL;

  tDeleteSMqConsumerObj(pConsumerNew);
  taosMemoryFree(pConsumerNew);
  mndTransDrop(pTrans);
  return 0;
227

L
Liu Jicong 已提交
228 229 230 231 232 233 234
FAIL:
  tDeleteSMqConsumerObj(pConsumerNew);
  taosMemoryFree(pConsumerNew);
  mndTransDrop(pTrans);
  return -1;
}

L
Liu Jicong 已提交
235
static SMqRebInfo *mndGetOrCreateRebSub(SHashObj *pHash, const char *key) {
L
Liu Jicong 已提交
236 237 238 239
  SMqRebInfo *pRebInfo = taosHashGet(pHash, key, strlen(key) + 1);
  if (pRebInfo == NULL) {
    pRebInfo = tNewSMqRebSubscribe(key);
    if (pRebInfo == NULL) {
240 241 242
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      return NULL;
    }
L
Liu Jicong 已提交
243 244 245
    taosHashPut(pHash, key, strlen(key) + 1, pRebInfo, sizeof(SMqRebInfo));
    taosMemoryFree(pRebInfo);
    pRebInfo = taosHashGet(pHash, key, strlen(key) + 1);
246
  }
L
Liu Jicong 已提交
247
  return pRebInfo;
248 249
}

S
Shengliang Guan 已提交
250 251
static int32_t mndProcessMqTimerMsg(SRpcMsg *pMsg) {
  SMnode         *pMnode = pMsg->info.node;
252 253 254 255
  SSdb           *pSdb = pMnode->pSdb;
  SMqConsumerObj *pConsumer;
  void           *pIter = NULL;

256 257
  mTrace("start to process mq timer");

258
  // rebalance cannot be parallel
L
Liu Jicong 已提交
259
  if (!mndRebTryStart()) {
260 261 262 263 264 265 266 267 268 269 270
    mInfo("mq rebalance already in progress, do nothing");
    return 0;
  }

  SMqDoRebalanceMsg *pRebMsg = rpcMallocCont(sizeof(SMqDoRebalanceMsg));
  pRebMsg->rebSubHash = taosHashInit(64, MurmurHash3_32, true, HASH_NO_LOCK);
  // TODO set cleanfp

  // iterate all consumers, find all modification
  while (1) {
    pIter = sdbFetch(pSdb, SDB_CONSUMER, pIter, (void **)&pConsumer);
271 272 273
    if (pIter == NULL) {
      break;
    }
274 275 276

    int32_t hbStatus = atomic_add_fetch_32(&pConsumer->hbStatus, 1);
    int32_t status = atomic_load_32(&pConsumer->status);
H
Haojun Liao 已提交
277

X
Xiaoyu Wang 已提交
278 279 280
    mDebug("check for consumer:0x%" PRIx64 " status:%d(%s), sub-time:%" PRId64 ", uptime:%" PRId64 ", hbstatus:%d",
           pConsumer->consumerId, status, mndConsumerStatusName(status), pConsumer->subscribeTime, pConsumer->upTime,
           hbStatus);
L
Liu Jicong 已提交
281 282

    if (status == MQ_CONSUMER_STATUS__READY) {
H
Haojun Liao 已提交
283 284 285 286 287 288 289 290 291
      if (hbStatus > MND_CONSUMER_LOST_HB_CNT) {
        SMqConsumerLostMsg *pLostMsg = rpcMallocCont(sizeof(SMqConsumerLostMsg));

        pLostMsg->consumerId = pConsumer->consumerId;
        SRpcMsg rpcMsg = {
            .msgType = TDMT_MND_TMQ_CONSUMER_LOST,
            .pCont = pLostMsg,
            .contLen = sizeof(SMqConsumerLostMsg),
        };
292

H
Haojun Liao 已提交
293 294
        tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg);
      }
L
Liu Jicong 已提交
295
    } else if (status == MQ_CONSUMER_STATUS__LOST_REBD) {
H
Haojun Liao 已提交
296
      // if the client is lost longer than one day, clear it. Otherwise, do nothing about the lost consumers.
L
Liu Jicong 已提交
297 298 299 300 301 302 303 304 305
      if (hbStatus > MND_CONSUMER_LOST_CLEAR_THRESHOLD) {
        SMqConsumerClearMsg *pClearMsg = rpcMallocCont(sizeof(SMqConsumerClearMsg));

        pClearMsg->consumerId = pConsumer->consumerId;
        SRpcMsg rpcMsg = {
            .msgType = TDMT_MND_TMQ_LOST_CONSUMER_CLEAR,
            .pCont = pClearMsg,
            .contLen = sizeof(SMqConsumerClearMsg),
        };
306

L
Liu Jicong 已提交
307 308
        tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg);
      }
309
    } else if (status == MQ_CONSUMER_STATUS__LOST) {
L
Liu Jicong 已提交
310
      taosRLockLatch(&pConsumer->lock);
311 312 313 314 315
      int32_t topicNum = taosArrayGetSize(pConsumer->currentTopics);
      for (int32_t i = 0; i < topicNum; i++) {
        char  key[TSDB_SUBSCRIBE_KEY_LEN];
        char *removedTopic = taosArrayGetP(pConsumer->currentTopics, i);
        mndMakeSubscribeKey(key, pConsumer->cgroup, removedTopic);
L
Liu Jicong 已提交
316
        SMqRebInfo *pRebSub = mndGetOrCreateRebSub(pRebMsg->rebSubHash, key);
317 318
        taosArrayPush(pRebSub->removedConsumers, &pConsumer->consumerId);
      }
L
Liu Jicong 已提交
319
      taosRUnLockLatch(&pConsumer->lock);
320
    } else if (status == MQ_CONSUMER_STATUS__MODIFY) {
L
Liu Jicong 已提交
321
      taosRLockLatch(&pConsumer->lock);
322

323 324 325 326 327
      int32_t newTopicNum = taosArrayGetSize(pConsumer->rebNewTopics);
      for (int32_t i = 0; i < newTopicNum; i++) {
        char  key[TSDB_SUBSCRIBE_KEY_LEN];
        char *newTopic = taosArrayGetP(pConsumer->rebNewTopics, i);
        mndMakeSubscribeKey(key, pConsumer->cgroup, newTopic);
L
Liu Jicong 已提交
328
        SMqRebInfo *pRebSub = mndGetOrCreateRebSub(pRebMsg->rebSubHash, key);
329 330 331 332 333 334 335 336
        taosArrayPush(pRebSub->newConsumers, &pConsumer->consumerId);
      }

      int32_t removedTopicNum = taosArrayGetSize(pConsumer->rebRemovedTopics);
      for (int32_t i = 0; i < removedTopicNum; i++) {
        char  key[TSDB_SUBSCRIBE_KEY_LEN];
        char *removedTopic = taosArrayGetP(pConsumer->rebRemovedTopics, i);
        mndMakeSubscribeKey(key, pConsumer->cgroup, removedTopic);
L
Liu Jicong 已提交
337
        SMqRebInfo *pRebSub = mndGetOrCreateRebSub(pRebMsg->rebSubHash, key);
338 339
        taosArrayPush(pRebSub->removedConsumers, &pConsumer->consumerId);
      }
L
Liu Jicong 已提交
340
      taosRUnLockLatch(&pConsumer->lock);
341 342 343 344 345 346 347 348 349 350
    } else {
      // do nothing
    }

    mndReleaseConsumer(pMnode, pConsumer);
  }

  if (taosHashGetSize(pRebMsg->rebSubHash) != 0) {
    mInfo("mq rebalance will be triggered");
    SRpcMsg rpcMsg = {
L
Liu Jicong 已提交
351
        .msgType = TDMT_MND_TMQ_DO_REBALANCE,
352 353 354 355 356 357 358
        .pCont = pRebMsg,
        .contLen = sizeof(SMqDoRebalanceMsg),
    };
    tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg);
  } else {
    taosHashCleanup(pRebMsg->rebSubHash);
    rpcFreeCont(pRebMsg);
H
Haojun Liao 已提交
359
    mInfo("mq rebalance finished, no modification");
L
Liu Jicong 已提交
360
    mndRebEnd();
361 362 363 364
  }
  return 0;
}

365
static int32_t mndProcessMqHbReq(SRpcMsg *pMsg) {
L
Liu Jicong 已提交
366 367
  SMnode  *pMnode = pMsg->info.node;
  SMqHbReq req = {0};
368

D
dapan1121 已提交
369 370 371 372 373
  if (tDeserializeSMqHbReq(pMsg->pCont, pMsg->contLen, &req) < 0) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }

L
Liu Jicong 已提交
374
  int64_t         consumerId = req.consumerId;
375
  SMqConsumerObj *pConsumer = mndAcquireConsumer(pMnode, consumerId);
L
Liu Jicong 已提交
376
  if (pConsumer == NULL) {
X
Xiaoyu Wang 已提交
377
    mError("consumer:0x%" PRIx64 " not exist", consumerId);
L
Liu Jicong 已提交
378 379 380
    terrno = TSDB_CODE_MND_CONSUMER_NOT_EXIST;
    return -1;
  }
381 382 383 384 385 386

  atomic_store_32(&pConsumer->hbStatus, 0);

  int32_t status = atomic_load_32(&pConsumer->status);

  if (status == MQ_CONSUMER_STATUS__LOST_REBD) {
X
Xiaoyu Wang 已提交
387
    mInfo("try to recover consumer:0x%" PRIx64 "", consumerId);
388 389 390
    SMqConsumerRecoverMsg *pRecoverMsg = rpcMallocCont(sizeof(SMqConsumerRecoverMsg));

    pRecoverMsg->consumerId = consumerId;
391
    SRpcMsg pRpcMsg = {
L
Liu Jicong 已提交
392
        .msgType = TDMT_MND_TMQ_CONSUMER_RECOVER,
393 394 395 396
        .pCont = pRecoverMsg,
        .contLen = sizeof(SMqConsumerRecoverMsg),
    };
    tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &pRpcMsg);
397 398 399 400 401 402 403
  }

  mndReleaseConsumer(pMnode, pConsumer);

  return 0;
}

S
Shengliang Guan 已提交
404
static int32_t mndProcessAskEpReq(SRpcMsg *pMsg) {
L
Liu Jicong 已提交
405 406 407
  SMnode     *pMnode = pMsg->info.node;
  SMqAskEpReq req = {0};
  SMqAskEpRsp rsp = {0};
D
dapan1121 已提交
408 409 410 411 412 413

  if (tDeserializeSMqAskEpReq(pMsg->pCont, pMsg->contLen, &req) < 0) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }

L
Liu Jicong 已提交
414 415
  int64_t consumerId = req.consumerId;
  int32_t epoch = req.epoch;
416

417
  SMqConsumerObj *pConsumer = mndAcquireConsumer(pMnode, consumerId);
L
Liu Jicong 已提交
418
  if (pConsumer == NULL) {
H
Haojun Liao 已提交
419
    mError("consumer:0x%" PRIx64 " group:%s not exists in sdb", consumerId, req.cgroup);
420 421 422 423
    terrno = TSDB_CODE_MND_CONSUMER_NOT_EXIST;
    return -1;
  }

H
Haojun Liao 已提交
424 425 426 427 428 429 430
  int32_t ret = strncmp(req.cgroup, pConsumer->cgroup, tListLen(pConsumer->cgroup));
  if (ret != 0) {
    mError("consumer:0x%" PRIx64 " group:%s not consistent with data in sdb, saved cgroup:%s", consumerId, req.cgroup,
           pConsumer->cgroup);
    terrno = TSDB_CODE_MND_CONSUMER_NOT_EXIST;
    return -1;
  }
431

432 433 434 435
  atomic_store_32(&pConsumer->hbStatus, 0);

  // 1. check consumer status
  int32_t status = atomic_load_32(&pConsumer->status);
L
Liu Jicong 已提交
436

L
Liu Jicong 已提交
437
#if 1
L
Liu Jicong 已提交
438
  if (status == MQ_CONSUMER_STATUS__LOST_REBD) {
X
Xiaoyu Wang 已提交
439
    mInfo("try to recover consumer:0x%" PRIx64, consumerId);
L
Liu Jicong 已提交
440 441 442
    SMqConsumerRecoverMsg *pRecoverMsg = rpcMallocCont(sizeof(SMqConsumerRecoverMsg));

    pRecoverMsg->consumerId = consumerId;
443
    SRpcMsg pRpcMsg = {
L
Liu Jicong 已提交
444
        .msgType = TDMT_MND_TMQ_CONSUMER_RECOVER,
445 446 447
        .pCont = pRecoverMsg,
        .contLen = sizeof(SMqConsumerRecoverMsg),
    };
H
Haojun Liao 已提交
448

449
    tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &pRpcMsg);
450
  }
451
#endif
452 453

  if (status != MQ_CONSUMER_STATUS__READY) {
X
Xiaoyu Wang 已提交
454
    mInfo("consumer:0x%" PRIx64 " not ready, status: %s", consumerId, mndConsumerStatusName(status));
455 456 457 458 459 460
    terrno = TSDB_CODE_MND_CONSUMER_NOT_READY;
    return -1;
  }

  int32_t serverEpoch = atomic_load_32(&pConsumer->epoch);

461
  // 2. check epoch, only send ep info when epochs do not match
462 463
  if (epoch != serverEpoch) {
    taosRLockLatch(&pConsumer->lock);
X
Xiaoyu Wang 已提交
464 465
    mInfo("process ask ep, consumer:0x%" PRIx64 "(epoch %d) update with server epoch %d", consumerId, epoch,
          serverEpoch);
466 467 468 469 470
    int32_t numOfTopics = taosArrayGetSize(pConsumer->currentTopics);

    rsp.topics = taosArrayInit(numOfTopics, sizeof(SMqSubTopicEp));
    if (rsp.topics == NULL) {
      terrno = TSDB_CODE_OUT_OF_MEMORY;
L
Liu Jicong 已提交
471
      taosRUnLockLatch(&pConsumer->lock);
472 473 474
      goto FAIL;
    }

H
Haojun Liao 已提交
475
    // handle all topics subscribed by this consumer
476 477 478 479
    for (int32_t i = 0; i < numOfTopics; i++) {
      char            *topic = taosArrayGetP(pConsumer->currentTopics, i);
      SMqSubscribeObj *pSub = mndAcquireSubscribe(pMnode, pConsumer->cgroup, topic);
      // txn guarantees pSub is created
L
Liu Jicong 已提交
480

481 482 483 484 485 486 487 488
      taosRLockLatch(&pSub->lock);

      SMqSubTopicEp topicEp = {0};
      strcpy(topicEp.topic, topic);

      // 2.1 fetch topic schema
      SMqTopicObj *pTopic = mndAcquireTopic(pMnode, topic);
      taosRLockLatch(&pTopic->lock);
L
Liu Jicong 已提交
489
      tstrncpy(topicEp.db, pTopic->db, TSDB_DB_FNAME_LEN);
490
      topicEp.schema.nCols = pTopic->schema.nCols;
L
Liu Jicong 已提交
491 492 493 494
      if (topicEp.schema.nCols) {
        topicEp.schema.pSchema = taosMemoryCalloc(topicEp.schema.nCols, sizeof(SSchema));
        memcpy(topicEp.schema.pSchema, pTopic->schema.pSchema, topicEp.schema.nCols * sizeof(SSchema));
      }
495 496 497 498
      taosRUnLockLatch(&pTopic->lock);
      mndReleaseTopic(pMnode, pTopic);

      // 2.2 iterate all vg assigned to the consumer of that topic
499 500
      SMqConsumerEp *pConsumerEp = taosHashGet(pSub->consumerHash, &consumerId, sizeof(int64_t));
      int32_t        vgNum = taosArrayGetSize(pConsumerEp->vgs);
501

H
Haojun Liao 已提交
502
      // this customer assigned vgroups
503 504 505
      topicEp.vgs = taosArrayInit(vgNum, sizeof(SMqSubVgEp));
      if (topicEp.vgs == NULL) {
        terrno = TSDB_CODE_OUT_OF_MEMORY;
L
Liu Jicong 已提交
506
        taosRUnLockLatch(&pConsumer->lock);
L
Liu Jicong 已提交
507 508
        taosRUnLockLatch(&pSub->lock);
        mndReleaseSubscribe(pMnode, pSub);
509 510 511 512
        goto FAIL;
      }

      for (int32_t j = 0; j < vgNum; j++) {
513
        SMqVgEp *pVgEp = taosArrayGetP(pConsumerEp->vgs, j);
514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531
        char     offsetKey[TSDB_PARTITION_KEY_LEN];
        mndMakePartitionKey(offsetKey, pConsumer->cgroup, topic, pVgEp->vgId);
        // 2.2.1 build vg ep
        SMqSubVgEp vgEp = {
            .epSet = pVgEp->epSet,
            .vgId = pVgEp->vgId,
            .offset = -1,
        };

        taosArrayPush(topicEp.vgs, &vgEp);
      }
      taosArrayPush(rsp.topics, &topicEp);

      taosRUnLockLatch(&pSub->lock);
      mndReleaseSubscribe(pMnode, pSub);
    }
    taosRUnLockLatch(&pConsumer->lock);
  }
H
Haojun Liao 已提交
532

533
  // encode rsp
L
Liu Jicong 已提交
534
  int32_t tlen = sizeof(SMqRspHead) + tEncodeSMqAskEpRsp(NULL, &rsp);
535 536
  void   *buf = rpcMallocCont(tlen);
  if (buf == NULL) {
L
Liu Jicong 已提交
537
    terrno = TSDB_CODE_OUT_OF_MEMORY;
538
    return -1;
L
Liu Jicong 已提交
539
  }
H
Haojun Liao 已提交
540

541 542 543
  ((SMqRspHead *)buf)->mqMsgType = TMQ_MSG_TYPE__EP_RSP;
  ((SMqRspHead *)buf)->epoch = serverEpoch;
  ((SMqRspHead *)buf)->consumerId = pConsumer->consumerId;
S
Shengliang Guan 已提交
544

545
  void *abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));
L
Liu Jicong 已提交
546
  tEncodeSMqAskEpRsp(&abuf, &rsp);
547 548

  // release consumer and free memory
L
Liu Jicong 已提交
549
  tDeleteSMqAskEpRsp(&rsp);
550 551 552
  mndReleaseConsumer(pMnode, pConsumer);

  // send rsp
S
Shengliang Guan 已提交
553 554
  pMsg->info.rsp = buf;
  pMsg->info.rspLen = tlen;
555
  return 0;
H
Haojun Liao 已提交
556

557
FAIL:
L
Liu Jicong 已提交
558
  tDeleteSMqAskEpRsp(&rsp);
559 560 561 562
  mndReleaseConsumer(pMnode, pConsumer);
  return -1;
}

L
Liu Jicong 已提交
563 564 565 566 567 568 569 570
int32_t mndSetConsumerDropLogs(SMnode *pMnode, STrans *pTrans, SMqConsumerObj *pConsumer) {
  SSdbRaw *pCommitRaw = mndConsumerActionEncode(pConsumer);
  if (pCommitRaw == NULL) return -1;
  if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) return -1;
  if (sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED) != 0) return -1;
  return 0;
}

571 572 573 574 575 576 577 578
int32_t mndSetConsumerCommitLogs(SMnode *pMnode, STrans *pTrans, SMqConsumerObj *pConsumer) {
  SSdbRaw *pCommitRaw = mndConsumerActionEncode(pConsumer);
  if (pCommitRaw == NULL) return -1;
  if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) return -1;
  if (sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY) != 0) return -1;
  return 0;
}

579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599
static int32_t validateTopics(const SArray* pTopicList, SMnode* pMnode, const char* pUser) {
  int32_t numOfTopics = taosArrayGetSize(pTopicList);

  for (int32_t i = 0; i < numOfTopics; i++) {
    char        *pOneTopic = taosArrayGetP(pTopicList, i);
    SMqTopicObj *pTopic = mndAcquireTopic(pMnode, pOneTopic);
    if (pTopic == NULL) {  // terrno has been set by callee function
      return -1;
    }

    if (mndCheckTopicPrivilege(pMnode, pUser, MND_OPER_SUBSCRIBE, pTopic) != 0) {
      mndReleaseTopic(pMnode, pTopic);
      return -1;
    }

    mndReleaseTopic(pMnode, pTopic);
  }

  return 0;
}

600 601 602 603
static void* topicNameDup(void* p){
  return taosStrdup((char*) p);
}

604 605 606 607
int32_t mndProcessSubscribeReq(SRpcMsg *pMsg) {
  SMnode *pMnode = pMsg->info.node;
  char   *msgStr = pMsg->pCont;

608 609
  SCMSubscribeReq subscribe = {0};
  tDeserializeSCMSubscribeReq(msgStr, &subscribe);
610 611

  uint64_t        consumerId = subscribe.consumerId;
612
  char           *cgroup = subscribe.cgroup;
H
Haojun Liao 已提交
613
  SMqConsumerObj *pExistedConsumer = NULL;
614 615 616
  SMqConsumerObj *pConsumerNew = NULL;

  int32_t code = -1;
617 618 619
  SArray *pTopicList = subscribe.topicNames;
  taosArraySort(pTopicList, taosArrayCompareString);
  taosArrayRemoveDuplicateP(pTopicList, taosArrayCompareString, taosMemoryFree);
620

621
  int32_t newTopicNum = taosArrayGetSize(pTopicList);
622

H
Haojun Liao 已提交
623
  // check topic existence
624
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_NOTHING, pMsg, "subscribe");
625 626 627
  if (pTrans == NULL) {
    goto _over;
  }
L
Liu Jicong 已提交
628

629 630 631
  code = validateTopics(pTopicList, pMnode, pMsg->info.conn.user);
  if (code != TSDB_CODE_SUCCESS) {
    goto _over;
632 633
  }

H
Haojun Liao 已提交
634 635
  pExistedConsumer = mndAcquireConsumer(pMnode, consumerId);
  if (pExistedConsumer == NULL) {
H
Haojun Liao 已提交
636 637
    mInfo("receive subscribe request from new consumer:0x%" PRIx64" cgroup:%s, numOfTopics:%d", consumerId,
        subscribe.cgroup, (int32_t) taosArrayGetSize(pTopicList));
L
Liu Jicong 已提交
638

639
    pConsumerNew = tNewSMqConsumerObj(consumerId, cgroup);
640
    tstrncpy(pConsumerNew->clientId, subscribe.clientId, tListLen(pConsumerNew->clientId));
641 642

    // set the update type
643
    pConsumerNew->updateType = CONSUMER_UPDATE__MODIFY;
H
Haojun Liao 已提交
644
    taosArrayDestroy(pConsumerNew->assignedTopics);
645 646 647
    pConsumerNew->assignedTopics = taosArrayDup(pTopicList, topicNameDup);

    // all subscribed topics should re-balance.
L
Liu Jicong 已提交
648
    taosArrayDestroy(pConsumerNew->rebNewTopics);
649
    pConsumerNew->rebNewTopics = pTopicList;
650 651
    subscribe.topicNames = NULL;

652 653
    if (mndSetConsumerCommitLogs(pMnode, pTrans, pConsumerNew) != 0) goto _over;
    if (mndTransPrepare(pMnode, pTrans) != 0) goto _over;
654 655

  } else {
H
Haojun Liao 已提交
656 657
    /*taosRLockLatch(&pExistedConsumer->lock);*/
    int32_t status = atomic_load_32(&pExistedConsumer->status);
L
Liu Jicong 已提交
658

H
Haojun Liao 已提交
659 660
    mInfo("receive subscribe request from existed consumer:0x%" PRIx64 " cgroup:%s, current status:%d(%s), subscribe topic num: %d",
          consumerId, subscribe.cgroup, status,mndConsumerStatusName(status), newTopicNum);
L
Liu Jicong 已提交
661

662 663
    if (status != MQ_CONSUMER_STATUS__READY) {
      terrno = TSDB_CODE_MND_CONSUMER_NOT_READY;
664
      goto _over;
665 666 667 668
    }

    pConsumerNew = tNewSMqConsumerObj(consumerId, cgroup);
    if (pConsumerNew == NULL) {
669
      goto _over;
670
    }
H
Haojun Liao 已提交
671

672
    // set the update type
673
    pConsumerNew->updateType = CONSUMER_UPDATE__MODIFY;
H
Haojun Liao 已提交
674
    taosArrayDestroy(pConsumerNew->assignedTopics);
675
    pConsumerNew->assignedTopics = taosArrayDup(pTopicList, topicNameDup);
676

677
    int32_t oldTopicNum = (pExistedConsumer->currentTopics)? taosArrayGetSize(pExistedConsumer->currentTopics):0;
678 679 680 681

    int32_t i = 0, j = 0;
    while (i < oldTopicNum || j < newTopicNum) {
      if (i >= oldTopicNum) {
H
Haojun Liao 已提交
682
        char *newTopicCopy = taosStrdup(taosArrayGetP(pTopicList, j));
683 684 685 686
        taosArrayPush(pConsumerNew->rebNewTopics, &newTopicCopy);
        j++;
        continue;
      } else if (j >= newTopicNum) {
H
Haojun Liao 已提交
687
        char *oldTopicCopy = taosStrdup(taosArrayGetP(pExistedConsumer->currentTopics, i));
688 689 690 691
        taosArrayPush(pConsumerNew->rebRemovedTopics, &oldTopicCopy);
        i++;
        continue;
      } else {
H
Haojun Liao 已提交
692
        char *oldTopic = taosArrayGetP(pExistedConsumer->currentTopics, i);
693
        char *newTopic = taosArrayGetP(pTopicList, j);
694
        int   comp = strcmp(oldTopic, newTopic);
695 696 697 698 699
        if (comp == 0) {
          i++;
          j++;
          continue;
        } else if (comp < 0) {
700
          char *oldTopicCopy = taosStrdup(oldTopic);
701 702 703 704
          taosArrayPush(pConsumerNew->rebRemovedTopics, &oldTopicCopy);
          i++;
          continue;
        } else {
705
          char *newTopicCopy = taosStrdup(newTopic);
706 707 708 709 710 711 712
          taosArrayPush(pConsumerNew->rebNewTopics, &newTopicCopy);
          j++;
          continue;
        }
      }
    }

713 714
    // no topics need to be rebalanced
    if (taosArrayGetSize(pConsumerNew->rebNewTopics) == 0 && taosArrayGetSize(pConsumerNew->rebRemovedTopics) == 0) {
715
//      mInfo();
716
      goto _over;
717 718
    }

719 720
    if (mndSetConsumerCommitLogs(pMnode, pTrans, pConsumerNew) != 0) goto _over;
    if (mndTransPrepare(pMnode, pTrans) != 0) goto _over;
721 722
  }

S
Shengliang Guan 已提交
723
  code = TSDB_CODE_ACTION_IN_PROGRESS;
724

725
_over:
L
Liu Jicong 已提交
726 727
  mndTransDrop(pTrans);

H
Haojun Liao 已提交
728 729 730
  if (pExistedConsumer) {
    /*taosRUnLockLatch(&pExistedConsumer->lock);*/
    mndReleaseConsumer(pMnode, pExistedConsumer);
731
  }
H
Haojun Liao 已提交
732

733 734
  if (pConsumerNew) {
    tDeleteSMqConsumerObj(pConsumerNew);
L
Liu Jicong 已提交
735
    taosMemoryFree(pConsumerNew);
736
  }
737

738
  // TODO: replace with destroy subscribe msg
739
  taosArrayDestroyP(subscribe.topicNames, (FDelete)taosMemoryFree);
740
  return code;
L
Liu Jicong 已提交
741 742
}

L
Liu Jicong 已提交
743 744
SSdbRaw *mndConsumerActionEncode(SMqConsumerObj *pConsumer) {
  terrno = TSDB_CODE_OUT_OF_MEMORY;
S
Shengliang Guan 已提交
745 746

  void   *buf = NULL;
L
Liu Jicong 已提交
747
  int32_t tlen = tEncodeSMqConsumerObj(NULL, pConsumer);
L
Liu Jicong 已提交
748 749 750
  int32_t size = sizeof(int32_t) + tlen + MND_CONSUMER_RESERVE_SIZE;

  SSdbRaw *pRaw = sdbAllocRaw(SDB_CONSUMER, MND_CONSUMER_VER_NUMBER, size);
751
  if (pRaw == NULL) goto CM_ENCODE_OVER;
L
Liu Jicong 已提交
752

wafwerar's avatar
wafwerar 已提交
753
  buf = taosMemoryMalloc(tlen);
L
Liu Jicong 已提交
754 755
  if (buf == NULL) goto CM_ENCODE_OVER;

L
Liu Jicong 已提交
756
  void *abuf = buf;
L
Liu Jicong 已提交
757 758
  tEncodeSMqConsumerObj(&abuf, pConsumer);

L
Liu Jicong 已提交
759
  int32_t dataPos = 0;
L
Liu Jicong 已提交
760 761
  SDB_SET_INT32(pRaw, dataPos, tlen, CM_ENCODE_OVER);
  SDB_SET_BINARY(pRaw, dataPos, buf, tlen, CM_ENCODE_OVER);
L
Liu Jicong 已提交
762 763
  SDB_SET_RESERVE(pRaw, dataPos, MND_CONSUMER_RESERVE_SIZE, CM_ENCODE_OVER);
  SDB_SET_DATALEN(pRaw, dataPos, CM_ENCODE_OVER);
L
Liu Jicong 已提交
764

L
Liu Jicong 已提交
765 766
  terrno = TSDB_CODE_SUCCESS;

767
CM_ENCODE_OVER:
wafwerar's avatar
wafwerar 已提交
768
  taosMemoryFreeClear(buf);
769
  if (terrno != 0) {
770
    mError("consumer:0x%" PRIx64 " failed to encode to raw:%p since %s", pConsumer->consumerId, pRaw, terrstr());
771 772 773 774
    sdbFreeRaw(pRaw);
    return NULL;
  }

775
  mTrace("consumer:0x%" PRIx64 ", encode to raw:%p, row:%p", pConsumer->consumerId, pRaw, pConsumer);
L
Liu Jicong 已提交
776 777 778
  return pRaw;
}

L
Liu Jicong 已提交
779
SSdbRow *mndConsumerActionDecode(SSdbRaw *pRaw) {
780 781 782
  SSdbRow        *pRow = NULL;
  SMqConsumerObj *pConsumer = NULL;
  void           *buf = NULL;
783

L
Liu Jicong 已提交
784
  int8_t sver = 0;
H
Haojun Liao 已提交
785 786 787
  if (sdbGetRawSoftVer(pRaw, &sver) != 0) {
    goto CM_DECODE_OVER;
  }
L
Liu Jicong 已提交
788 789 790

  if (sver != MND_CONSUMER_VER_NUMBER) {
    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
L
Liu Jicong 已提交
791
    goto CM_DECODE_OVER;
L
Liu Jicong 已提交
792 793
  }

794
  pRow = sdbAllocRow(sizeof(SMqConsumerObj));
H
Haojun Liao 已提交
795 796 797
  if (pRow == NULL) {
    goto CM_DECODE_OVER;
  }
798

799
  pConsumer = sdbGetRowObj(pRow);
H
Haojun Liao 已提交
800 801 802
  if (pConsumer == NULL) {
    goto CM_DECODE_OVER;
  }
L
Liu Jicong 已提交
803 804

  int32_t dataPos = 0;
L
Liu Jicong 已提交
805
  int32_t len;
L
Liu Jicong 已提交
806
  SDB_GET_INT32(pRaw, dataPos, &len, CM_DECODE_OVER);
wafwerar's avatar
wafwerar 已提交
807
  buf = taosMemoryMalloc(len);
H
Haojun Liao 已提交
808 809 810 811 812
  if (buf == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    goto CM_DECODE_OVER;
  }

L
Liu Jicong 已提交
813 814
  SDB_GET_BINARY(pRaw, dataPos, buf, len, CM_DECODE_OVER);
  SDB_GET_RESERVE(pRaw, dataPos, MND_CONSUMER_RESERVE_SIZE, CM_DECODE_OVER);
L
Liu Jicong 已提交
815

L
Liu Jicong 已提交
816
  if (tDecodeSMqConsumerObj(buf, pConsumer) == NULL) {
H
Haojun Liao 已提交
817
    terrno = TSDB_CODE_OUT_OF_MEMORY;  // TODO set correct error code
L
Liu Jicong 已提交
818 819
    goto CM_DECODE_OVER;
  }
L
Liu Jicong 已提交
820

H
Haojun Liao 已提交
821
  tmsgUpdateDnodeEpSet(&pConsumer->ep);
L
Liu Jicong 已提交
822

L
Liu Jicong 已提交
823
CM_DECODE_OVER:
wafwerar's avatar
wafwerar 已提交
824
  taosMemoryFreeClear(buf);
L
Liu Jicong 已提交
825
  if (terrno != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
826 827
    mError("consumer:0x%" PRIx64 " failed to decode from raw:%p since %s",
           pConsumer == NULL ? 0 : pConsumer->consumerId, pRaw, terrstr());
wafwerar's avatar
wafwerar 已提交
828
    taosMemoryFreeClear(pRow);
L
Liu Jicong 已提交
829
  }
L
Liu Jicong 已提交
830 831 832 833

  return pRow;
}

L
Liu Jicong 已提交
834
static int32_t mndConsumerActionInsert(SSdb *pSdb, SMqConsumerObj *pConsumer) {
H
Haojun Liao 已提交
835 836 837
  mDebug("consumer:0x%" PRIx64 " cgroup:%s status:%d(%s) epoch:%d load from sdb, perform insert action",
         pConsumer->consumerId, pConsumer->cgroup, pConsumer->status, mndConsumerStatusName(pConsumer->status),
         pConsumer->epoch);
L
Liu Jicong 已提交
838
  pConsumer->subscribeTime = pConsumer->upTime;
L
Liu Jicong 已提交
839 840 841
  return 0;
}

L
Liu Jicong 已提交
842
static int32_t mndConsumerActionDelete(SSdb *pSdb, SMqConsumerObj *pConsumer) {
843 844
  mDebug("consumer:0x%" PRIx64 " perform delete action, status:(%d)%s", pConsumer->consumerId,
         pConsumer->status, mndConsumerStatusName(pConsumer->status));
845
  tDeleteSMqConsumerObj(pConsumer);
L
Liu Jicong 已提交
846 847 848
  return 0;
}

L
Liu Jicong 已提交
849
static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pOldConsumer, SMqConsumerObj *pNewConsumer) {
X
Xiaoyu Wang 已提交
850 851
  mDebug("consumer:0x%" PRIx64 " perform update action, update type:%d, subscribe-time:%" PRId64 ", uptime:%" PRId64,
         pOldConsumer->consumerId, pNewConsumer->updateType, pOldConsumer->subscribeTime, pOldConsumer->upTime);
L
Liu Jicong 已提交
852

853 854 855
  taosWLockLatch(&pOldConsumer->lock);

  if (pNewConsumer->updateType == CONSUMER_UPDATE__MODIFY) {
L
Liu Jicong 已提交
856 857
    /*A(taosArrayGetSize(pOldConsumer->rebNewTopics) == 0);*/
    /*A(taosArrayGetSize(pOldConsumer->rebRemovedTopics) == 0);*/
858

859 860 861 862 863 864 865 866 867 868
    if (taosArrayGetSize(pNewConsumer->rebNewTopics) == 0 && taosArrayGetSize(pNewConsumer->rebRemovedTopics) == 0) {
      pOldConsumer->status = MQ_CONSUMER_STATUS__READY;
    } else {
      SArray *tmp = pOldConsumer->rebNewTopics;
      pOldConsumer->rebNewTopics = pNewConsumer->rebNewTopics;
      pNewConsumer->rebNewTopics = tmp;

      tmp = pOldConsumer->rebRemovedTopics;
      pOldConsumer->rebRemovedTopics = pNewConsumer->rebRemovedTopics;
      pNewConsumer->rebRemovedTopics = tmp;
869

870 871 872
      tmp = pOldConsumer->assignedTopics;
      pOldConsumer->assignedTopics = pNewConsumer->assignedTopics;
      pNewConsumer->assignedTopics = tmp;
L
Liu Jicong 已提交
873

874
      pOldConsumer->subscribeTime = pNewConsumer->upTime;
L
Liu Jicong 已提交
875

876 877
      pOldConsumer->status = MQ_CONSUMER_STATUS__MODIFY;
    }
878
  } else if (pNewConsumer->updateType == CONSUMER_UPDATE__LOST) {
L
Liu Jicong 已提交
879 880
    /*A(taosArrayGetSize(pOldConsumer->rebNewTopics) == 0);*/
    /*A(taosArrayGetSize(pOldConsumer->rebRemovedTopics) == 0);*/
L
Liu Jicong 已提交
881

882
    int32_t sz = taosArrayGetSize(pOldConsumer->currentTopics);
L
Liu Jicong 已提交
883
    /*pOldConsumer->rebRemovedTopics = taosArrayInit(sz, sizeof(void *));*/
884
    for (int32_t i = 0; i < sz; i++) {
885
      char *topic = taosStrdup(taosArrayGetP(pOldConsumer->currentTopics, i));
L
Liu Jicong 已提交
886
      taosArrayPush(pOldConsumer->rebRemovedTopics, &topic);
887
    }
L
Liu Jicong 已提交
888 889 890

    pOldConsumer->rebalanceTime = pNewConsumer->upTime;

H
Haojun Liao 已提交
891
    int32_t status = pOldConsumer->status;
892
    pOldConsumer->status = MQ_CONSUMER_STATUS__LOST;
H
Haojun Liao 已提交
893 894 895
    mDebug("consumer:0x%" PRIx64 " state %s -> %s, reb-time:%" PRId64 ", reb-removed-topics:%d",
           pOldConsumer->consumerId, mndConsumerStatusName(status), mndConsumerStatusName(pOldConsumer->status),
           pOldConsumer->rebalanceTime, (int)taosArrayGetSize(pOldConsumer->rebRemovedTopics));
L
Liu Jicong 已提交
896
  } else if (pNewConsumer->updateType == CONSUMER_UPDATE__RECOVER) {
L
Liu Jicong 已提交
897 898
    /*A(taosArrayGetSize(pOldConsumer->currentTopics) == 0);*/
    /*A(taosArrayGetSize(pOldConsumer->rebNewTopics) == 0);*/
L
Liu Jicong 已提交
899 900 901

    int32_t sz = taosArrayGetSize(pOldConsumer->assignedTopics);
    for (int32_t i = 0; i < sz; i++) {
902
      char *topic = taosStrdup(taosArrayGetP(pOldConsumer->assignedTopics, i));
L
Liu Jicong 已提交
903 904 905 906 907 908
      taosArrayPush(pOldConsumer->rebNewTopics, &topic);
    }

    pOldConsumer->rebalanceTime = pNewConsumer->upTime;

    pOldConsumer->status = MQ_CONSUMER_STATUS__MODIFY;
909 910
  } else if (pNewConsumer->updateType == CONSUMER_UPDATE__TOUCH) {
    atomic_add_fetch_32(&pOldConsumer->epoch, 1);
L
Liu Jicong 已提交
911 912 913

    pOldConsumer->rebalanceTime = pNewConsumer->upTime;

914
  } else if (pNewConsumer->updateType == CONSUMER_UPDATE__ADD) {
L
Liu Jicong 已提交
915 916
    /*A(taosArrayGetSize(pNewConsumer->rebNewTopics) == 1);*/
    /*A(taosArrayGetSize(pNewConsumer->rebRemovedTopics) == 0);*/
917

918
    char *addedTopic = taosStrdup(taosArrayGetP(pNewConsumer->rebNewTopics, 0));
919
    // not exist in current topic
X
Xiaoyu Wang 已提交
920

L
Liu Jicong 已提交
921
    bool existing = false;
922
#if 1
H
Haojun Liao 已提交
923 924
    int32_t numOfExistedTopics = taosArrayGetSize(pOldConsumer->currentTopics);
    for (int32_t i = 0; i < numOfExistedTopics; i++) {
925
      char *topic = taosArrayGetP(pOldConsumer->currentTopics, i);
L
Liu Jicong 已提交
926 927 928
      if (strcmp(topic, addedTopic) == 0) {
        existing = true;
      }
929 930 931 932 933 934 935 936 937 938 939 940 941 942
    }
#endif

    // remove from new topic
    for (int32_t i = 0; i < taosArrayGetSize(pOldConsumer->rebNewTopics); i++) {
      char *topic = taosArrayGetP(pOldConsumer->rebNewTopics, i);
      if (strcmp(addedTopic, topic) == 0) {
        taosArrayRemove(pOldConsumer->rebNewTopics, i);
        taosMemoryFree(topic);
        break;
      }
    }

    // add to current topic
L
Liu Jicong 已提交
943 944 945
    if (!existing) {
      taosArrayPush(pOldConsumer->currentTopics, &addedTopic);
      taosArraySort(pOldConsumer->currentTopics, taosArrayCompareString);
M
Minglei Jin 已提交
946 947
    } else {
      taosMemoryFree(addedTopic);
L
Liu Jicong 已提交
948
    }
H
Haojun Liao 已提交
949

950
    // set status
H
Haojun Liao 已提交
951
    int32_t status = pOldConsumer->status;
952
    if (taosArrayGetSize(pOldConsumer->rebNewTopics) == 0 && taosArrayGetSize(pOldConsumer->rebRemovedTopics) == 0) {
H
Haojun Liao 已提交
953
      if (status == MQ_CONSUMER_STATUS__MODIFY || status == MQ_CONSUMER_STATUS__MODIFY_IN_REB) {
954
        pOldConsumer->status = MQ_CONSUMER_STATUS__READY;
H
Haojun Liao 已提交
955
      } else if (status == MQ_CONSUMER_STATUS__LOST_IN_REB || status == MQ_CONSUMER_STATUS__LOST) {
956 957 958
        pOldConsumer->status = MQ_CONSUMER_STATUS__LOST_REBD;
      }
    } else {
H
Haojun Liao 已提交
959
      if (status == MQ_CONSUMER_STATUS__MODIFY || status == MQ_CONSUMER_STATUS__MODIFY_IN_REB) {
960
        pOldConsumer->status = MQ_CONSUMER_STATUS__MODIFY_IN_REB;
H
Haojun Liao 已提交
961
      } else if (status == MQ_CONSUMER_STATUS__LOST || status == MQ_CONSUMER_STATUS__LOST_IN_REB) {
962 963 964
        pOldConsumer->status = MQ_CONSUMER_STATUS__LOST_IN_REB;
      }
    }
L
Liu Jicong 已提交
965

H
Haojun Liao 已提交
966
    // the re-balance is triggered when the new consumer is launched.
L
Liu Jicong 已提交
967 968
    pOldConsumer->rebalanceTime = pNewConsumer->upTime;

969
    atomic_add_fetch_32(&pOldConsumer->epoch, 1);
H
Haojun Liao 已提交
970 971 972
    mDebug("consumer:0x%" PRIx64 " state (%d)%s -> (%d)%s, new epoch:%d, reb-time:%" PRId64 ", current topics:%d",
           pOldConsumer->consumerId, status, mndConsumerStatusName(status), pOldConsumer->status,
           mndConsumerStatusName(pOldConsumer->status),
H
Haojun Liao 已提交
973
           pOldConsumer->epoch, pOldConsumer->rebalanceTime, (int)taosArrayGetSize(pOldConsumer->currentTopics));
974
  } else if (pNewConsumer->updateType == CONSUMER_UPDATE__REMOVE) {
L
Liu Jicong 已提交
975 976
    /*A(taosArrayGetSize(pNewConsumer->rebNewTopics) == 0);*/
    /*A(taosArrayGetSize(pNewConsumer->rebRemovedTopics) == 1);*/
977 978 979
    char *removedTopic = taosArrayGetP(pNewConsumer->rebRemovedTopics, 0);

    // not exist in new topic
L
Liu Jicong 已提交
980
#if 0
981 982
    for (int32_t i = 0; i < taosArrayGetSize(pOldConsumer->rebNewTopics); i++) {
      char *topic = taosArrayGetP(pOldConsumer->rebNewTopics, i);
L
Liu Jicong 已提交
983
      A(strcmp(topic, removedTopic) != 0);
984 985 986 987 988
    }
#endif

    // remove from removed topic
    for (int32_t i = 0; i < taosArrayGetSize(pOldConsumer->rebRemovedTopics); i++) {
L
fix  
Liu Jicong 已提交
989
      char *topic = taosArrayGetP(pOldConsumer->rebRemovedTopics, i);
990
      if (strcmp(removedTopic, topic) == 0) {
L
fix  
Liu Jicong 已提交
991
        taosArrayRemove(pOldConsumer->rebRemovedTopics, i);
992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008
        taosMemoryFree(topic);
        break;
      }
    }

    // remove from current topic
    int32_t i = 0;
    int32_t sz = taosArrayGetSize(pOldConsumer->currentTopics);
    for (i = 0; i < sz; i++) {
      char *topic = taosArrayGetP(pOldConsumer->currentTopics, i);
      if (strcmp(removedTopic, topic) == 0) {
        taosArrayRemove(pOldConsumer->currentTopics, i);
        taosMemoryFree(topic);
        break;
      }
    }
    // must find the topic
L
Liu Jicong 已提交
1009
    /*A(i < sz);*/
1010 1011

    // set status
H
Haojun Liao 已提交
1012
    int32_t status = pOldConsumer->status;
1013
    if (taosArrayGetSize(pOldConsumer->rebNewTopics) == 0 && taosArrayGetSize(pOldConsumer->rebRemovedTopics) == 0) {
H
Haojun Liao 已提交
1014
      if (status == MQ_CONSUMER_STATUS__MODIFY || status == MQ_CONSUMER_STATUS__MODIFY_IN_REB) {
1015
        pOldConsumer->status = MQ_CONSUMER_STATUS__READY;
H
Haojun Liao 已提交
1016
      } else if (status == MQ_CONSUMER_STATUS__LOST_IN_REB || status == MQ_CONSUMER_STATUS__LOST) {
1017 1018 1019
        pOldConsumer->status = MQ_CONSUMER_STATUS__LOST_REBD;
      }
    } else {
H
Haojun Liao 已提交
1020
      if (status == MQ_CONSUMER_STATUS__MODIFY || status == MQ_CONSUMER_STATUS__MODIFY_IN_REB) {
1021
        pOldConsumer->status = MQ_CONSUMER_STATUS__MODIFY_IN_REB;
H
Haojun Liao 已提交
1022
      } else if (status == MQ_CONSUMER_STATUS__LOST || status == MQ_CONSUMER_STATUS__LOST_IN_REB) {
1023 1024 1025
        pOldConsumer->status = MQ_CONSUMER_STATUS__LOST_IN_REB;
      }
    }
L
Liu Jicong 已提交
1026 1027

    pOldConsumer->rebalanceTime = pNewConsumer->upTime;
1028
    atomic_add_fetch_32(&pOldConsumer->epoch, 1);
H
Haojun Liao 已提交
1029

H
Haojun Liao 已提交
1030 1031 1032
    mDebug("consumer:0x%" PRIx64 " state %d(%s) -> %d(%s), new epoch:%d, reb-time:%" PRId64 ", current topics:%d",
           pOldConsumer->consumerId, status, mndConsumerStatusName(status), pOldConsumer->status,
           mndConsumerStatusName(pOldConsumer->status),
H
Haojun Liao 已提交
1033
           pOldConsumer->epoch, pOldConsumer->rebalanceTime, (int)taosArrayGetSize(pOldConsumer->currentTopics));
1034 1035 1036
  }

  taosWUnLockLatch(&pOldConsumer->lock);
L
Liu Jicong 已提交
1037 1038 1039
  return 0;
}

L
Liu Jicong 已提交
1040
SMqConsumerObj *mndAcquireConsumer(SMnode *pMnode, int64_t consumerId) {
L
Liu Jicong 已提交
1041 1042
  SSdb           *pSdb = pMnode->pSdb;
  SMqConsumerObj *pConsumer = sdbAcquire(pSdb, SDB_CONSUMER, &consumerId);
L
Liu Jicong 已提交
1043
  if (pConsumer == NULL) {
1044
    terrno = TSDB_CODE_MND_CONSUMER_NOT_EXIST;
L
Liu Jicong 已提交
1045 1046 1047 1048
  }
  return pConsumer;
}

L
Liu Jicong 已提交
1049
void mndReleaseConsumer(SMnode *pMnode, SMqConsumerObj *pConsumer) {
L
Liu Jicong 已提交
1050 1051 1052
  SSdb *pSdb = pMnode->pSdb;
  sdbRelease(pSdb, pConsumer);
}
L
Liu Jicong 已提交
1053

S
Shengliang Guan 已提交
1054 1055
static int32_t mndRetrieveConsumer(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rowsCapacity) {
  SMnode         *pMnode = pReq->info.node;
L
Liu Jicong 已提交
1056 1057 1058 1059 1060 1061
  SSdb           *pSdb = pMnode->pSdb;
  int32_t         numOfRows = 0;
  SMqConsumerObj *pConsumer = NULL;

  while (numOfRows < rowsCapacity) {
    pShow->pIter = sdbFetch(pSdb, SDB_CONSUMER, pShow->pIter, (void **)&pConsumer);
1062 1063 1064 1065
    if (pShow->pIter == NULL) {
      break;
    }

L
Liu Jicong 已提交
1066
    if (taosArrayGetSize(pConsumer->assignedTopics) == 0) {
X
Xiaoyu Wang 已提交
1067
      mDebug("showing consumer:0x%" PRIx64 " no assigned topic, skip", pConsumer->consumerId);
L
Liu Jicong 已提交
1068 1069 1070
      sdbRelease(pSdb, pConsumer);
      continue;
    }
L
Liu Jicong 已提交
1071 1072

    taosRLockLatch(&pConsumer->lock);
X
Xiaoyu Wang 已提交
1073
    mDebug("showing consumer:0x%" PRIx64, pConsumer->consumerId);
L
Liu Jicong 已提交
1074

L
Liu Jicong 已提交
1075 1076 1077 1078 1079 1080 1081
    int32_t topicSz = taosArrayGetSize(pConsumer->assignedTopics);
    bool    hasTopic = true;
    if (topicSz == 0) {
      hasTopic = false;
      topicSz = 1;
    }

L
Liu Jicong 已提交
1082 1083 1084 1085
    if (numOfRows + topicSz > rowsCapacity) {
      blockDataEnsureCapacity(pBlock, numOfRows + topicSz);
    }

L
Liu Jicong 已提交
1086 1087 1088 1089 1090 1091
    for (int32_t i = 0; i < topicSz; i++) {
      SColumnInfoData *pColInfo;
      int32_t          cols = 0;

      // consumer id
      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
1092
      colDataSetVal(pColInfo, numOfRows, (const char *)&pConsumer->consumerId, false);
L
Liu Jicong 已提交
1093

L
Liu Jicong 已提交
1094 1095
      // consumer group
      char cgroup[TSDB_CGROUP_LEN + VARSTR_HEADER_SIZE] = {0};
1096 1097
      STR_TO_VARSTR(cgroup, pConsumer->cgroup);

L
Liu Jicong 已提交
1098
      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
1099
      colDataSetVal(pColInfo, numOfRows, (const char *)cgroup, false);
L
Liu Jicong 已提交
1100

1101
      // client id
L
Liu Jicong 已提交
1102
      char clientId[256 + VARSTR_HEADER_SIZE] = {0};
1103 1104
      STR_TO_VARSTR(clientId, pConsumer->clientId);

L
Liu Jicong 已提交
1105
      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
1106
      colDataSetVal(pColInfo, numOfRows, (const char *)clientId, false);
L
Liu Jicong 已提交
1107 1108 1109

      // status
      char status[20 + VARSTR_HEADER_SIZE] = {0};
1110 1111 1112
      const char* pStatusName = mndConsumerStatusName(pConsumer->status);
      STR_TO_VARSTR(status, pStatusName);

L
Liu Jicong 已提交
1113
      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
1114
      colDataSetVal(pColInfo, numOfRows, (const char *)status, false);
L
Liu Jicong 已提交
1115 1116 1117 1118 1119 1120

      // one subscribed topic
      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
      if (hasTopic) {
        char        topic[TSDB_TOPIC_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
        const char *topicName = mndTopicGetShowName(taosArrayGetP(pConsumer->assignedTopics, i));
H
Haojun Liao 已提交
1121
        STR_TO_VARSTR(topic, topicName);
1122
        colDataSetVal(pColInfo, numOfRows, (const char *)topic, false);
L
Liu Jicong 已提交
1123
      } else {
1124
        colDataSetVal(pColInfo, numOfRows, NULL, true);
L
Liu Jicong 已提交
1125 1126 1127
      }

      // end point
L
Liu Jicong 已提交
1128
      /*pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);*/
1129
      /*colDataSetVal(pColInfo, numOfRows, (const char *)&pConsumer->ep, true);*/
L
Liu Jicong 已提交
1130 1131 1132

      // up time
      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
1133
      colDataSetVal(pColInfo, numOfRows, (const char *)&pConsumer->upTime, false);
L
Liu Jicong 已提交
1134 1135 1136

      // subscribe time
      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
1137
      colDataSetVal(pColInfo, numOfRows, (const char *)&pConsumer->subscribeTime, false);
L
Liu Jicong 已提交
1138 1139 1140

      // rebalance time
      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
1141
      colDataSetVal(pColInfo, numOfRows, (const char *)&pConsumer->rebalanceTime, pConsumer->rebalanceTime == 0);
L
Liu Jicong 已提交
1142 1143 1144

      numOfRows++;
    }
1145

L
Liu Jicong 已提交
1146 1147
    taosRUnLockLatch(&pConsumer->lock);
    sdbRelease(pSdb, pConsumer);
1148 1149

    pBlock->info.rows = numOfRows;
L
Liu Jicong 已提交
1150 1151 1152 1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175
  }

  pShow->numOfRows += numOfRows;
  return numOfRows;
}

static void mndCancelGetNextConsumer(SMnode *pMnode, void *pIter) {
  SSdb *pSdb = pMnode->pSdb;
  sdbCancelFetch(pSdb, pIter);
}

static const char *mndConsumerStatusName(int status) {
  switch (status) {
    case MQ_CONSUMER_STATUS__READY:
      return "ready";
    case MQ_CONSUMER_STATUS__LOST:
    case MQ_CONSUMER_STATUS__LOST_REBD:
    case MQ_CONSUMER_STATUS__LOST_IN_REB:
      return "lost";
    case MQ_CONSUMER_STATUS__MODIFY:
    case MQ_CONSUMER_STATUS__MODIFY_IN_REB:
      return "rebalancing";
    default:
      return "unknown";
  }
}