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

S
Shengliang Guan 已提交
16
#define _DEFAULT_SOURCE
L
Liu Jicong 已提交
17
#include "mndSubscribe.h"
L
Liu Jicong 已提交
18 19 20 21
#include "mndConsumer.h"
#include "mndDb.h"
#include "mndDnode.h"
#include "mndMnode.h"
L
Liu Jicong 已提交
22
#include "mndScheduler.h"
L
Liu Jicong 已提交
23 24 25 26 27 28 29 30 31
#include "mndShow.h"
#include "mndStb.h"
#include "mndTopic.h"
#include "mndTrans.h"
#include "mndUser.h"
#include "mndVgroup.h"
#include "tcompare.h"
#include "tname.h"

L
Liu Jicong 已提交
32
#define MND_SUBSCRIBE_VER_NUMBER   1
L
Liu Jicong 已提交
33 34
#define MND_SUBSCRIBE_RESERVE_SIZE 64

L
Liu Jicong 已提交
35
#define MND_SUBSCRIBE_REBALANCE_CNT 3
L
Liu Jicong 已提交
36

L
Liu Jicong 已提交
37 38 39 40 41 42
static SSdbRaw *mndSubActionEncode(SMqSubscribeObj *);
static SSdbRow *mndSubActionDecode(SSdbRaw *pRaw);
static int32_t  mndSubActionInsert(SSdb *pSdb, SMqSubscribeObj *);
static int32_t  mndSubActionDelete(SSdb *pSdb, SMqSubscribeObj *);
static int32_t  mndSubActionUpdate(SSdb *pSdb, SMqSubscribeObj *pOldSub, SMqSubscribeObj *pNewSub);

S
Shengliang Guan 已提交
43
static int32_t mndProcessRebalanceReq(SRpcMsg *pMsg);
L
Liu Jicong 已提交
44
static int32_t mndProcessDropCgroupReq(SRpcMsg *pMsg);
45

S
Shengliang Guan 已提交
46
static int32_t mndRetrieveSubscribe(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
L
Liu Jicong 已提交
47 48
static void    mndCancelGetNextSubscribe(SMnode *pMnode, void *pIter);

49 50 51 52 53 54 55
static int32_t mndSetSubRedoLogs(SMnode *pMnode, STrans *pTrans, SMqSubscribeObj *pSub) {
  SSdbRaw *pRedoRaw = mndSubActionEncode(pSub);
  if (pRedoRaw == NULL) return -1;
  if (mndTransAppendRedolog(pTrans, pRedoRaw) != 0) return -1;
  if (sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY) != 0) return -1;
  return 0;
}
L
Liu Jicong 已提交
56

57 58 59 60 61 62 63
static int32_t mndSetSubCommitLogs(SMnode *pMnode, STrans *pTrans, SMqSubscribeObj *pSub) {
  SSdbRaw *pCommitRaw = mndSubActionEncode(pSub);
  if (pCommitRaw == NULL) return -1;
  if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) return -1;
  if (sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY) != 0) return -1;
  return 0;
}
L
Liu Jicong 已提交
64

L
Liu Jicong 已提交
65
int32_t mndInitSubscribe(SMnode *pMnode) {
66 67 68 69 70 71 72 73 74 75
  SSdbTable table = {
      .sdbType = SDB_SUBSCRIBE,
      .keyType = SDB_KEY_BINARY,
      .encodeFp = (SdbEncodeFp)mndSubActionEncode,
      .decodeFp = (SdbDecodeFp)mndSubActionDecode,
      .insertFp = (SdbInsertFp)mndSubActionInsert,
      .updateFp = (SdbUpdateFp)mndSubActionUpdate,
      .deleteFp = (SdbDeleteFp)mndSubActionDelete,
  };

L
Liu Jicong 已提交
76 77 78 79 80
  mndSetMsgHandle(pMnode, TDMT_VND_TMQ_SUBSCRIBE_RSP, mndTransProcessRsp);
  mndSetMsgHandle(pMnode, TDMT_VND_TMQ_DELETE_SUB_RSP, mndTransProcessRsp);
  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_DO_REBALANCE, mndProcessRebalanceReq);
  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_DROP_CGROUP, mndProcessDropCgroupReq);
  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_DROP_CGROUP_RSP, mndTransProcessRsp);
L
Liu Jicong 已提交
81 82 83 84

  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_SUBSCRIPTIONS, mndRetrieveSubscribe);
  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_TOPICS, mndCancelGetNextSubscribe);

L
Liu Jicong 已提交
85 86 87
  return sdbSetTable(pMnode->pSdb, table);
}

88 89 90 91 92 93
static SMqSubscribeObj *mndCreateSub(SMnode *pMnode, const SMqTopicObj *pTopic, const char *subKey) {
  SMqSubscribeObj *pSub = tNewSubscribeObj(subKey);
  if (pSub == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }
L
Liu Jicong 已提交
94
  pSub->dbUid = pTopic->dbUid;
L
Liu Jicong 已提交
95
  pSub->stbUid = pTopic->stbUid;
L
Liu Jicong 已提交
96
  pSub->subType = pTopic->subType;
L
Liu Jicong 已提交
97
  pSub->withMeta = pTopic->withMeta;
L
Liu Jicong 已提交
98

99 100
  ASSERT(pSub->unassignedVgs->size == 0);
  ASSERT(taosHashGetSize(pSub->consumerHash) == 0);
101 102 103 104 105 106 107

  if (mndSchedInitSubEp(pMnode, pTopic, pSub) < 0) {
    tDeleteSubscribeObj(pSub);
    taosMemoryFree(pSub);
    return NULL;
  }

108 109
  ASSERT(pSub->unassignedVgs->size > 0);
  ASSERT(taosHashGetSize(pSub->consumerHash) == 0);
L
Liu Jicong 已提交
110

111 112 113
  return pSub;
}

L
Liu Jicong 已提交
114 115
static int32_t mndBuildSubChangeReq(void **pBuf, int32_t *pLen, const SMqSubscribeObj *pSub,
                                    const SMqRebOutputVg *pRebVg) {
116 117 118 119
  SMqRebVgReq req = {0};
  req.oldConsumerId = pRebVg->oldConsumerId;
  req.newConsumerId = pRebVg->newConsumerId;
  req.vgId = pRebVg->pVgEp->vgId;
L
Liu Jicong 已提交
120
  req.qmsg = pRebVg->pVgEp->qmsg;
L
Liu Jicong 已提交
121
  req.subType = pSub->subType;
L
Liu Jicong 已提交
122
  req.withMeta = pSub->withMeta;
L
Liu Jicong 已提交
123
  req.suid = pSub->stbUid;
S
Shengliang Guan 已提交
124
  tstrncpy(req.subKey, pSub->key, TSDB_SUBSCRIBE_KEY_LEN);
125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144

  int32_t tlen = sizeof(SMsgHead) + tEncodeSMqRebVgReq(NULL, &req);
  void   *buf = taosMemoryMalloc(tlen);
  if (buf == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }
  SMsgHead *pMsgHead = (SMsgHead *)buf;

  pMsgHead->contLen = htonl(tlen);
  pMsgHead->vgId = htonl(pRebVg->pVgEp->vgId);

  void *abuf = POINTER_SHIFT(buf, sizeof(SMsgHead));
  tEncodeSMqRebVgReq(&abuf, &req);
  *pBuf = buf;
  *pLen = tlen;

  return 0;
}

