mndSubscribe.c 36.6 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
      }
L
Liu Jicong 已提交
238
      taosArrayDestroy(pConsumerEp->vgs);
239 240 241 242 243 244 245 246 247
      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
  {
248
    int32_t consumerVgNum = taosArrayGetSize(pOutput->pSub->unassignedVgs);
249
    for (int32_t i = 0; i < consumerVgNum; i++) {
250
      SMqVgEp       *pVgEp = *(SMqVgEp **)taosArrayPop(pOutput->pSub->unassignedVgs);
251 252 253 254 255 256
      SMqRebOutputVg rebOutput = {
          .oldConsumerId = -1,
          .newConsumerId = -1,
          .pVgEp = pVgEp,
      };
      taosHashPut(pHash, &pVgEp->vgId, sizeof(int32_t), &rebOutput, sizeof(SMqRebOutputVg));
S
Shengliang Guan 已提交
257
      mInfo("mq rebalance: remove vgId:%d from unassigned", pVgEp->vgId);
258 259 260 261
    }
  }

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

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

  // 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);
329
      SMqConsumerEp newConsumerEp;
330 331
      newConsumerEp.consumerId = consumerId;
      newConsumerEp.vgs = taosArrayInit(0, sizeof(void *));
332
      taosHashPut(pOutput->pSub->consumerHash, &consumerId, sizeof(int64_t), &newConsumerEp, sizeof(SMqConsumerEp));
333
      taosArrayPush(pOutput->newConsumers, &consumerId);
S
Shengliang Guan 已提交
334
      mInfo("mq rebalance: add new consumer:%" PRId64, consumerId);
335 336 337 338 339 340 341 342 343 344 345
    }
  }

  // 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;
346 347
    SMqConsumerEp *pConsumerEp = (SMqConsumerEp *)pIter;
    ASSERT(pConsumerEp->consumerId > 0);
L
Liu Jicong 已提交
348 349

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

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

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

  // 9. clear
  taosHashCleanup(pHash);

  return 0;
}

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

447 448 449 450 451 452
  // 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 已提交
453
    if (mndPersistSubChangeVgReq(pMnode, pTrans, pOutput->pSub, pRebVg) < 0) {
454 455 456 457 458 459
      goto REB_FAIL;
    }
  }

  // 2. redo log: subscribe and vg assignment
  // subscribe
L
Liu Jicong 已提交
460
  if (mndSetSubCommitLogs(pMnode, pTrans, pOutput->pSub) != 0) {
461 462 463 464 465 466 467 468 469 470 471 472 473
    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) {
L
Liu Jicong 已提交
474 475
      tDeleteSMqConsumerObj(pConsumerNew);
      taosMemoryFree(pConsumerNew);
476 477
      goto REB_FAIL;
    }
L
Liu Jicong 已提交
478 479
    tDeleteSMqConsumerObj(pConsumerNew);
    taosMemoryFree(pConsumerNew);
480 481 482 483
  }
  // 3.2 set new consumer
  consumerNum = taosArrayGetSize(pOutput->newConsumers);
  for (int32_t i = 0; i < consumerNum; i++) {
L
Liu Jicong 已提交
484
    int64_t consumerId = *(int64_t *)taosArrayGet(pOutput->newConsumers, i);
485 486 487 488 489 490
    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 已提交
491
    mndSplitSubscribeKey(pOutput->pSub->key, topic, cgroup, true);
492 493 494
    taosArrayPush(pConsumerNew->rebNewTopics, &topic);
    mndReleaseConsumer(pMnode, pConsumerOld);
    if (mndSetConsumerCommitLogs(pMnode, pTrans, pConsumerNew) != 0) {
L
Liu Jicong 已提交
495
      ASSERT(0);
L
Liu Jicong 已提交
496 497
      tDeleteSMqConsumerObj(pConsumerNew);
      taosMemoryFree(pConsumerNew);
498 499
      goto REB_FAIL;
    }
L
Liu Jicong 已提交
500 501
    tDeleteSMqConsumerObj(pConsumerNew);
    taosMemoryFree(pConsumerNew);
502 503 504 505 506 507 508 509 510 511 512 513
  }

  // 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 已提交
514
    mndSplitSubscribeKey(pOutput->pSub->key, topic, cgroup, true);
515 516 517
    taosArrayPush(pConsumerNew->rebRemovedTopics, &topic);
    mndReleaseConsumer(pMnode, pConsumerOld);
    if (mndSetConsumerCommitLogs(pMnode, pTrans, pConsumerNew) != 0) {
L
Liu Jicong 已提交
518
      ASSERT(0);
L
Liu Jicong 已提交
519 520
      tDeleteSMqConsumerObj(pConsumerNew);
      taosMemoryFree(pConsumerNew);
521 522
      goto REB_FAIL;
    }