L
Liu Jicong 已提交
145
static int32_t mndPersistSubChangeVgReq(SMnode *pMnode, STrans *pTrans, const SMqSubscribeObj *pSub,
146 147 148 149 150
                                        const SMqRebOutputVg *pRebVg) {
  ASSERT(pRebVg->oldConsumerId != pRebVg->newConsumerId);

  void   *buf;
  int32_t tlen;
L
Liu Jicong 已提交
151
  if (mndBuildSubChangeReq(&buf, &tlen, pSub, pRebVg) < 0) {
152 153 154 155 156
    return -1;
  }

  int32_t vgId = pRebVg->pVgEp->vgId;
  SVgObj *pVgObj = mndAcquireVgroup(pMnode, vgId);
L
Liu Jicong 已提交
157
  if (pVgObj == NULL) {
L
Liu Jicong 已提交
158
    ASSERT(0);
L
Liu Jicong 已提交
159 160 161
    taosMemoryFree(buf);
    return -1;
  }
162 163 164 165 166

  STransAction action = {0};
  action.epSet = mndGetVgroupEpset(pMnode, pVgObj);
  action.pCont = buf;
  action.contLen = tlen;
L
Liu Jicong 已提交
167
  action.msgType = TDMT_VND_TMQ_SUBSCRIBE;
168 169 170 171 172 173 174 175

  mndReleaseVgroup(pMnode, pVgObj);
  if (mndTransAppendRedoAction(pTrans, &action) != 0) {
    taosMemoryFree(buf);
    return -1;
  }
  return 0;
}
L
Liu Jicong 已提交
176

L
Liu Jicong 已提交
177
static int32_t mndSplitSubscribeKey(const char *key, char *topic, char *cgroup, bool fullName) {
S
Shengliang Guan 已提交
178
  int32_t i = 0;
L
Liu Jicong 已提交
179
  while (key[i] != TMQ_SEPARATOR) {
L
Liu Jicong 已提交
180 181
    i++;
  }
L
Liu Jicong 已提交
182 183
  memcpy(cgroup, key, i);
  cgroup[i] = 0;
L
Liu Jicong 已提交
184 185 186 187 188 189 190 191
  if (fullName) {
    strcpy(topic, &key[i + 1]);
  } else {
    while (key[i] != '.') {
      i++;
    }
    strcpy(topic, &key[i + 1]);
  }
L
Liu Jicong 已提交
192 193 194
  return 0;
}

L
Liu Jicong 已提交
195 196
static SMqRebInfo *mndGetOrCreateRebSub(SHashObj *pHash, const char *key) {
  SMqRebInfo *pRebSub = taosHashGet(pHash, key, strlen(key) + 1);
L
Liu Jicong 已提交
197 198 199
  if (pRebSub == NULL) {
    pRebSub = tNewSMqRebSubscribe(key);
    if (pRebSub == NULL) {
L
Liu Jicong 已提交
200
      terrno = TSDB_CODE_OUT_OF_MEMORY;
L
Liu Jicong 已提交
201 202
      return NULL;
    }
L
Liu Jicong 已提交
203
    taosHashPut(pHash, key, strlen(key) + 1, pRebSub, sizeof(SMqRebInfo));
L
Liu Jicong 已提交
204 205 206 207
  }
  return pRebSub;
}

208 209 210
static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqRebOutputObj *pOutput) {
  int32_t totalVgNum = pOutput->pSub->vgNum;

211
  mInfo("mq rebalance: subscription: %s, vgNum: %d", pOutput->pSub->key, pOutput->pSub->vgNum);
L
Liu Jicong 已提交
212

213 214 215 216 217 218 219 220 221
  // 1. build temporary hash(vgId -> SMqRebOutputVg) to store modified vg
  SHashObj *pHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_NO_LOCK);

  // 2. check and get actual removed consumers, put their vg into hash
  int32_t removedNum = taosArrayGetSize(pInput->pRebInfo->removedConsumers);
  int32_t actualRemoved = 0;
  for (int32_t i = 0; i < removedNum; i++) {
    int64_t consumerId = *(int64_t *)taosArrayGet(pInput->pRebInfo->removedConsumers, i);
    ASSERT(consumerId > 0);
222 223 224 225
    SMqConsumerEp *pConsumerEp = taosHashGet(pOutput->pSub->consumerHash, &consumerId, sizeof(int64_t));
    ASSERT(pConsumerEp);
    if (pConsumerEp) {
      ASSERT(consumerId == pConsumerEp->consumerId);
226
      actualRemoved++;
227
      int32_t consumerVgNum = taosArrayGetSize(pConsumerEp->vgs);
228
      for (int32_t j = 0; j < consumerVgNum; j++) {
229
        SMqVgEp       *pVgEp = taosArrayGetP(pConsumerEp->vgs, j);
230 231 232 233 234 235
        SMqRebOutputVg outputVg = {
            .oldConsumerId = consumerId,
            .newConsumerId = -1,
            .pVgEp = pVgEp,
        };
        taosHashPut(pHash, &pVgEp->vgId, sizeof(int32_t), &outputVg, sizeof(SMqRebOutputVg));
S
Shengliang Guan 已提交
236
        mInfo("mq rebalance: remove vgId:%d from consumer:%" PRId64, pVgEp->vgId, consumerId);
237 238 239 240 241 242 243 244 245 246
      }
      taosHashRemove(pOutput->pSub->consumerHash, &consumerId, sizeof(int64_t));
      // put into removed
      taosArrayPush(pOutput->removedConsumers, &consumerId);
    }
  }
  ASSERT(removedNum == actualRemoved);

  // if previously no consumer, there are vgs not assigned
  {
247
    int32_t consumerVgNum = taosArrayGetSize(pOutput->pSub->unassignedVgs);
248
    for (int32_t i = 0; i < consumerVgNum; i++) {
249
      SMqVgEp       *pVgEp = *(SMqVgEp **)taosArrayPop(pOutput->pSub->unassignedVgs);
250 251 252 253 254 255
      SMqRebOutputVg rebOutput = {
          .oldConsumerId = -1,
          .newConsumerId = -1,
          .pVgEp = pVgEp,
      };
      taosHashPut(pHash, &pVgEp->vgId, sizeof(int32_t), &rebOutput, sizeof(SMqRebOutputVg));
S
Shengliang Guan 已提交
256
      mInfo("mq rebalance: remove vgId:%d from unassigned", pVgEp->vgId);
257 258 259 260
    }
  }

  // 3. calc vg number of each consumer
L
Liu Jicong 已提交
261 262
  int32_t afterRebConsumerNum = pInput->oldConsumerNum + taosArrayGetSize(pInput->pRebInfo->newConsumers) -
                                taosArrayGetSize(pInput->pRebInfo->removedConsumers);
L
Liu Jicong 已提交
263 264
  int32_t minVgCnt = 0;
  int32_t imbConsumerNum = 0;
265
  // calc num
L
Liu Jicong 已提交
266 267 268 269
  if (afterRebConsumerNum) {
    minVgCnt = totalVgNum / afterRebConsumerNum;
    imbConsumerNum = totalVgNum % afterRebConsumerNum;
  }
270 271
  mInfo("mq rebalance: %d consumer after rebalance, at least %d vg each, %d consumer has more vg", afterRebConsumerNum,
        minVgCnt, imbConsumerNum);
272 273 274 275 276 277 278

  // 4. first scan: remove consumer more than wanted, put to remove hash
  int32_t imbCnt = 0;
  void   *pIter = NULL;
  while (1) {
    pIter = taosHashIterate(pOutput->pSub->consumerHash, pIter);
    if (pIter == NULL) break;
279 280 281
    SMqConsumerEp *pConsumerEp = (SMqConsumerEp *)pIter;
    ASSERT(pConsumerEp->consumerId > 0);
    int32_t consumerVgNum = taosArrayGetSize(pConsumerEp->vgs);
282 283
    // all old consumers still existing are touched
    // TODO optimize: touch only consumer whose vgs changed
284
    taosArrayPush(pOutput->touchedConsumers, &pConsumerEp->consumerId);
285 286 287
    if (consumerVgNum > minVgCnt) {
      if (imbCnt < imbConsumerNum) {
        if (consumerVgNum == minVgCnt + 1) {
L
Liu Jicong 已提交
288
          imbCnt++;
289 290 291
          continue;
        } else {
          // pop until equal minVg + 1
292 293
          while (taosArrayGetSize(pConsumerEp->vgs) > minVgCnt + 1) {
            SMqVgEp       *pVgEp = *(SMqVgEp **)taosArrayPop(pConsumerEp->vgs);
294
            SMqRebOutputVg outputVg = {
295
                .oldConsumerId = pConsumerEp->consumerId,
296 297 298 299
                .newConsumerId = -1,
                .pVgEp = pVgEp,
            };
            taosHashPut(pHash, &pVgEp->vgId, sizeof(int32_t), &outputVg, sizeof(SMqRebOutputVg));
L
Liu Jicong 已提交
300 301
            mInfo("mq rebalance: remove vgId:%d from consumer:%" PRId64 ",(first scan)", pVgEp->vgId,
                  pConsumerEp->consumerId);
302 303 304 305 306
          }
          imbCnt++;
        }
      } else {
        // pop until equal minVg
307 308
        while (taosArrayGetSize(pConsumerEp->vgs) > minVgCnt) {
          SMqVgEp       *pVgEp = *(SMqVgEp **)taosArrayPop(pConsumerEp->vgs);
309
          SMqRebOutputVg outputVg = {
310
              .oldConsumerId = pConsumerEp->consumerId,
311 312 313 314
              .newConsumerId = -1,
              .pVgEp = pVgEp,
          };
          taosHashPut(pHash, &pVgEp->vgId, sizeof(int32_t), &outputVg, sizeof(SMqRebOutputVg));
L
Liu Jicong 已提交
315 316
          mInfo("mq rebalance: remove vgId:%d from consumer:%" PRId64 ",(first scan)", pVgEp->vgId,
                pConsumerEp->consumerId);
317 318 319 320 321 322 323 324 325 326 327
        }
      }
    }
  }

  // 5. add new consumer into sub
  {
    int32_t consumerNum = taosArrayGetSize(pInput->pRebInfo->newConsumers);
    for (int32_t i = 0; i < consumerNum; i++) {
      int64_t consumerId = *(int64_t *)taosArrayGet(pInput->pRebInfo->newConsumers, i);
      ASSERT(consumerId > 0);
328
      SMqConsumerEp newConsumerEp;
329 330
      newConsumerEp.consumerId = consumerId;
      newConsumerEp.vgs = taosArrayInit(0, sizeof(void *));
331
      taosHashPut(pOutput->pSub->consumerHash, &consumerId, sizeof(int64_t), &newConsumerEp, sizeof(SMqConsumerEp));
332
      taosArrayPush(pOutput->newConsumers, &consumerId);
S
Shengliang Guan 已提交
333
      mInfo("mq rebalance: add new consumer:%" PRId64, consumerId);
334 335 336 337 338 339 340 341 342 343 344
    }
  }

  // 6. second scan: find consumer do not have enough vg, extract from temporary hash and assign to new consumer.
  // All related vg should be put into rebVgs
  SMqRebOutputVg *pRebVg = NULL;
  void           *pRemovedIter = NULL;
  pIter = NULL;
  while (1) {
    pIter = taosHashIterate(pOutput->pSub->consumerHash, pIter);
    if (pIter == NULL) break;
345 346
    SMqConsumerEp *pConsumerEp = (SMqConsumerEp *)pIter;
    ASSERT(pConsumerEp->consumerId > 0);
L
Liu Jicong 已提交
347 348

    // push until equal minVg
349
    while (taosArrayGetSize(pConsumerEp->vgs) < minVgCnt) {
L
Liu Jicong 已提交
350 351 352 353 354
      // iter hash and find one vg
      pRemovedIter = taosHashIterate(pHash, pRemovedIter);
      ASSERT(pRemovedIter);
      pRebVg = (SMqRebOutputVg *)pRemovedIter;
      // push
355 356
      taosArrayPush(pConsumerEp->vgs, &pRebVg->pVgEp);
      pRebVg->newConsumerId = pConsumerEp->consumerId;
L
Liu Jicong 已提交
357
      taosArrayPush(pOutput->rebVgs, pRebVg);
L
Liu Jicong 已提交
358
      mInfo("mq rebalance: add vgId:%d to consumer:%" PRId64 " (second scan) (not enough)", pRebVg->pVgEp->vgId,
L
Liu Jicong 已提交
359
            pConsumerEp->consumerId);
L
Liu Jicong 已提交
360
    }
361 362
  }

363
  ASSERT(pIter == NULL);
364
  // 7. handle unassigned vg
365
  if (taosHashGetSize(pOutput->pSub->consumerHash) != 0) {
L
Liu Jicong 已提交
366 367
    // if has consumer, assign all left vg
    while (1) {
L
Liu Jicong 已提交
368
      SMqConsumerEp *pConsumerEp = NULL;
L
Liu Jicong 已提交
369
      pRemovedIter = taosHashIterate(pHash, pRemovedIter);
L
Liu Jicong 已提交
370 371 372 373 374 375 376 377 378 379 380 381
      if (pRemovedIter == NULL) {
        if (pIter != NULL) {
          taosHashCancelIterate(pOutput->pSub->consumerHash, pIter);
          pIter = NULL;
        }
        break;
      }
      while (1) {
        pIter = taosHashIterate(pOutput->pSub->consumerHash, pIter);
        ASSERT(pIter);
        pConsumerEp = (SMqConsumerEp *)pIter;
        ASSERT(pConsumerEp->consumerId > 0);
382 383
        if (taosArrayGetSize(pConsumerEp->vgs) == minVgCnt) {
          break;
L
Liu Jicong 已提交
384 385
        }
      }
L
Liu Jicong 已提交
386
      pRebVg = (SMqRebOutputVg *)pRemovedIter;
387 388
      taosArrayPush(pConsumerEp->vgs, &pRebVg->pVgEp);
      pRebVg->newConsumerId = pConsumerEp->consumerId;
L
Liu Jicong 已提交
389
      if (pRebVg->newConsumerId == pRebVg->oldConsumerId) {
L
Liu Jicong 已提交
390
        mInfo("mq rebalance: skip vg %d for same consumer:%" PRId64 " (second scan)", pRebVg->pVgEp->vgId,
L
Liu Jicong 已提交
391 392 393
              pConsumerEp->consumerId);
        continue;
      }
L
Liu Jicong 已提交
394
      taosArrayPush(pOutput->rebVgs, pRebVg);
L
Liu Jicong 已提交
395
      mInfo("mq rebalance: add vgId:%d to consumer:%" PRId64 " (second scan) (unassigned)", pRebVg->pVgEp->vgId,
L
Liu Jicong 已提交
396
            pConsumerEp->consumerId);
L
Liu Jicong 已提交
397
    }
398 399 400 401 402 403 404 405 406
  } else {
    // if all consumer is removed, put all vg into unassigned
    pIter = NULL;
    SMqRebOutputVg *pRebOutput = NULL;
    while (1) {
      pIter = taosHashIterate(pHash, pIter);
      if (pIter == NULL) break;
      pRebOutput = (SMqRebOutputVg *)pIter;
      ASSERT(pRebOutput->newConsumerId == -1);
407
      taosArrayPush(pOutput->pSub->unassignedVgs, &pRebOutput->pVgEp);
L
Liu Jicong 已提交
408
      taosArrayPush(pOutput->rebVgs, pRebOutput);
S
Shengliang Guan 已提交
409
      mInfo("mq rebalance: unassign vgId:%d (second scan)", pRebOutput->pVgEp->vgId);
410 411 412
    }
  }