L
Liu Jicong 已提交
523 524
    tDeleteSMqConsumerObj(pConsumerNew);
    taosMemoryFree(pConsumerNew);
525
  }
L
Liu Jicong 已提交
526
#if 0
L
Liu Jicong 已提交
527 528 529 530 531 532 533 534 535 536
  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 已提交
537 538 539 540
      // 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 已提交
541 542 543 544
      if (mndSetTopicCommitLogs(pMnode, pTrans, &topicObj) != 0) {
        ASSERT(0);
        goto REB_FAIL;
      }
L
Liu Jicong 已提交
545 546
    }
  }
L
Liu Jicong 已提交
547
#endif
L
Liu Jicong 已提交
548

L
Liu Jicong 已提交
549 550
  // 4. TODO commit log: modification log

L
Liu Jicong 已提交
551
  // 5. set cb
552
  mndTransSetCb(pTrans, TRANS_START_FUNC_MQ_REB, TRANS_STOP_FUNC_MQ_REB, NULL, 0);
L
Liu Jicong 已提交
553 554

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

  mndTransDrop(pTrans);
  return 0;

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

S
Shengliang Guan 已提交
568 569 570
static int32_t mndProcessRebalanceReq(SRpcMsg *pMsg) {
  SMnode            *pMnode = pMsg->info.node;
  SMqDoRebalanceMsg *pReq = pMsg->pCont;
571 572 573 574 575 576 577
  void              *pIter = NULL;

  mInfo("mq rebalance start");

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

    SMqRebOutputObj rebOutput = {0};
L
Liu Jicong 已提交
581 582 583
    rebOutput.newConsumers = taosArrayInit(0, sizeof(int64_t));
    rebOutput.removedConsumers = taosArrayInit(0, sizeof(int64_t));
    rebOutput.touchedConsumers = taosArrayInit(0, sizeof(int64_t));
L
Liu Jicong 已提交
584 585
    rebOutput.rebVgs = taosArrayInit(0, sizeof(SMqRebOutputVg));

L
Liu Jicong 已提交
586 587 588 589
    SMqRebInfo      *pRebInfo = (SMqRebInfo *)pIter;
    SMqSubscribeObj *pSub = mndAcquireSubscribeByKey(pMnode, pRebInfo->key);

    rebInput.pRebInfo = pRebInfo;
590 591 592 593 594

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

L
Liu Jicong 已提交
604
      rebOutput.pSub = mndCreateSub(pMnode, pTopic, pRebInfo->key);
L
Liu Jicong 已提交
605
      memcpy(rebOutput.pSub->dbName, pTopic->db, TSDB_DB_FNAME_LEN);
L
Liu Jicong 已提交
606 607 608 609 610 611 612 613 614 615 616 617 618
      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);
    }
619

620 621 622
    if (mndDoRebalance(pMnode, &rebInput, &rebOutput) < 0) {
      mError("mq rebalance internal error");
    }
623

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

L
Liu Jicong 已提交
628
    if (mndPersistRebResult(pMnode, pMsg, &rebOutput) < 0) {
L
Liu Jicong 已提交
629
      mError("mq rebalance persist rebalance output error, possibly vnode splitted or dropped");
L
Liu Jicong 已提交
630
    }
L
Liu Jicong 已提交
631 632 633 634 635 636 637 638 639 640
    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);
641 642 643 644 645
  }

  // reset flag
  mInfo("mq rebalance completed successfully");
  taosHashCleanup(pReq->rebSubHash);
L
Liu Jicong 已提交
646
  mndRebEnd();
647 648 649

  return 0;
}
L
Liu Jicong 已提交
650

L
Liu Jicong 已提交
651
static int32_t mndProcessDropCgroupReq(SRpcMsg *pReq) {
L
Liu Jicong 已提交
652 653
  SMnode         *pMnode = pReq->info.node;
  SSdb           *pSdb = pMnode->pSdb;
L
Liu Jicong 已提交
654 655 656 657 658 659 660 661 662 663
  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) {
664
      mInfo("cgroup:%s on topic:%s, not exist, ignore not exist is set", dropReq.cgroup, dropReq.topic);
L
Liu Jicong 已提交
665 666 667 668 669 670 671 672
      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 已提交
673
  if (taosHashGetSize(pSub->consumerHash) != 0) {
L
Liu Jicong 已提交
674 675
    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 已提交
676
    mndReleaseSubscribe(pMnode, pSub);
L
Liu Jicong 已提交
677 678 679
    return -1;
  }

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

688
  mInfo("trans:%d, used to drop cgroup:%s on topic %s", pTrans->id, dropReq.cgroup, dropReq.topic);
L
Liu Jicong 已提交
689 690 691

  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 已提交
692
    mndReleaseSubscribe(pMnode, pSub);
L
Liu Jicong 已提交
693
    mndTransDrop(pTrans);
L
Liu Jicong 已提交
694 695 696
    return -1;
  }