L
Liu Jicong 已提交
413 414
  // 8. generate logs
  mInfo("mq rebalance: calculation completed, rebalanced vg:");
L
Liu Jicong 已提交
415 416
  for (int32_t i = 0; i < taosArrayGetSize(pOutput->rebVgs); i++) {
    SMqRebOutputVg *pOutputRebVg = taosArrayGet(pOutput->rebVgs, i);
L
Liu Jicong 已提交
417
    mInfo("mq rebalance: vgId:%d, moved from consumer:%" PRId64 ", to consumer:%" PRId64, pOutputRebVg->pVgEp->vgId,
418
          pOutputRebVg->oldConsumerId, pOutputRebVg->newConsumerId);
L
Liu Jicong 已提交
419
  }
L
Liu Jicong 已提交
420 421 422 423 424 425 426
  {
    void *pIter = NULL;
    while (1) {
      pIter = taosHashIterate(pOutput->pSub->consumerHash, pIter);
      if (pIter == NULL) break;
      SMqConsumerEp *pConsumerEp = (SMqConsumerEp *)pIter;
      int32_t        sz = taosArrayGetSize(pConsumerEp->vgs);
427
      mInfo("mq rebalance: final cfg: consumer %" PRId64 " has %d vg", pConsumerEp->consumerId, sz);
L
Liu Jicong 已提交
428 429
      for (int32_t i = 0; i < sz; i++) {
        SMqVgEp *pVgEp = taosArrayGetP(pConsumerEp->vgs, i);
430
        mInfo("mq rebalance: final cfg: vg %d to consumer %" PRId64 "", pVgEp->vgId, pConsumerEp->consumerId);
L
Liu Jicong 已提交
431 432 433
      }
    }
  }
434 435 436 437 438 439 440

  // 9. clear
  taosHashCleanup(pHash);

  return 0;
}

S
Shengliang Guan 已提交
441
static int32_t mndPersistRebResult(SMnode *pMnode, SRpcMsg *pMsg, const SMqRebOutputObj *pOutput) {
442
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_DB_INSIDE, pMsg, "persist-reb");
443
  mndTransSetDbName(pTrans, pOutput->pSub->dbName, NULL);
444 445
  if (pTrans == NULL) return -1;

446 447 448 449 450 451
  // make txn:
  // 1. redo action: action to all vg
  const SArray *rebVgs = pOutput->rebVgs;
  int32_t       vgNum = taosArrayGetSize(rebVgs);
  for (int32_t i = 0; i < vgNum; i++) {
    SMqRebOutputVg *pRebVg = taosArrayGet(rebVgs, i);
L
Liu Jicong 已提交
452
    if (mndPersistSubChangeVgReq(pMnode, pTrans, pOutput->pSub, pRebVg) < 0) {
453 454 455 456 457 458
      goto REB_FAIL;
    }
  }

  // 2. redo log: subscribe and vg assignment
  // subscribe
L
Liu Jicong 已提交
459
  if (mndSetSubCommitLogs(pMnode, pTrans, pOutput->pSub) != 0) {
460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478
    goto REB_FAIL;
  }

  // 3. commit log: consumer to update status and epoch
  // 3.1 set touched consumer
  int32_t consumerNum = taosArrayGetSize(pOutput->touchedConsumers);
  for (int32_t i = 0; i < consumerNum; i++) {
    int64_t         consumerId = *(int64_t *)taosArrayGet(pOutput->touchedConsumers, i);
    SMqConsumerObj *pConsumerOld = mndAcquireConsumer(pMnode, consumerId);
    SMqConsumerObj *pConsumerNew = tNewSMqConsumerObj(pConsumerOld->consumerId, pConsumerOld->cgroup);
    pConsumerNew->updateType = CONSUMER_UPDATE__TOUCH;
    mndReleaseConsumer(pMnode, pConsumerOld);
    if (mndSetConsumerCommitLogs(pMnode, pTrans, pConsumerNew) != 0) {
      goto REB_FAIL;
    }
  }
  // 3.2 set new consumer
  consumerNum = taosArrayGetSize(pOutput->newConsumers);
  for (int32_t i = 0; i < consumerNum; i++) {
L
Liu Jicong 已提交
479
    int64_t consumerId = *(int64_t *)taosArrayGet(pOutput->newConsumers, i);
480 481 482 483 484 485
    ASSERT(consumerId > 0);
    SMqConsumerObj *pConsumerOld = mndAcquireConsumer(pMnode, consumerId);
    SMqConsumerObj *pConsumerNew = tNewSMqConsumerObj(pConsumerOld->consumerId, pConsumerOld->cgroup);
    pConsumerNew->updateType = CONSUMER_UPDATE__ADD;
    char *topic = taosMemoryCalloc(1, TSDB_TOPIC_FNAME_LEN);
    char  cgroup[TSDB_CGROUP_LEN];
L
Liu Jicong 已提交
486
    mndSplitSubscribeKey(pOutput->pSub->key, topic, cgroup, true);
487 488 489
    taosArrayPush(pConsumerNew->rebNewTopics, &topic);
    mndReleaseConsumer(pMnode, pConsumerOld);
    if (mndSetConsumerCommitLogs(pMnode, pTrans, pConsumerNew) != 0) {
L
Liu Jicong 已提交
490
      ASSERT(0);
L
Liu Jicong 已提交
491 492
      tDeleteSMqConsumerObj(pConsumerNew);
      taosMemoryFree(pConsumerNew);
493 494
      goto REB_FAIL;
    }
L
Liu Jicong 已提交
495 496
    tDeleteSMqConsumerObj(pConsumerNew);
    taosMemoryFree(pConsumerNew);
497 498 499 500 501 502 503 504 505 506 507 508
  }

  // 3.3 set removed consumer
  consumerNum = taosArrayGetSize(pOutput->removedConsumers);
  for (int32_t i = 0; i < consumerNum; i++) {
    int64_t consumerId = *(int64_t *)taosArrayGet(pOutput->removedConsumers, i);
    ASSERT(consumerId > 0);
    SMqConsumerObj *pConsumerOld = mndAcquireConsumer(pMnode, consumerId);
    SMqConsumerObj *pConsumerNew = tNewSMqConsumerObj(pConsumerOld->consumerId, pConsumerOld->cgroup);
    pConsumerNew->updateType = CONSUMER_UPDATE__REMOVE;
    char *topic = taosMemoryCalloc(1, TSDB_TOPIC_FNAME_LEN);
    char  cgroup[TSDB_CGROUP_LEN];
L
Liu Jicong 已提交
509
    mndSplitSubscribeKey(pOutput->pSub->key, topic, cgroup, true);
510 511 512
    taosArrayPush(pConsumerNew->rebRemovedTopics, &topic);
    mndReleaseConsumer(pMnode, pConsumerOld);
    if (mndSetConsumerCommitLogs(pMnode, pTrans, pConsumerNew) != 0) {
L
Liu Jicong 已提交
513
      ASSERT(0);
L
Liu Jicong 已提交
514 515
      tDeleteSMqConsumerObj(pConsumerNew);
      taosMemoryFree(pConsumerNew);
516 517
      goto REB_FAIL;
    }
L
Liu Jicong 已提交
518 519
    tDeleteSMqConsumerObj(pConsumerNew);
    taosMemoryFree(pConsumerNew);
520
  }
L
Liu Jicong 已提交
521
#if 0
L
Liu Jicong 已提交
522 523 524 525 526 527 528 529 530 531
  if (consumerNum) {
    char topic[TSDB_TOPIC_FNAME_LEN];
    char cgroup[TSDB_CGROUP_LEN];
    mndSplitSubscribeKey(pOutput->pSub->key, topic, cgroup, true);
    SMqTopicObj *pTopic = mndAcquireTopic(pMnode, topic);
    if (pTopic) {
      // TODO make topic complete
      SMqTopicObj topicObj = {0};
      memcpy(&topicObj, pTopic, sizeof(SMqTopicObj));
      topicObj.refConsumerCnt = pTopic->refConsumerCnt - consumerNum;
L
Liu Jicong 已提交
532 533 534 535
      // TODO is that correct?
      pTopic->refConsumerCnt = topicObj.refConsumerCnt;
      mInfo("subscribe topic %s unref %d consumer cgroup %s, refcnt %d", pTopic->name, consumerNum, cgroup,
            topicObj.refConsumerCnt);
L
Liu Jicong 已提交
536 537 538 539
      if (mndSetTopicCommitLogs(pMnode, pTrans, &topicObj) != 0) {
        ASSERT(0);
        goto REB_FAIL;
      }
L
Liu Jicong 已提交
540 541
    }
  }
L
Liu Jicong 已提交
542
#endif
L
Liu Jicong 已提交
543

L
Liu Jicong 已提交
544 545
  // 4. TODO commit log: modification log

L
Liu Jicong 已提交
546
  // 5. set cb
547
  mndTransSetCb(pTrans, TRANS_START_FUNC_MQ_REB, TRANS_STOP_FUNC_MQ_REB, NULL, 0);
L
Liu Jicong 已提交
548 549

  // 6. execution
L
Liu Jicong 已提交
550
  if (mndTransPrepare(pMnode, pTrans) != 0) {
L
Liu Jicong 已提交
551
    mError("failed to prepare trans rebalance since %s", terrstr());
L
Liu Jicong 已提交
552 553
    goto REB_FAIL;
  }
554 555 556 557 558 559 560 561 562

  mndTransDrop(pTrans);
  return 0;

REB_FAIL:
  mndTransDrop(pTrans);
  return -1;
}

S
Shengliang Guan 已提交
563 564 565
static int32_t mndProcessRebalanceReq(SRpcMsg *pMsg) {
  SMnode            *pMnode = pMsg->info.node;
  SMqDoRebalanceMsg *pReq = pMsg->pCont;
566 567 568 569 570 571 572
  void              *pIter = NULL;

  mInfo("mq rebalance start");

  while (1) {
    pIter = taosHashIterate(pReq->rebSubHash, pIter);
    if (pIter == NULL) break;
L
Liu Jicong 已提交
573 574 575
    SMqRebInputObj rebInput = {0};

    SMqRebOutputObj rebOutput = {0};
L
Liu Jicong 已提交
576 577 578
    rebOutput.newConsumers = taosArrayInit(0, sizeof(int64_t));
    rebOutput.removedConsumers = taosArrayInit(0, sizeof(int64_t));
    rebOutput.touchedConsumers = taosArrayInit(0, sizeof(int64_t));
L
Liu Jicong 已提交
579 580
    rebOutput.rebVgs = taosArrayInit(0, sizeof(SMqRebOutputVg));

L
Liu Jicong 已提交
581 582 583 584
    SMqRebInfo      *pRebInfo = (SMqRebInfo *)pIter;
    SMqSubscribeObj *pSub = mndAcquireSubscribeByKey(pMnode, pRebInfo->key);

    rebInput.pRebInfo = pRebInfo;
585 586 587 588 589

    if (pSub == NULL) {
      // split sub key and extract topic
      char topic[TSDB_TOPIC_FNAME_LEN];
      char cgroup[TSDB_CGROUP_LEN];
L
Liu Jicong 已提交
590
      mndSplitSubscribeKey(pRebInfo->key, topic, cgroup, true);
591
      SMqTopicObj *pTopic = mndAcquireTopic(pMnode, topic);
L
Liu Jicong 已提交
592 593
      /*ASSERT(pTopic);*/
      if (pTopic == NULL) {
L
Liu Jicong 已提交
594
        mError("mq rebalance %s failed since topic %s not exist, abort", pRebInfo->key, topic);
L
Liu Jicong 已提交
595 596
        continue;
      }
597 598
      taosRLockLatch(&pTopic->lock);

L
Liu Jicong 已提交
599
      rebOutput.pSub = mndCreateSub(pMnode, pTopic, pRebInfo->key);
L
Liu Jicong 已提交
600
      memcpy(rebOutput.pSub->dbName, pTopic->db, TSDB_DB_FNAME_LEN);
L
Liu Jicong 已提交
601 602 603 604 605 606 607 608 609 610 611 612 613
      ASSERT(taosHashGetSize(rebOutput.pSub->consumerHash) == 0);

      taosRUnLockLatch(&pTopic->lock);
      mndReleaseTopic(pMnode, pTopic);

      rebInput.oldConsumerNum = 0;
    } else {
      taosRLockLatch(&pSub->lock);
      rebInput.oldConsumerNum = taosHashGetSize(pSub->consumerHash);
      rebOutput.pSub = tCloneSubscribeObj(pSub);
      taosRUnLockLatch(&pSub->lock);
      mndReleaseSubscribe(pMnode, pSub);
    }
614

615 616 617
    if (mndDoRebalance(pMnode, &rebInput, &rebOutput) < 0) {
      mError("mq rebalance internal error");
    }
618

L
Liu Jicong 已提交
619 620 621
    // if add more consumer to balanced subscribe,
    // possibly no vg is changed
    /*ASSERT(taosArrayGetSize(rebOutput.rebVgs) != 0);*/
622

L
Liu Jicong 已提交
623
    if (mndPersistRebResult(pMnode, pMsg, &rebOutput) < 0) {
L
Liu Jicong 已提交
624
      mError("mq rebalance persist rebalance output error, possibly vnode splitted or dropped");
L
Liu Jicong 已提交
625
    }
L
Liu Jicong 已提交
626 627 628 629 630 631 632 633 634 635
    taosArrayDestroy(pRebInfo->lostConsumers);
    taosArrayDestroy(pRebInfo->newConsumers);
    taosArrayDestroy(pRebInfo->removedConsumers);

    taosArrayDestroy(rebOutput.newConsumers);
    taosArrayDestroy(rebOutput.touchedConsumers);
    taosArrayDestroy(rebOutput.removedConsumers);
    taosArrayDestroy(rebOutput.rebVgs);
    tDeleteSubscribeObj(rebOutput.pSub);
    taosMemoryFree(rebOutput.pSub);
636 637 638 639 640
  }

  // reset flag
  mInfo("mq rebalance completed successfully");
  taosHashCleanup(pReq->rebSubHash);
L
Liu Jicong 已提交
641
  mndRebEnd();
642 643 644

  return 0;
}
L
Liu Jicong 已提交
645