L
Liu Jicong 已提交
697 698 699 700 701
  if (mndTransPrepare(pMnode, pTrans) < 0) {
    mndReleaseSubscribe(pMnode, pSub);
    mndTransDrop(pTrans);
    return -1;
  }
L
Liu Jicong 已提交
702 703 704 705 706
  mndReleaseSubscribe(pMnode, pSub);

  return TSDB_CODE_ACTION_IN_PROGRESS;
}

L
Liu Jicong 已提交
707 708 709
void mndCleanupSubscribe(SMnode *pMnode) {}

static SSdbRaw *mndSubActionEncode(SMqSubscribeObj *pSub) {
L
Liu Jicong 已提交
710
  terrno = TSDB_CODE_OUT_OF_MEMORY;
L
Liu Jicong 已提交
711
  void   *buf = NULL;
L
Liu Jicong 已提交
712
  int32_t tlen = tEncodeSubscribeObj(NULL, pSub);
L
Liu Jicong 已提交
713
  int32_t size = sizeof(int32_t) + tlen + MND_SUBSCRIBE_RESERVE_SIZE;
L
Liu Jicong 已提交
714 715 716 717

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

wafwerar's avatar
wafwerar 已提交
718
  buf = taosMemoryMalloc(tlen);
L
Liu Jicong 已提交
719
  if (buf == NULL) goto SUB_ENCODE_OVER;
L
Liu Jicong 已提交
720

L
Liu Jicong 已提交
721 722
  void *abuf = buf;
  tEncodeSubscribeObj(&abuf, pSub);
L
Liu Jicong 已提交
723 724 725 726 727 728 729

  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 已提交
730 731
  terrno = TSDB_CODE_SUCCESS;

L
Liu Jicong 已提交
732
SUB_ENCODE_OVER:
wafwerar's avatar
wafwerar 已提交
733
  taosMemoryFreeClear(buf);
L
Liu Jicong 已提交
734
  if (terrno != TSDB_CODE_SUCCESS) {
L
Liu Jicong 已提交
735 736 737 738 739 740 741 742 743 744 745
    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 已提交
746
  void *buf = NULL;
L
Liu Jicong 已提交
747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764

  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 已提交
765
  SDB_GET_INT32(pRaw, dataPos, &tlen, SUB_DECODE_OVER);
L
Liu Jicong 已提交
766
  buf = taosMemoryMalloc(tlen);
L
Liu Jicong 已提交
767 768 769 770 771 772 773 774
  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 已提交
775 776
  terrno = TSDB_CODE_SUCCESS;

L
Liu Jicong 已提交
777
SUB_DECODE_OVER:
wafwerar's avatar
wafwerar 已提交
778
  taosMemoryFreeClear(buf);
L
Liu Jicong 已提交
779
  if (terrno != TSDB_CODE_SUCCESS) {
L
Liu Jicong 已提交
780
    mError("subscribe:%s, failed to decode from raw:%p since %s", pSub->key, pRaw, terrstr());
wafwerar's avatar
wafwerar 已提交
781
    taosMemoryFreeClear(pRow);
L
Liu Jicong 已提交
782 783 784 785 786 787 788 789 790 791 792 793 794
    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);
795
  tDeleteSubscribeObj(pSub);
L
Liu Jicong 已提交
796 797 798 799 800
  return 0;
}

static int32_t mndSubActionUpdate(SSdb *pSdb, SMqSubscribeObj *pOldSub, SMqSubscribeObj *pNewSub) {
  mTrace("subscribe:%s, perform update action", pOldSub->key);
801 802 803 804 805 806
  taosWLockLatch(&pOldSub->lock);

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

807 808 809 810
  SArray *tmp1 = pOldSub->unassignedVgs;
  pOldSub->unassignedVgs = pNewSub->unassignedVgs;
  pNewSub->unassignedVgs = tmp1;

811
  taosWUnLockLatch(&pOldSub->lock);
L
Liu Jicong 已提交
812 813 814
  return 0;
}

815
int32_t mndMakeSubscribeKey(char *key, const char *cgroup, const char *topicName) {
S
Shengliang Guan 已提交
816
  int32_t tlen = strlen(cgroup);
L
Liu Jicong 已提交
817
  memcpy(key, cgroup, tlen);
L
Liu Jicong 已提交
818
  key[tlen] = TMQ_SEPARATOR;
L
Liu Jicong 已提交
819
  strcpy(key + tlen + 1, topicName);
L
Liu Jicong 已提交
820
  return 0;
L
Liu Jicong 已提交
821 822
}

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

L
Liu Jicong 已提交
834 835 836 837
SMqSubscribeObj *mndAcquireSubscribeByKey(SMnode *pMnode, const char *key) {
  SSdb            *pSdb = pMnode->pSdb;
  SMqSubscribeObj *pSub = sdbAcquire(pSdb, SDB_SUBSCRIBE, key);
  if (pSub == NULL) {
838
    terrno = TSDB_CODE_MND_SUBSCRIBE_NOT_EXIST;
L
Liu Jicong 已提交
839 840 841 842
  }
  return pSub;
}

L
Liu Jicong 已提交
843 844 845 846 847
void mndReleaseSubscribe(SMnode *pMnode, SMqSubscribeObj *pSub) {
  SSdb *pSdb = pMnode->pSdb;
  sdbRelease(pSdb, pSub);
}

L
Liu Jicong 已提交
848 849 850 851 852 853 854 855
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 已提交
856
int32_t mndSetDropSubCommitLogs(SMnode *pMnode, STrans *pTrans, SMqSubscribeObj *pSub) {
L
Liu Jicong 已提交
857 858 859 860 861 862 863 864
  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) {
865
  int32_t code = 0;
L
Liu Jicong 已提交
866 867 868 869 870 871 872 873 874 875 876 877 878 879
  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 已提交
880
      sdbRelease(pSdb, pSub);
881 882 883
      sdbCancelFetch(pSdb, pIter);
      code = -1;
      break;
L
Liu Jicong 已提交
884
    }
885 886

    sdbRelease(pSdb, pSub);
L
Liu Jicong 已提交
887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910
  }

  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