L
Liu Jicong 已提交
646
static int32_t mndProcessDropCgroupReq(SRpcMsg *pReq) {
L
Liu Jicong 已提交
647 648
  SMnode         *pMnode = pReq->info.node;
  SSdb           *pSdb = pMnode->pSdb;
L
Liu Jicong 已提交
649 650 651 652 653 654 655 656 657 658
  SMDropCgroupReq dropReq = {0};

  if (tDeserializeSMDropCgroupReq(pReq->pCont, pReq->contLen, &dropReq) != 0) {
    terrno = TSDB_CODE_INVALID_MSG;
    return -1;
  }

  SMqSubscribeObj *pSub = mndAcquireSubscribe(pMnode, dropReq.cgroup, dropReq.topic);
  if (pSub == NULL) {
    if (dropReq.igNotExists) {
659
      mInfo("cgroup:%s on topic:%s, not exist, ignore not exist is set", dropReq.cgroup, dropReq.topic);
L
Liu Jicong 已提交
660 661 662 663 664 665 666 667
      return 0;
    } else {
      terrno = TSDB_CODE_MND_SUBSCRIBE_NOT_EXIST;
      mError("topic:%s, cgroup:%s, failed to drop since %s", dropReq.topic, dropReq.cgroup, terrstr());
      return -1;
    }
  }

L
Liu Jicong 已提交
668
  if (taosHashGetSize(pSub->consumerHash) != 0) {
L
Liu Jicong 已提交
669 670
    terrno = TSDB_CODE_MND_CGROUP_USED;
    mError("cgroup:%s on topic:%s, failed to drop since %s", dropReq.cgroup, dropReq.topic, terrstr());
L
Liu Jicong 已提交
671
    mndReleaseSubscribe(pMnode, pSub);
L
Liu Jicong 已提交
672 673 674
    return -1;
  }

675
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pReq, "drop-cgroup");
L
Liu Jicong 已提交
676 677
  if (pTrans == NULL) {
    mError("cgroup: %s on topic:%s, failed to drop since %s", dropReq.cgroup, dropReq.topic, terrstr());
L
Liu Jicong 已提交
678
    mndReleaseSubscribe(pMnode, pSub);
L
Liu Jicong 已提交
679
    mndTransDrop(pTrans);
L
Liu Jicong 已提交
680 681 682
    return -1;
  }

683
  mInfo("trans:%d, used to drop cgroup:%s on topic %s", pTrans->id, dropReq.cgroup, dropReq.topic);
L
Liu Jicong 已提交
684 685 686

  if (mndSetDropSubCommitLogs(pMnode, pTrans, pSub) < 0) {
    mError("cgroup %s on topic:%s, failed to drop since %s", dropReq.cgroup, dropReq.topic, terrstr());
L
Liu Jicong 已提交
687
    mndReleaseSubscribe(pMnode, pSub);
L
Liu Jicong 已提交
688
    mndTransDrop(pTrans);
L
Liu Jicong 已提交
689 690 691
    return -1;
  }

L
Liu Jicong 已提交
692 693 694 695 696
  if (mndTransPrepare(pMnode, pTrans) < 0) {
    mndReleaseSubscribe(pMnode, pSub);
    mndTransDrop(pTrans);
    return -1;
  }
L
Liu Jicong 已提交
697 698 699 700 701
  mndReleaseSubscribe(pMnode, pSub);

  return TSDB_CODE_ACTION_IN_PROGRESS;
}

L
Liu Jicong 已提交
702 703 704
void mndCleanupSubscribe(SMnode *pMnode) {}

static SSdbRaw *mndSubActionEncode(SMqSubscribeObj *pSub) {
L
Liu Jicong 已提交
705
  terrno = TSDB_CODE_OUT_OF_MEMORY;
L
Liu Jicong 已提交
706
  void   *buf = NULL;
L
Liu Jicong 已提交
707
  int32_t tlen = tEncodeSubscribeObj(NULL, pSub);
L
Liu Jicong 已提交
708
  int32_t size = sizeof(int32_t) + tlen + MND_SUBSCRIBE_RESERVE_SIZE;
L
Liu Jicong 已提交
709 710 711 712

  SSdbRaw *pRaw = sdbAllocRaw(SDB_SUBSCRIBE, MND_SUBSCRIBE_VER_NUMBER, size);
  if (pRaw == NULL) goto SUB_ENCODE_OVER;

wafwerar's avatar
wafwerar 已提交
713
  buf = taosMemoryMalloc(tlen);
L
Liu Jicong 已提交
714
  if (buf == NULL) goto SUB_ENCODE_OVER;
L
Liu Jicong 已提交
715

L
Liu Jicong 已提交
716 717
  void *abuf = buf;
  tEncodeSubscribeObj(&abuf, pSub);
L
Liu Jicong 已提交
718 719 720 721 722 723 724

  int32_t dataPos = 0;
  SDB_SET_INT32(pRaw, dataPos, tlen, SUB_ENCODE_OVER);
  SDB_SET_BINARY(pRaw, dataPos, buf, tlen, SUB_ENCODE_OVER);
  SDB_SET_RESERVE(pRaw, dataPos, MND_SUBSCRIBE_RESERVE_SIZE, SUB_ENCODE_OVER);
  SDB_SET_DATALEN(pRaw, dataPos, SUB_ENCODE_OVER);

L
Liu Jicong 已提交
725 726
  terrno = TSDB_CODE_SUCCESS;

L
Liu Jicong 已提交
727
SUB_ENCODE_OVER:
wafwerar's avatar
wafwerar 已提交
728
  taosMemoryFreeClear(buf);
L
Liu Jicong 已提交
729
  if (terrno != TSDB_CODE_SUCCESS) {
L
Liu Jicong 已提交
730 731 732 733 734 735 736 737 738 739 740
    mError("subscribe:%s, failed to encode to raw:%p since %s", pSub->key, pRaw, terrstr());
    sdbFreeRaw(pRaw);
    return NULL;
  }

  mTrace("subscribe:%s, encode to raw:%p, row:%p", pSub->key, pRaw, pSub);
  return pRaw;
}

static SSdbRow *mndSubActionDecode(SSdbRaw *pRaw) {
  terrno = TSDB_CODE_OUT_OF_MEMORY;
L
Liu Jicong 已提交
741
  void *buf = NULL;
L
Liu Jicong 已提交
742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759

  int8_t sver = 0;
  if (sdbGetRawSoftVer(pRaw, &sver) != 0) goto SUB_DECODE_OVER;

  if (sver != MND_SUBSCRIBE_VER_NUMBER) {
    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
    goto SUB_DECODE_OVER;
  }

  int32_t  size = sizeof(SMqSubscribeObj);
  SSdbRow *pRow = sdbAllocRow(size);
  if (pRow == NULL) goto SUB_DECODE_OVER;

  SMqSubscribeObj *pSub = sdbGetRowObj(pRow);
  if (pSub == NULL) goto SUB_DECODE_OVER;

  int32_t dataPos = 0;
  int32_t tlen;
L
Liu Jicong 已提交
760
  SDB_GET_INT32(pRaw, dataPos, &tlen, SUB_DECODE_OVER);
L
Liu Jicong 已提交
761
  buf = taosMemoryMalloc(tlen);
L
Liu Jicong 已提交
762 763 764 765 766 767 768 769
  if (buf == NULL) goto SUB_DECODE_OVER;
  SDB_GET_BINARY(pRaw, dataPos, buf, tlen, SUB_DECODE_OVER);
  SDB_GET_RESERVE(pRaw, dataPos, MND_SUBSCRIBE_RESERVE_SIZE, SUB_DECODE_OVER);

  if (tDecodeSubscribeObj(buf, pSub) == NULL) {
    goto SUB_DECODE_OVER;
  }

L
Liu Jicong 已提交
770 771
  terrno = TSDB_CODE_SUCCESS;

L
Liu Jicong 已提交
772
SUB_DECODE_OVER:
wafwerar's avatar
wafwerar 已提交
773
  taosMemoryFreeClear(buf);
L
Liu Jicong 已提交
774
  if (terrno != TSDB_CODE_SUCCESS) {
L
Liu Jicong 已提交
775
    mError("subscribe:%s, failed to decode from raw:%p since %s", pSub->key, pRaw, terrstr());
wafwerar's avatar
wafwerar 已提交
776
    taosMemoryFreeClear(pRow);
L
Liu Jicong 已提交
777 778 779 780 781 782 783 784 785 786 787 788 789
    return NULL;
  }

  return pRow;
}

static int32_t mndSubActionInsert(SSdb *pSdb, SMqSubscribeObj *pSub) {
  mTrace("subscribe:%s, perform insert action", pSub->key);
  return 0;
}

static int32_t mndSubActionDelete(SSdb *pSdb, SMqSubscribeObj *pSub) {
  mTrace("subscribe:%s, perform delete action", pSub->key);
790
  tDeleteSubscribeObj(pSub);
L
Liu Jicong 已提交
791 792 793 794 795
  return 0;
}

static int32_t mndSubActionUpdate(SSdb *pSdb, SMqSubscribeObj *pOldSub, SMqSubscribeObj *pNewSub) {
  mTrace("subscribe:%s, perform update action", pOldSub->key);
796 797 798 799 800 801
  taosWLockLatch(&pOldSub->lock);

  SHashObj *tmp = pOldSub->consumerHash;
  pOldSub->consumerHash = pNewSub->consumerHash;
  pNewSub->consumerHash = tmp;

802 803 804 805
  SArray *tmp1 = pOldSub->unassignedVgs;
  pOldSub->unassignedVgs = pNewSub->unassignedVgs;
  pNewSub->unassignedVgs = tmp1;

806
  taosWUnLockLatch(&pOldSub->lock);
L
Liu Jicong 已提交
807 808 809
  return 0;
}

810
int32_t mndMakeSubscribeKey(char *key, const char *cgroup, const char *topicName) {
S
Shengliang Guan 已提交
811
  int32_t tlen = strlen(cgroup);
L
Liu Jicong 已提交
812
  memcpy(key, cgroup, tlen);
L
Liu Jicong 已提交
813
  key[tlen] = TMQ_SEPARATOR;
L
Liu Jicong 已提交
814
  strcpy(key + tlen + 1, topicName);
L
Liu Jicong 已提交
815
  return 0;
L
Liu Jicong 已提交
816 817
}

L
Liu Jicong 已提交
818
SMqSubscribeObj *mndAcquireSubscribe(SMnode *pMnode, const char *cgroup, const char *topicName) {
L
Liu Jicong 已提交
819 820 821
  SSdb *pSdb = pMnode->pSdb;
  char  key[TSDB_SUBSCRIBE_KEY_LEN];
  mndMakeSubscribeKey(key, cgroup, topicName);
L
Liu Jicong 已提交
822 823
  SMqSubscribeObj *pSub = sdbAcquire(pSdb, SDB_SUBSCRIBE, key);
  if (pSub == NULL) {
824
    terrno = TSDB_CODE_MND_SUBSCRIBE_NOT_EXIST;
L
Liu Jicong 已提交
825 826 827 828
  }
  return pSub;
}

L
Liu Jicong 已提交
829 830 831 832
SMqSubscribeObj *mndAcquireSubscribeByKey(SMnode *pMnode, const char *key) {
  SSdb            *pSdb = pMnode->pSdb;
  SMqSubscribeObj *pSub = sdbAcquire(pSdb, SDB_SUBSCRIBE, key);
  if (pSub == NULL) {
833
    terrno = TSDB_CODE_MND_SUBSCRIBE_NOT_EXIST;
L
Liu Jicong 已提交
834 835 836 837
  }
  return pSub;
}

L
Liu Jicong 已提交
838 839 840 841 842
void mndReleaseSubscribe(SMnode *pMnode, SMqSubscribeObj *pSub) {
  SSdb *pSdb = pMnode->pSdb;
  sdbRelease(pSdb, pSub);
}

L
Liu Jicong 已提交
843 844 845 846 847 848 849 850
static int32_t mndSetDropSubRedoLogs(SMnode *pMnode, STrans *pTrans, SMqSubscribeObj *pSub) {
  SSdbRaw *pRedoRaw = mndSubActionEncode(pSub);
  if (pRedoRaw == NULL) return -1;
  if (mndTransAppendRedolog(pTrans, pRedoRaw) != 0) return -1;
  if (sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPED) != 0) return -1;
  return 0;
}

L
Liu Jicong 已提交
851
int32_t mndSetDropSubCommitLogs(SMnode *pMnode, STrans *pTrans, SMqSubscribeObj *pSub) {
L
Liu Jicong 已提交
852 853 854 855 856 857 858 859
  SSdbRaw *pCommitRaw = mndSubActionEncode(pSub);
  if (pCommitRaw == NULL) return -1;
  if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) return -1;
  if (sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED) != 0) return -1;
  return 0;
}

int32_t mndDropSubByDB(SMnode *pMnode, STrans *pTrans, SDbObj *pDb) {
860
  int32_t code = 0;
L
Liu Jicong 已提交
861 862 863 864 865 866 867 868 869 870 871 872 873 874
  SSdb   *pSdb = pMnode->pSdb;

  void            *pIter = NULL;
  SMqSubscribeObj *pSub = NULL;
  while (1) {
    pIter = sdbFetch(pSdb, SDB_SUBSCRIBE, pIter, (void **)&pSub);
    if (pIter == NULL) break;

    if (pSub->dbUid != pDb->uid) {
      sdbRelease(pSdb, pSub);
      continue;
    }

    if (mndSetDropSubCommitLogs(pMnode, pTrans, pSub) < 0) {
L
Liu Jicong 已提交
875
      sdbRelease(pSdb, pSub);
876 877 878
      sdbCancelFetch(pSdb, pIter);
      code = -1;
      break;
L
Liu Jicong 已提交
879
    }
880 881

    sdbRelease(pSdb, pSub);
L
Liu Jicong 已提交
882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905
  }

  return code;
}