911 912
    if (taosHashGetSize(pSub->consumerHash) != 0) {
      sdbRelease(pSdb, pSub);
L
Liu Jicong 已提交
913
      terrno = TSDB_CODE_MND_IN_REBALANCE;
914 915
      return -1;
    }
L
Liu Jicong 已提交
916 917 918 919 920 921 922 923 924 925 926 927
    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 已提交
928
      action.msgType = TDMT_VND_TMQ_DELETE_SUB;
L
Liu Jicong 已提交
929 930 931 932 933
      if (mndTransAppendRedoAction(pTrans, &action) != 0) {
        taosMemoryFree(pReq);
        return -1;
      }
    }
L
Liu Jicong 已提交
934 935 936

    if (mndSetDropSubRedoLogs(pMnode, pTrans, pSub) < 0) {
      sdbRelease(pSdb, pSub);
L
Liu Jicong 已提交
937 938 939 940 941 942 943 944
      goto END;
    }
  }

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

S
Shengliang Guan 已提交
946 947
static int32_t mndRetrieveSubscribe(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rowsCapacity) {
  SMnode          *pMnode = pReq->info.node;
L
Liu Jicong 已提交
948 949 950 951
  SSdb            *pSdb = pMnode->pSdb;
  int32_t          numOfRows = 0;
  SMqSubscribeObj *pSub = NULL;

L
Liu Jicong 已提交
952 953
  mDebug("mnd show subscriptions begin");

L
Liu Jicong 已提交
954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980
  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 已提交
981
        mndSplitSubscribeKey(pSub->key, varDataVal(topic), varDataVal(cgroup), false);
L
Liu Jicong 已提交
982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998
        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);

L
Liu Jicong 已提交
999 1000 1001
        mDebug("mnd show subscrptions: topic %s, consumer %" PRId64 "cgroup %s vgid %d", varDataVal(topic),
               pConsumerEp->consumerId, varDataVal(cgroup), pVgEp->vgId);

L
Liu Jicong 已提交
1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016
        // 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 已提交
1017
    // do not show for cleared subscription
1018
#if 1
L
Liu Jicong 已提交
1019 1020 1021 1022 1023 1024 1025 1026 1027 1028
    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 已提交
1029
      mndSplitSubscribeKey(pSub->key, varDataVal(topic), varDataVal(cgroup), false);
L
Liu Jicong 已提交
1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046
      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);

L
Liu Jicong 已提交
1047 1048 1049
      mDebug("mnd show subscrptions(unassigned): topic %s, cgroup %s vgid %d", varDataVal(topic), varDataVal(cgroup),
             pVgEp->vgId);

L
Liu Jicong 已提交
1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063
      // 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 已提交
1064
#endif
L
Liu Jicong 已提交
1065 1066 1067 1068
    taosRUnLockLatch(&pSub->lock);
    sdbRelease(pSdb, pSub);
  }

L
Liu Jicong 已提交
1069 1070
  mDebug("mnd end show subscriptions");

L
Liu Jicong 已提交
1071 1072 1073 1074 1075 1076 1077 1078
  pShow->numOfRows += numOfRows;
  return numOfRows;
}

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