int32_t mndDropSubByTopic(SMnode *pMnode, STrans *pTrans, const char *topicName) {
  int32_t code = -1;
  SSdb   *pSdb = pMnode->pSdb;

  void            *pIter = NULL;
  SMqSubscribeObj *pSub = NULL;
  while (1) {
    pIter = sdbFetch(pSdb, SDB_SUBSCRIBE, pIter, (void **)&pSub);
    if (pIter == NULL) break;

    char topic[TSDB_TOPIC_FNAME_LEN];
    char cgroup[TSDB_CGROUP_LEN];
    mndSplitSubscribeKey(pSub->key, topic, cgroup, true);
    if (strcmp(topic, topicName) != 0) {
      sdbRelease(pSdb, pSub);
      continue;
    }

    // iter all vnode to delete handle
906 907
    if (taosHashGetSize(pSub->consumerHash) != 0) {
      sdbRelease(pSdb, pSub);
L
Liu Jicong 已提交
908
      terrno = TSDB_CODE_MND_IN_REBALANCE;
909 910
      return -1;
    }
L
Liu Jicong 已提交
911 912 913 914 915 916 917 918 919 920 921 922
    int32_t sz = taosArrayGetSize(pSub->unassignedVgs);
    for (int32_t i = 0; i < sz; i++) {
      SMqVgEp       *pVgEp = taosArrayGetP(pSub->unassignedVgs, i);
      SMqVDeleteReq *pReq = taosMemoryCalloc(1, sizeof(SMqVDeleteReq));
      pReq->head.vgId = htonl(pVgEp->vgId);
      pReq->vgId = pVgEp->vgId;
      pReq->consumerId = -1;
      memcpy(pReq->subKey, pSub->key, TSDB_SUBSCRIBE_KEY_LEN);
      STransAction action = {0};
      action.epSet = pVgEp->epSet;
      action.pCont = pReq;
      action.contLen = sizeof(SMqVDeleteReq);
L
Liu Jicong 已提交
923
      action.msgType = TDMT_VND_TMQ_DELETE_SUB;
L
Liu Jicong 已提交
924 925 926 927 928
      if (mndTransAppendRedoAction(pTrans, &action) != 0) {
        taosMemoryFree(pReq);
        return -1;
      }
    }
L
Liu Jicong 已提交
929 930 931

    if (mndSetDropSubRedoLogs(pMnode, pTrans, pSub) < 0) {
      sdbRelease(pSdb, pSub);
L
Liu Jicong 已提交
932 933 934 935 936 937 938 939
      goto END;
    }
  }

  code = 0;
END:
  return code;
}
L
Liu Jicong 已提交
940

S
Shengliang Guan 已提交
941 942
static int32_t mndRetrieveSubscribe(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rowsCapacity) {
  SMnode          *pMnode = pReq->info.node;
L
Liu Jicong 已提交
943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973
  SSdb            *pSdb = pMnode->pSdb;
  int32_t          numOfRows = 0;
  SMqSubscribeObj *pSub = NULL;

  while (numOfRows < rowsCapacity) {
    pShow->pIter = sdbFetch(pSdb, SDB_SUBSCRIBE, pShow->pIter, (void **)&pSub);
    if (pShow->pIter == NULL) break;

    taosRLockLatch(&pSub->lock);

    if (numOfRows + pSub->vgNum > rowsCapacity) {
      blockDataEnsureCapacity(pBlock, numOfRows + pSub->vgNum);
    }

    SMqConsumerEp *pConsumerEp = NULL;
    void          *pIter = NULL;
    while (1) {
      pIter = taosHashIterate(pSub->consumerHash, pIter);
      if (pIter == NULL) break;
      pConsumerEp = (SMqConsumerEp *)pIter;

      int32_t sz = taosArrayGetSize(pConsumerEp->vgs);
      for (int32_t j = 0; j < sz; j++) {
        SMqVgEp *pVgEp = taosArrayGetP(pConsumerEp->vgs, j);

        SColumnInfoData *pColInfo;
        int32_t          cols = 0;

        // topic and cgroup
        char topic[TSDB_TOPIC_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
        char cgroup[TSDB_CGROUP_LEN + VARSTR_HEADER_SIZE] = {0};
L
Liu Jicong 已提交
974
        mndSplitSubscribeKey(pSub->key, varDataVal(topic), varDataVal(cgroup), false);
L
Liu Jicong 已提交
975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006
        varDataSetLen(topic, strlen(varDataVal(topic)));
        varDataSetLen(cgroup, strlen(varDataVal(cgroup)));

        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
        colDataAppend(pColInfo, numOfRows, (const char *)topic, false);

        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
        colDataAppend(pColInfo, numOfRows, (const char *)cgroup, false);

        // vg id
        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
        colDataAppend(pColInfo, numOfRows, (const char *)&pVgEp->vgId, false);

        // consumer id
        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
        colDataAppend(pColInfo, numOfRows, (const char *)&pConsumerEp->consumerId, false);

        // offset
#if 0
      // subscribe time
      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
      colDataAppend(pColInfo, numOfRows, (const char *)&pSub->subscribeTime, false);

      // rebalance time
      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
      colDataAppend(pColInfo, numOfRows, (const char *)&pSub->rebalanceTime, pConsumer->rebalanceTime == 0);
#endif

        numOfRows++;
      }
    }

L
Liu Jicong 已提交
1007
    // do not show for cleared subscription
1008
#if 1
L
Liu Jicong 已提交
1009 1010 1011 1012 1013 1014 1015 1016 1017 1018
    int32_t sz = taosArrayGetSize(pSub->unassignedVgs);
    for (int32_t i = 0; i < sz; i++) {
      SMqVgEp *pVgEp = taosArrayGetP(pSub->unassignedVgs, i);

      SColumnInfoData *pColInfo;
      int32_t          cols = 0;

      // topic and cgroup
      char topic[TSDB_TOPIC_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
      char cgroup[TSDB_CGROUP_LEN + VARSTR_HEADER_SIZE] = {0};
L
Liu Jicong 已提交
1019
      mndSplitSubscribeKey(pSub->key, varDataVal(topic), varDataVal(cgroup), false);
L
Liu Jicong 已提交
1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050
      varDataSetLen(topic, strlen(varDataVal(topic)));
      varDataSetLen(cgroup, strlen(varDataVal(cgroup)));

      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
      colDataAppend(pColInfo, numOfRows, (const char *)topic, false);

      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
      colDataAppend(pColInfo, numOfRows, (const char *)cgroup, false);

      // vg id
      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
      colDataAppend(pColInfo, numOfRows, (const char *)&pVgEp->vgId, false);

      // consumer id
      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
      colDataAppend(pColInfo, numOfRows, NULL, true);

      // offset
#if 0
      // subscribe time
      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
      colDataAppend(pColInfo, numOfRows, (const char *)&pSub->subscribeTime, false);

      // rebalance time
      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
      colDataAppend(pColInfo, numOfRows, (const char *)&pSub->rebalanceTime, pConsumer->rebalanceTime == 0);
#endif

      numOfRows++;
    }

L
Liu Jicong 已提交
1051
#endif
L
Liu Jicong 已提交
1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063
    taosRUnLockLatch(&pSub->lock);
    sdbRelease(pSdb, pSub);
  }

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

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