taosdata/TDengine | Build 3562 | source/dnode/mnode/impl/src/mndConsumer.c | Coveralls

56

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

71

int32_t mndSendConsumerMsg(SMnode *pMnode, int64_t consumerId, uint16_t msgType, SRpcHandleInfo *info) {

×

UNCOV

72

  if (pMnode == NULL || info == NULL) {

×

73

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

75

  int32_t code = 0;

×

UNCOV

76

  void   *msg  = rpcMallocCont(sizeof(int64_t));

×

UNCOV

77

  MND_TMQ_NULL_CHECK(msg);

×

UNCOV

79

  *(int64_t*)msg = consumerId;

×

UNCOV

80

  SRpcMsg rpcMsg = {

×

UNCOV

87

  mInfo("mndSendConsumerMsg type:%d consumer:0x%" PRIx64, msgType, consumerId);

×

UNCOV

88

  MND_TMQ_RETURN_CHECK(tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg));

×

UNCOV

89

  return code;

×

91

END:

×

92

  taosMemoryFree(msg);

×

93

  return code;

×

UNCOV

96

static int32_t validateTopics(STrans* pTrans, SCMSubscribeReq *subscribe, SMnode *pMnode, const char *pUser) {

×

UNCOV

97

  if (pTrans == NULL || subscribe == NULL || pMnode == NULL || pUser == NULL) {

×

98

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

100

  SMqTopicObj *pTopic = NULL;

×

UNCOV

101

  int32_t      code = 0;

×

UNCOV

103

  int32_t numOfTopics = taosArrayGetSize(subscribe->topicNames);

×

UNCOV

104

  for (int32_t i = 0; i < numOfTopics; i++) {

×

UNCOV

105

    char *pOneTopic = taosArrayGetP(subscribe->topicNames, i);

×

UNCOV

106

    MND_TMQ_RETURN_CHECK(mndAcquireTopic(pMnode, pOneTopic, &pTopic));

×

UNCOV

107

    MND_TMQ_RETURN_CHECK(mndCheckTopicPrivilege(pMnode, pUser, MND_OPER_SUBSCRIBE, pTopic));

×

UNCOV

108

    MND_TMQ_RETURN_CHECK(grantCheckExpire(TSDB_GRANT_SUBSCRIPTION));

×

UNCOV

110

    if (subscribe->enableReplay) {

×

UNCOV

111

      if (pTopic->subType != TOPIC_SUB_TYPE__COLUMN) {

×

UNCOV

112

        code = TSDB_CODE_TMQ_REPLAY_NOT_SUPPORT;

×

UNCOV

113

        goto END;

×

UNCOV

114

      } else if (pTopic->ntbUid == 0 && pTopic->ctbStbUid == 0) {

×

UNCOV

115

        SDbObj *pDb = mndAcquireDb(pMnode, pTopic->db);

×

UNCOV

116

        if (pDb == NULL) {

×

117

          code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

118

          goto END;

×

UNCOV

120

        if (pDb->cfg.numOfVgroups != 1) {

×

UNCOV

121

          mndReleaseDb(pMnode, pDb);

×

UNCOV

122

          code = TSDB_CODE_TMQ_REPLAY_NEED_ONE_VGROUP;

×

UNCOV

123

          goto END;

×

UNCOV

125

        mndReleaseDb(pMnode, pDb);

×

UNCOV

128

    char  key[TSDB_CONSUMER_ID_LEN] = {0};

×

UNCOV

129

    (void)snprintf(key, TSDB_CONSUMER_ID_LEN, "%"PRIx64, subscribe->consumerId);

×

UNCOV

130

    mndTransSetDbName(pTrans, pTopic->db, key);

×

UNCOV

131

    MND_TMQ_RETURN_CHECK(mndTransCheckConflict(pMnode, pTrans));

×

UNCOV

132

    mndReleaseTopic(pMnode, pTopic);

×

UNCOV

134

  return 0;

×

UNCOV

136

END:

×

UNCOV

137

  mndReleaseTopic(pMnode, pTopic);

×

UNCOV

138

  return code;

×

UNCOV

141

static int32_t mndProcessConsumerClearMsg(SRpcMsg *pMsg) {

×

UNCOV

142

  if (pMsg == NULL || pMsg->pCont == NULL) {

×

143

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

145

  int32_t              code = 0;

×

UNCOV

146

  SMnode              *pMnode = pMsg->info.node;

×

UNCOV

147

  SMqConsumerClearMsg *pClearMsg = pMsg->pCont;

×

UNCOV

148

  SMqConsumerObj      *pConsumerNew = NULL;

×

UNCOV

149

  STrans              *pTrans = NULL;

×

UNCOV

150

  SMqConsumerObj      *pConsumer = NULL;

×

UNCOV

152

  MND_TMQ_RETURN_CHECK(mndAcquireConsumer(pMnode, pClearMsg->consumerId, &pConsumer));

×

UNCOV

153

  mInfo("consumer:0x%" PRIx64 " needs to be cleared, status %s", pClearMsg->consumerId,

×

UNCOV

156

  MND_TMQ_RETURN_CHECK(tNewSMqConsumerObj(pConsumer->consumerId, pConsumer->cgroup, -1, NULL, NULL, &pConsumerNew));

×

UNCOV

157

  pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_NOTHING, pMsg, "clear-csm");

×

UNCOV

158

  MND_TMQ_NULL_CHECK(pTrans);

×

UNCOV

159

  MND_TMQ_RETURN_CHECK(mndSetConsumerDropLogs(pTrans, pConsumerNew));

×

UNCOV

160

  code = mndTransPrepare(pMnode, pTrans);

×

UNCOV

162

END:

×

UNCOV

163

  mndReleaseConsumer(pMnode, pConsumer);

×

UNCOV

164

  tDeleteSMqConsumerObj(pConsumerNew);

×

UNCOV

165

  mndTransDrop(pTrans);

×

UNCOV

166

  return code;

×

UNCOV

169

static int32_t checkPrivilege(SMnode *pMnode, SMqConsumerObj *pConsumer, SMqHbRsp *rsp, char *user) {

×

UNCOV

170

  if (pMnode == NULL || pConsumer == NULL || rsp == NULL || user == NULL) {

×

171

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

173

  int32_t code = 0;

×

UNCOV

174

  rsp->topicPrivileges = taosArrayInit(taosArrayGetSize(pConsumer->currentTopics), sizeof(STopicPrivilege));

×

UNCOV

175

  MND_TMQ_NULL_CHECK(rsp->topicPrivileges);

×

UNCOV

176

  for (int32_t i = 0; i < taosArrayGetSize(pConsumer->currentTopics); i++) {

×

UNCOV

177

    char        *topic = taosArrayGetP(pConsumer->currentTopics, i);

×

UNCOV

178

    SMqTopicObj *pTopic = NULL;

×

UNCOV

179

    code = mndAcquireTopic(pMnode, topic, &pTopic);

×

UNCOV

180

    if (code != TDB_CODE_SUCCESS) {

×

181

      continue;

×

UNCOV

183

    STopicPrivilege *data = taosArrayReserve(rsp->topicPrivileges, 1);

×

UNCOV

184

    MND_TMQ_NULL_CHECK(data);

×

UNCOV

185

    tstrncpy(data->topic, topic, TSDB_TOPIC_FNAME_LEN);

×

UNCOV

186

    if (mndCheckTopicPrivilege(pMnode, user, MND_OPER_SUBSCRIBE, pTopic) != 0 ||

×

UNCOV

187

        grantCheckExpire(TSDB_GRANT_SUBSCRIPTION) < 0) {

×

188

      data->noPrivilege = 1;

×

UNCOV

190

      data->noPrivilege = 0;

×

UNCOV

192

    mndReleaseTopic(pMnode, pTopic);

×

UNCOV

194

END:

×

UNCOV

195

  return code;

×

UNCOV

198

static void storeOffsetRows(SMnode *pMnode, SMqHbReq *req, SMqConsumerObj *pConsumer){

×

UNCOV

199

  if (pMnode == NULL || req == NULL || pConsumer == NULL){

×

200

    return;

×

UNCOV

202

  for (int i = 0; i < taosArrayGetSize(req->topics); i++) {

×

UNCOV

203

    TopicOffsetRows *data = taosArrayGet(req->topics, i);

×

UNCOV

204

    if (data == NULL){

×

205

      continue;

×

UNCOV

207

    mInfo("heartbeat report offset rows.%s:%s", pConsumer->cgroup, data->topicName);

×

UNCOV

209

    SMqSubscribeObj *pSub = NULL;

×

UNCOV

210

    char  key[TSDB_SUBSCRIBE_KEY_LEN] = {0};

×

UNCOV

211

    (void)snprintf(key, TSDB_SUBSCRIBE_KEY_LEN, "%s%s%s", pConsumer->cgroup, TMQ_SEPARATOR, data->topicName);

×

UNCOV

212

    int32_t code = mndAcquireSubscribeByKey(pMnode, key, &pSub);

×

UNCOV

213

    if (code != 0) {

×

214

      mError("failed to acquire subscribe by key:%s, code:%d", key, code);

×

215

      continue;

×

UNCOV

217

    taosWLockLatch(&pSub->lock);

×

UNCOV

218

    SMqConsumerEp *pConsumerEp = taosHashGet(pSub->consumerHash, &pConsumer->consumerId, sizeof(int64_t));

×

UNCOV

219

    if (pConsumerEp) {

×

UNCOV

220

      taosArrayDestroy(pConsumerEp->offsetRows);

×

UNCOV

221

      pConsumerEp->offsetRows = data->offsetRows;

×

UNCOV

222

      data->offsetRows = NULL;

×

UNCOV

224

    taosWUnLockLatch(&pSub->lock);

×

UNCOV

226

    mndReleaseSubscribe(pMnode, pSub);

×

UNCOV

230

static int32_t buildMqHbRsp(SRpcMsg *pMsg, SMqHbRsp *rsp){

×

UNCOV

231

  if (pMsg == NULL || rsp == NULL){

×

232

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

234

  int32_t tlen = tSerializeSMqHbRsp(NULL, 0, rsp);

×

UNCOV

235

  if (tlen <= 0){

×

236

    return TSDB_CODE_TMQ_INVALID_MSG;

×

UNCOV

238

  void   *buf = rpcMallocCont(tlen);

×

UNCOV

239

  if (buf == NULL) {

×

240

    return terrno;

×

UNCOV

243

  if(tSerializeSMqHbRsp(buf, tlen, rsp) <= 0){

×

244

    rpcFreeCont(buf);

×

245

    return TSDB_CODE_TMQ_INVALID_MSG;

×

UNCOV

247

  pMsg->info.rsp = buf;

×

UNCOV

248

  pMsg->info.rspLen = tlen;

×

UNCOV

249

  return 0;

×

UNCOV

252

static int32_t mndProcessMqHbReq(SRpcMsg *pMsg) {

×

UNCOV

253

  if (pMsg == NULL) {

×

254

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

256

  int32_t         code = 0;

×

UNCOV

257

  SMnode         *pMnode = pMsg->info.node;

×

UNCOV

258

  SMqHbReq        req = {0};

×

UNCOV

259

  SMqHbRsp        rsp = {0};

×

UNCOV

260

  SMqConsumerObj *pConsumer = NULL;

×

UNCOV

261

  MND_TMQ_RETURN_CHECK(tDeserializeSMqHbReq(pMsg->pCont, pMsg->contLen, &req));

×

UNCOV

262

  int64_t consumerId = req.consumerId;

×

UNCOV

263

  MND_TMQ_RETURN_CHECK(mndAcquireConsumer(pMnode, consumerId, &pConsumer));

×

UNCOV

264

  MND_TMQ_RETURN_CHECK(checkPrivilege(pMnode, pConsumer, &rsp, pMsg->info.conn.user));

×

UNCOV

265

  atomic_store_32(&pConsumer->hbStatus, 0);

×

UNCOV

266

  mDebug("consumer:0x%" PRIx64 " receive hb pollFlag:%d %d", consumerId, req.pollFlag, pConsumer->pollStatus);

×

UNCOV

267

  if (req.pollFlag == 1){

×

UNCOV

268

    atomic_store_32(&pConsumer->pollStatus, 0);

×

UNCOV

271

  storeOffsetRows(pMnode, &req, pConsumer);

×

UNCOV

272

  rsp.debugFlag = tqClientDebugFlag;

×

UNCOV

273

  code = buildMqHbRsp(pMsg, &rsp);

×

UNCOV

275

END:

×

UNCOV

276

  tDestroySMqHbRsp(&rsp);

×

UNCOV

277

  mndReleaseConsumer(pMnode, pConsumer);

×

UNCOV

278

  tDestroySMqHbReq(&req);

×

UNCOV

279

  return code;

×

UNCOV

282

static int32_t addEpSetInfo(SMnode *pMnode, SMqConsumerObj *pConsumer, int32_t epoch, SMqAskEpRsp *rsp){

×

UNCOV

283

  if (pMnode == NULL || pConsumer == NULL || rsp == NULL){

×

284

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

286

  taosRLockLatch(&pConsumer->lock);

×

UNCOV

288

  int32_t numOfTopics = taosArrayGetSize(pConsumer->currentTopics);

×

UNCOV

290

  rsp->topics = taosArrayInit(numOfTopics, sizeof(SMqSubTopicEp));

×

UNCOV

291

  if (rsp->topics == NULL) {

×

292

    taosRUnLockLatch(&pConsumer->lock);

×

293

    return terrno;

×

UNCOV

297

  for (int32_t i = 0; i < numOfTopics; i++) {

×

UNCOV

298

    char            *topic = taosArrayGetP(pConsumer->currentTopics, i);

×

UNCOV

299

    SMqSubscribeObj *pSub = NULL;

×

UNCOV

300

    char  key[TSDB_SUBSCRIBE_KEY_LEN] = {0};

×

UNCOV

301

    (void)snprintf(key, TSDB_SUBSCRIBE_KEY_LEN, "%s%s%s", pConsumer->cgroup, TMQ_SEPARATOR, topic);

×

UNCOV

302

    int32_t code = mndAcquireSubscribeByKey(pMnode, key, &pSub);

×

UNCOV

303

    if (code != 0) {

×

304

      continue;

×

UNCOV

306

    taosRLockLatch(&pSub->lock);

×

UNCOV

308

    SMqSubTopicEp topicEp = {0};

×

UNCOV

309

    tstrncpy(topicEp.topic, topic, TSDB_TOPIC_FNAME_LEN);

×

UNCOV

312

    SMqTopicObj *pTopic = NULL;

×

UNCOV

313

    code = mndAcquireTopic(pMnode, topic, &pTopic);

×

UNCOV

314

    if (code != TDB_CODE_SUCCESS) {

×

315

      taosRUnLockLatch(&pSub->lock);

×

316

      mndReleaseSubscribe(pMnode, pSub);

×

317

      continue;

×

UNCOV

319

    taosRLockLatch(&pTopic->lock);

×

UNCOV

320

    tstrncpy(topicEp.db, pTopic->db, TSDB_DB_FNAME_LEN);

×

UNCOV

321

    topicEp.schema.nCols = pTopic->schema.nCols;

×

UNCOV

322

    if (topicEp.schema.nCols) {

×

UNCOV

323

      topicEp.schema.pSchema = taosMemoryCalloc(topicEp.schema.nCols, sizeof(SSchema));

×

UNCOV

324

      if (topicEp.schema.pSchema == NULL) {

×

325

        taosRUnLockLatch(&pTopic->lock);

×

326

        taosRUnLockLatch(&pSub->lock);

×

327

        mndReleaseSubscribe(pMnode, pSub);

×

328

        mndReleaseTopic(pMnode, pTopic);

×

329

        return terrno;

×

UNCOV

331

      (void)memcpy(topicEp.schema.pSchema, pTopic->schema.pSchema, topicEp.schema.nCols * sizeof(SSchema));

×

UNCOV

333

    taosRUnLockLatch(&pTopic->lock);

×

UNCOV

334

    mndReleaseTopic(pMnode, pTopic);

×

UNCOV

337

    SMqConsumerEp *pConsumerEp = taosHashGet(pSub->consumerHash, &pConsumer->consumerId, sizeof(int64_t));

×

UNCOV

338

    if (pConsumerEp == NULL) {

×

339

      taosRUnLockLatch(&pConsumer->lock);

×

340

      taosRUnLockLatch(&pSub->lock);

×

341

      mndReleaseSubscribe(pMnode, pSub);

×

342

      return TSDB_CODE_OUT_OF_MEMORY;

×

UNCOV

344

    int32_t vgNum = taosArrayGetSize(pConsumerEp->vgs);

×

UNCOV

345

    topicEp.vgs = taosArrayInit(vgNum, sizeof(SMqSubVgEp));

×

UNCOV

346

    if (topicEp.vgs == NULL) {

×

347

      taosRUnLockLatch(&pConsumer->lock);

×

348

      taosRUnLockLatch(&pSub->lock);

×

349

      mndReleaseSubscribe(pMnode, pSub);

×

350

      return terrno;

×

UNCOV

353

    for (int32_t j = 0; j < vgNum; j++) {

×

UNCOV

354

      SMqVgEp *pVgEp = taosArrayGetP(pConsumerEp->vgs, j);

×

UNCOV

355

      if (pVgEp == NULL) {

×

356

        continue;

×

UNCOV

358

      if (epoch == -1) {

×

UNCOV

359

        SVgObj *pVgroup = mndAcquireVgroup(pMnode, pVgEp->vgId);

×

UNCOV

360

        if (pVgroup) {

×

UNCOV

361

          pVgEp->epSet = mndGetVgroupEpset(pMnode, pVgroup);

×

UNCOV

362

          mndReleaseVgroup(pMnode, pVgroup);

×

UNCOV

365

      SMqSubVgEp vgEp = {.epSet = pVgEp->epSet, .vgId = pVgEp->vgId, .offset = -1};

×

UNCOV

366

      if (taosArrayPush(topicEp.vgs, &vgEp) == NULL) {

×

367

        taosArrayDestroy(topicEp.vgs);

×

368

        taosRUnLockLatch(&pConsumer->lock);

×

369

        taosRUnLockLatch(&pSub->lock);

×

370

        mndReleaseSubscribe(pMnode, pSub);

×

371

        return terrno;

×

UNCOV

374

    if (taosArrayPush(rsp->topics, &topicEp) == NULL) {

×

375

      taosArrayDestroy(topicEp.vgs);

×

376

      taosRUnLockLatch(&pConsumer->lock);

×

377

      taosRUnLockLatch(&pSub->lock);

×

378

      mndReleaseSubscribe(pMnode, pSub);

×

379

      return terrno;

×

UNCOV

381

    taosRUnLockLatch(&pSub->lock);

×

UNCOV

382

    mndReleaseSubscribe(pMnode, pSub);

×

UNCOV

384

  taosRUnLockLatch(&pConsumer->lock);

×

UNCOV

385

  return 0;

×

UNCOV

388

static int32_t buildAskEpRsp(SRpcMsg *pMsg, SMqAskEpRsp *rsp, int32_t serverEpoch, int64_t consumerId){

×

UNCOV

389

  if (pMsg == NULL || rsp == NULL) {

×

390

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

392

  int32_t code = 0;

×

UNCOV

394

  int32_t tlen = sizeof(SMqRspHead) + tEncodeSMqAskEpRsp(NULL, rsp);

×

UNCOV

395

  void   *buf = rpcMallocCont(tlen);

×

UNCOV

396

  if (buf == NULL) {

×

397

    return terrno;

×

UNCOV

400

  SMqRspHead *pHead = buf;

×

UNCOV

402

  pHead->mqMsgType = TMQ_MSG_TYPE__EP_RSP;

×

UNCOV

403

  pHead->epoch = serverEpoch;

×

UNCOV

404

  pHead->consumerId = consumerId;

×

UNCOV

405

  pHead->walsver = 0;

×

UNCOV

406

  pHead->walever = 0;

×

UNCOV

408

  void *abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));

×

UNCOV

409

  if (tEncodeSMqAskEpRsp(&abuf, rsp) < 0) {

×

410

    rpcFreeCont(buf);

×

411

    return TSDB_CODE_TSC_INTERNAL_ERROR;

×

UNCOV

415

  pMsg->info.rsp = buf;

×

UNCOV

416

  pMsg->info.rspLen = tlen;

×

UNCOV

417

  return code;

×

UNCOV

420

static int32_t mndProcessAskEpReq(SRpcMsg *pMsg) {

×

UNCOV

421

  if (pMsg == NULL) {

×

422

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

424

  SMnode     *pMnode = pMsg->info.node;

×

UNCOV

425

  SMqAskEpReq req = {0};

×

UNCOV

426

  SMqAskEpRsp rsp = {0};

×

UNCOV

427

  int32_t     code = 0;

×

UNCOV

428

  SMqConsumerObj *pConsumer = NULL;

×

UNCOV

430

  MND_TMQ_RETURN_CHECK(tDeserializeSMqAskEpReq(pMsg->pCont, pMsg->contLen, &req));

×

UNCOV

431

  int64_t consumerId = req.consumerId;

×

UNCOV

432

  MND_TMQ_RETURN_CHECK(mndAcquireConsumer(pMnode, consumerId, &pConsumer));

×

UNCOV

433

  if (strncmp(req.cgroup, pConsumer->cgroup, tListLen(pConsumer->cgroup)) != 0) {

×

434

    mError("consumer:0x%" PRIx64 " group:%s not consistent with data in sdb, saved cgroup:%s", consumerId, req.cgroup,

×

436

    code = TSDB_CODE_MND_CONSUMER_NOT_EXIST;

×

437

    goto END;

×

UNCOV

441

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

×

UNCOV

442

  if (status != MQ_CONSUMER_STATUS_READY) {

×

UNCOV

443

    mInfo("consumer:0x%" PRIx64 " not ready, status: %s", consumerId, mndConsumerStatusName(status));

×

UNCOV

444

    code = TSDB_CODE_MND_CONSUMER_NOT_READY;

×

UNCOV

445

    goto END;

×

UNCOV

448

  int32_t epoch = req.epoch;

×

UNCOV

449

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

×

UNCOV

452

  if (epoch != serverEpoch) {

×

UNCOV

453

    mInfo("process ask ep, consumer:0x%" PRIx64 "(epoch %d) update with server epoch %d",

×

UNCOV

455

    MND_TMQ_RETURN_CHECK(addEpSetInfo(pMnode, pConsumer, epoch, &rsp));

×

UNCOV

458

  code = buildAskEpRsp(pMsg, &rsp, serverEpoch, consumerId);

×

UNCOV

460

END:

×

UNCOV

462

  mndReleaseConsumer(pMnode, pConsumer);

×

UNCOV

463

  return code;

×

UNCOV

466

int32_t mndSetConsumerDropLogs(STrans *pTrans, SMqConsumerObj *pConsumer) {

×

UNCOV

467

  if (pConsumer == NULL || pTrans == NULL) {

×

468

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

470

  int32_t  code = 0;

×

UNCOV

471

  SSdbRaw *pCommitRaw = mndConsumerActionEncode(pConsumer);

×

UNCOV

472

  MND_TMQ_NULL_CHECK(pCommitRaw);

×

UNCOV

473

  code = mndTransAppendCommitlog(pTrans, pCommitRaw);

×

UNCOV

474

  if (code != 0) {

×

475

    sdbFreeRaw(pCommitRaw);

×

476

    goto END;

×

UNCOV

478

  MND_TMQ_RETURN_CHECK(sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED));

×

UNCOV

479

END:

×

UNCOV

480

  return code;

×

UNCOV

483

int32_t mndSetConsumerCommitLogs(STrans *pTrans, SMqConsumerObj *pConsumer) {

×

UNCOV

484

  if (pConsumer == NULL || pTrans == NULL) {

×

485

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

487

  int32_t  code = 0;

×

UNCOV

488

  SSdbRaw *pCommitRaw = mndConsumerActionEncode(pConsumer);

×

UNCOV

489

  MND_TMQ_NULL_CHECK(pCommitRaw);

×

UNCOV

490

  code = mndTransAppendCommitlog(pTrans, pCommitRaw);

×

UNCOV

491

  if (code != 0) {

×

492

    sdbFreeRaw(pCommitRaw);

×

493

    goto END;

×

UNCOV

495

  MND_TMQ_RETURN_CHECK(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY));

×

UNCOV

496

END:

×

UNCOV

497

  return code;

×

UNCOV

500

static void freeItem(void *param) {

×

UNCOV

501

  if (param == NULL) {

×

502

    return;

×

UNCOV

504

  void *pItem = *(void **)param;

×

UNCOV

505

  if (pItem != NULL) {

×

UNCOV

506

    taosMemoryFree(pItem);

×

UNCOV

519

static int32_t getTopicAddDelete(SMqConsumerObj *pExistedConsumer, SMqConsumerObj *pConsumerNew){

×

UNCOV

520

  if (pExistedConsumer == NULL || pConsumerNew == NULL) {

×

521

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

523

  int32_t code = 0;

×

UNCOV

524

  pConsumerNew->rebNewTopics = taosArrayInit(0, sizeof(void *));

×

UNCOV

525

  MND_TMQ_NULL_CHECK(pConsumerNew->rebNewTopics);

×

UNCOV

526

  pConsumerNew->rebRemovedTopics = taosArrayInit(0, sizeof(void *));

×

UNCOV

527

  MND_TMQ_NULL_CHECK(pConsumerNew->rebRemovedTopics);

×

UNCOV

529

  int32_t newTopicNum = taosArrayGetSize(pConsumerNew->assignedTopics);

×

UNCOV

530

  int32_t oldTopicNum = taosArrayGetSize(pExistedConsumer->currentTopics);

×

UNCOV

531

  int32_t i = 0, j = 0;

×

UNCOV

532

  while (i < oldTopicNum || j < newTopicNum) {

×

UNCOV

533

    if (i >= oldTopicNum) {

×

UNCOV

534

      void* tmp = taosArrayGetP(pConsumerNew->assignedTopics, j);

×

UNCOV

535

      MND_TMQ_NULL_CHECK(tmp);

×

UNCOV

536

      ADD_TOPIC_TO_ARRAY(tmp, rebNewTopics);

×

UNCOV

537

      j++;

×

UNCOV

538

      continue;

×

UNCOV

539

    } else if (j >= newTopicNum) {

×

UNCOV

540

      void* tmp = taosArrayGetP(pExistedConsumer->currentTopics, i);

×

UNCOV

541

      MND_TMQ_NULL_CHECK(tmp);

×

UNCOV

542

      ADD_TOPIC_TO_ARRAY(tmp, rebRemovedTopics);

×

UNCOV

543

      i++;

×

UNCOV

544

      continue;

×

UNCOV

546

      char *oldTopic = taosArrayGetP(pExistedConsumer->currentTopics, i);

×

UNCOV

547

      MND_TMQ_NULL_CHECK(oldTopic);

×

UNCOV

548

      char *newTopic = taosArrayGetP(pConsumerNew->assignedTopics, j);

×

UNCOV

549

      MND_TMQ_NULL_CHECK(newTopic);

×

UNCOV

550

      int   comp = strcmp(oldTopic, newTopic);

×

UNCOV

551

      if (comp == 0) {

×

UNCOV

552

        i++;

×

UNCOV

553

        j++;

×

UNCOV

554

        continue;

×

555

      } else if (comp < 0) {

×

556

        ADD_TOPIC_TO_ARRAY(oldTopic, rebRemovedTopics);

×

557

        i++;

×

558

        continue;

×

559

      } else {

×

560

        ADD_TOPIC_TO_ARRAY(newTopic, rebNewTopics);

×

561

        j++;

×

562

        continue;

×

UNCOV

567

  if (taosArrayGetSize(pConsumerNew->rebNewTopics) == 0 && taosArrayGetSize(pConsumerNew->rebRemovedTopics) == 0) {

×

UNCOV

568

    code = TSDB_CODE_TMQ_NO_NEED_REBALANCE;

×

UNCOV

571

END:

×

UNCOV

572

  return code;

×

UNCOV

575

static int32_t checkAndSortTopic(SMnode *pMnode, SArray *pTopicList){

×

UNCOV

576

  if (pTopicList == NULL || pMnode == NULL) {

×

577

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

579

  taosArraySort(pTopicList, taosArrayCompareString);

×

UNCOV

580

  taosArrayRemoveDuplicate(pTopicList, taosArrayCompareString, freeItem);

×

UNCOV

582

  int32_t newTopicNum = taosArrayGetSize(pTopicList);

×

UNCOV

583

  for (int i = 0; i < newTopicNum; i++) {

×

UNCOV

584

    int32_t gNum = mndGetGroupNumByTopic(pMnode, (const char *)taosArrayGetP(pTopicList, i));

×

UNCOV

585

    if (gNum >= MND_MAX_GROUP_PER_TOPIC) {

×

586

      return TSDB_CODE_TMQ_GROUP_OUT_OF_RANGE;

×

UNCOV

589

  return 0;

×

UNCOV

592

static int32_t buildSubConsumer(SMnode *pMnode, SCMSubscribeReq *subscribe, SMqConsumerObj** ppConsumer){

×

UNCOV

593

  if (pMnode == NULL || subscribe == NULL) {

×

594

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

596

  int64_t         consumerId = subscribe->consumerId;

×

UNCOV

597

  char           *cgroup     = subscribe->cgroup;

×

UNCOV

598

  SMqConsumerObj *pConsumerNew     = NULL;

×

UNCOV

599

  SMqConsumerObj *pExistedConsumer = NULL;

×

UNCOV

600

  int32_t code = mndAcquireConsumer(pMnode, consumerId, &pExistedConsumer);

×

UNCOV

601

  if (code != 0) {

×

UNCOV

602

    mInfo("receive subscribe request from new consumer:0x%" PRIx64

×

UNCOV

606

    MND_TMQ_RETURN_CHECK(tNewSMqConsumerObj(consumerId, cgroup, CONSUMER_INSERT_SUB, NULL, subscribe, &pConsumerNew));

×

UNCOV

608

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

×

UNCOV

610

    mInfo("receive subscribe request from existed consumer:0x%" PRIx64

×

UNCOV

615

    if (status != MQ_CONSUMER_STATUS_READY) {

×

616

      code = TSDB_CODE_MND_CONSUMER_NOT_READY;

×

617

      goto END;

×

UNCOV

619

    MND_TMQ_RETURN_CHECK(tNewSMqConsumerObj(consumerId, cgroup, CONSUMER_UPDATE_SUB, NULL, subscribe, &pConsumerNew));

×

UNCOV

620

    MND_TMQ_RETURN_CHECK(getTopicAddDelete(pExistedConsumer, pConsumerNew));

×

UNCOV

622

  mndReleaseConsumer(pMnode, pExistedConsumer);

×

UNCOV

623

  if (ppConsumer){

×

UNCOV

624

    *ppConsumer = pConsumerNew;

×

UNCOV

626

  return code;

×

UNCOV

628

END:

×

UNCOV

629

  mndReleaseConsumer(pMnode, pExistedConsumer);

×

UNCOV

630

  tDeleteSMqConsumerObj(pConsumerNew);

×

UNCOV

631

  return code;

×

UNCOV

634

int32_t mndProcessSubscribeReq(SRpcMsg *pMsg) {

×

UNCOV

635

  if (pMsg == NULL) {

×

636

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

638

  SMnode *pMnode = pMsg->info.node;

×

UNCOV

639

  char   *msgStr = pMsg->pCont;

×

UNCOV

640

  int32_t code = 0;

×

UNCOV

641

  SMqConsumerObj *pConsumerNew = NULL;

×

UNCOV

642

  STrans         *pTrans = NULL;

×

UNCOV

644

  SCMSubscribeReq subscribe = {0};

×

UNCOV

645

  MND_TMQ_RETURN_CHECK(tDeserializeSCMSubscribeReq(msgStr, &subscribe, pMsg->contLen));

×

UNCOV

646

  bool unSubscribe = (taosArrayGetSize(subscribe.topicNames) == 0);

×

UNCOV

647

  if(unSubscribe){

×

UNCOV

648

    SMqConsumerObj *pConsumerTmp = NULL;

×

UNCOV

649

    MND_TMQ_RETURN_CHECK(mndAcquireConsumer(pMnode, subscribe.consumerId, &pConsumerTmp));

×

UNCOV

650

    if (taosArrayGetSize(pConsumerTmp->assignedTopics) == 0){

×

UNCOV

651

      mndReleaseConsumer(pMnode, pConsumerTmp);

×

UNCOV

652

      goto END;

×

UNCOV

654

    mndReleaseConsumer(pMnode, pConsumerTmp);

×

UNCOV

656

  MND_TMQ_RETURN_CHECK(checkAndSortTopic(pMnode, subscribe.topicNames));

×

UNCOV

657

  pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY,

×

UNCOV

660

  MND_TMQ_NULL_CHECK(pTrans);

×

UNCOV

662

  MND_TMQ_RETURN_CHECK(validateTopics(pTrans, &subscribe, pMnode, pMsg->info.conn.user));

×

UNCOV

663

  MND_TMQ_RETURN_CHECK(buildSubConsumer(pMnode, &subscribe, &pConsumerNew));

×

UNCOV

664

  MND_TMQ_RETURN_CHECK(mndSetConsumerCommitLogs(pTrans, pConsumerNew));

×

UNCOV

665

  MND_TMQ_RETURN_CHECK(mndTransPrepare(pMnode, pTrans));

×

UNCOV

666

  code = TSDB_CODE_ACTION_IN_PROGRESS;

×

UNCOV

668

END:

×

UNCOV

669

  mndTransDrop(pTrans);

×

UNCOV

670

  tDeleteSMqConsumerObj(pConsumerNew);

×

UNCOV

671

  taosArrayDestroyP(subscribe.topicNames, NULL);

×

UNCOV

672

  return (code == TSDB_CODE_TMQ_NO_NEED_REBALANCE || code == TSDB_CODE_MND_CONSUMER_NOT_EXIST) ? 0 : code;

×

UNCOV

675

SSdbRaw *mndConsumerActionEncode(SMqConsumerObj *pConsumer) {

×

UNCOV

676

  if (pConsumer == NULL) {

×

677

    return NULL;

×

UNCOV

679

  int32_t code = 0;

×

UNCOV

680

  int32_t lino = 0;

×

UNCOV

681

  terrno = TSDB_CODE_OUT_OF_MEMORY;

×

UNCOV

683

  void   *buf = NULL;

×

UNCOV

684

  int32_t tlen = tEncodeSMqConsumerObj(NULL, pConsumer);

×

UNCOV

685

  int32_t size = sizeof(int32_t) + tlen + MND_CONSUMER_RESERVE_SIZE;

×

UNCOV

687

  SSdbRaw *pRaw = sdbAllocRaw(SDB_CONSUMER, MND_CONSUMER_VER_NUMBER, size);

×

UNCOV

688

  if (pRaw == NULL) goto CM_ENCODE_OVER;

×

UNCOV

690

  buf = taosMemoryMalloc(tlen);

×

UNCOV

691

  if (buf == NULL) goto CM_ENCODE_OVER;

×

UNCOV

693

  void *abuf = buf;

×

UNCOV

694

  if(tEncodeSMqConsumerObj(&abuf, pConsumer) < 0){

×

695

    goto CM_ENCODE_OVER;

×

UNCOV

698

  int32_t dataPos = 0;

×

UNCOV

699

  SDB_SET_INT32(pRaw, dataPos, tlen, CM_ENCODE_OVER);

×

UNCOV

700

  SDB_SET_BINARY(pRaw, dataPos, buf, tlen, CM_ENCODE_OVER);

×

UNCOV

701

  SDB_SET_RESERVE(pRaw, dataPos, MND_CONSUMER_RESERVE_SIZE, CM_ENCODE_OVER);

×

UNCOV

702

  SDB_SET_DATALEN(pRaw, dataPos, CM_ENCODE_OVER);

×

UNCOV

704

  terrno = TSDB_CODE_SUCCESS;

×

UNCOV

706

CM_ENCODE_OVER:

×

UNCOV

707

  taosMemoryFreeClear(buf);

×

UNCOV

708

  if (terrno != 0) {

×

709

    mError("consumer:0x%" PRIx64 " failed to encode to raw:%p since %s", pConsumer->consumerId, pRaw, terrstr());

×

710

    sdbFreeRaw(pRaw);

×

711

    return NULL;

×

UNCOV

714

  mTrace("consumer:0x%" PRIx64 ", encode to raw:%p, row:%p", pConsumer->consumerId, pRaw, pConsumer);

×

UNCOV

715

  return pRaw;

×

UNCOV

718

SSdbRow *mndConsumerActionDecode(SSdbRaw *pRaw) {

×

UNCOV

719

  if (pRaw == NULL) {

×

720

    return NULL;

×

UNCOV

722

  int32_t         code = 0;

×

UNCOV

723

  int32_t         lino = 0;

×

UNCOV

724

  SSdbRow        *pRow = NULL;

×

UNCOV

725

  SMqConsumerObj *pConsumer = NULL;

×

UNCOV

726

  void           *buf = NULL;

×

UNCOV

728

  terrno = 0;

×

UNCOV

729

  int8_t sver = 0;

×

UNCOV

730

  if (sdbGetRawSoftVer(pRaw, &sver) != 0) {

×

731

    goto CM_DECODE_OVER;

×

UNCOV

734

  if (sver < 1 || sver > MND_CONSUMER_VER_NUMBER) {

×

735

    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;

×

736

    goto CM_DECODE_OVER;

×

UNCOV

739

  pRow = sdbAllocRow(sizeof(SMqConsumerObj));

×

UNCOV

740

  if (pRow == NULL) {

×

741

    goto CM_DECODE_OVER;

×

UNCOV

744

  pConsumer = sdbGetRowObj(pRow);

×

UNCOV

745

  if (pConsumer == NULL) {

×

746

    goto CM_DECODE_OVER;

×

UNCOV

749

  int32_t dataPos = 0;

×

UNCOV

751

  SDB_GET_INT32(pRaw, dataPos, &len, CM_DECODE_OVER);

×

UNCOV

752

  buf = taosMemoryMalloc(len);

×

UNCOV

753

  if (buf == NULL) {

×

754

    terrno = TSDB_CODE_OUT_OF_MEMORY;

×

755

    goto CM_DECODE_OVER;

×

UNCOV

758

  SDB_GET_BINARY(pRaw, dataPos, buf, len, CM_DECODE_OVER);

×

UNCOV

759

  SDB_GET_RESERVE(pRaw, dataPos, MND_CONSUMER_RESERVE_SIZE, CM_DECODE_OVER);

×

UNCOV

761

  if (tDecodeSMqConsumerObj(buf, pConsumer, sver) == NULL) {

×

762

    terrno = TSDB_CODE_OUT_OF_MEMORY;  // TODO set correct error code

×

763

    goto CM_DECODE_OVER;

×

UNCOV

766

  tmsgUpdateDnodeEpSet(&pConsumer->ep);

×

UNCOV

768

CM_DECODE_OVER:

×

UNCOV

769

  taosMemoryFreeClear(buf);

×

UNCOV

770

  if (terrno != TSDB_CODE_SUCCESS) {

×

771

    mError("consumer:0x%" PRIx64 " failed to decode from raw:%p since %s",

×

773

    taosMemoryFreeClear(pRow);

×

UNCOV

776

  return pRow;

×

UNCOV

779

static int32_t mndConsumerActionInsert(SSdb *pSdb, SMqConsumerObj *pConsumer) {

×

UNCOV

780

  if (pConsumer == NULL) {

×

781

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

783

  mInfo("consumer:0x%" PRIx64 " sub insert, cgroup:%s status:%d(%s) epoch:%d", pConsumer->consumerId, pConsumer->cgroup,

×

UNCOV

785

  pConsumer->subscribeTime = pConsumer->createTime;

×

UNCOV

786

  return 0;

×

UNCOV

789

static int32_t mndConsumerActionDelete(SSdb *pSdb, SMqConsumerObj *pConsumer) {

×

UNCOV

790

  if (pConsumer == NULL) {

×

791

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

793

  mInfo("consumer:0x%" PRIx64 " perform delete action, status:(%d)%s", pConsumer->consumerId, pConsumer->status,

×

UNCOV

795

  tClearSMqConsumerObj(pConsumer);

×

UNCOV

796

  return 0;

×

UNCOV

812

static void removeFromTopicList(SArray *topicList, const char *pTopic, int64_t consumerId, char *type) {

×

UNCOV

813

  if (topicList == NULL || pTopic == NULL) {

×

814

    return;

×

UNCOV

816

  int32_t size = taosArrayGetSize(topicList);

×

UNCOV

817

  for (int32_t i = 0; i < size; i++) {

×

UNCOV

818

    char *p = taosArrayGetP(topicList, i);

×

UNCOV

819

    if (strcmp(pTopic, p) == 0) {

×

UNCOV

820

      taosArrayRemove(topicList, i);

×

UNCOV

821

      taosMemoryFree(p);

×

UNCOV

823

      mInfo("[rebalance] consumer:0x%" PRIx64 " remove topic:%s in the %s topic list, remain newTopics:%d",

×

UNCOV

825

      break;

×

UNCOV

830

static bool existInCurrentTopicList(const SMqConsumerObj *pConsumer, const char *pTopic) {

×

UNCOV

831

  if (pConsumer == NULL || pTopic == NULL) {

×

832

    return false;

×

UNCOV

834

  bool    existing = false;

×

UNCOV

835

  int32_t size = taosArrayGetSize(pConsumer->currentTopics);

×

UNCOV

836

  for (int32_t i = 0; i < size; i++) {

×

UNCOV

837

    char *topic = taosArrayGetP(pConsumer->currentTopics, i);

×

UNCOV

838

    if (topic && strcmp(topic, pTopic) == 0) {

×

839

      existing = true;

×

840

      break;

×

UNCOV

844

  return existing;

×

UNCOV

847

static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pOldConsumer, SMqConsumerObj *pNewConsumer) {

×

UNCOV

848

  if (pOldConsumer == NULL || pNewConsumer == NULL) {

×

849

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

851

  mInfo("consumer:0x%" PRIx64 " perform update action, update type:%d, subscribe-time:%" PRId64 ", createTime:%" PRId64,

×

UNCOV

854

  taosWLockLatch(&pOldConsumer->lock);

×

UNCOV

856

  if (pNewConsumer->updateType == CONSUMER_UPDATE_SUB) {

×

UNCOV

857

    TSWAP(pOldConsumer->rebNewTopics, pNewConsumer->rebNewTopics);

×

UNCOV

858

    TSWAP(pOldConsumer->rebRemovedTopics, pNewConsumer->rebRemovedTopics);

×

UNCOV

859

    TSWAP(pOldConsumer->assignedTopics, pNewConsumer->assignedTopics);

×

UNCOV

861

    pOldConsumer->subscribeTime = taosGetTimestampMs();

×

UNCOV

862

    pOldConsumer->status = MQ_CONSUMER_STATUS_REBALANCE;

×

UNCOV

863

    mInfo("consumer:0x%" PRIx64 " subscribe update, modify existed consumer", pOldConsumer->consumerId);

×

UNCOV

864

  } else if (pNewConsumer->updateType == CONSUMER_UPDATE_REB) {

×

UNCOV

865

    (void)atomic_add_fetch_32(&pOldConsumer->epoch, 1);

×

UNCOV

866

    pOldConsumer->rebalanceTime = taosGetTimestampMs();

×

UNCOV

867

    mInfo("[rebalance] consumer:0x%" PRIx64 " rebalance update, only rebalance time", pOldConsumer->consumerId);

×

UNCOV

868

  } else if (pNewConsumer->updateType == CONSUMER_ADD_REB) {

×

UNCOV

869

    void *tmp = taosArrayGetP(pNewConsumer->rebNewTopics, 0);

×

UNCOV

870

    if (tmp == NULL){

×

871

      return TSDB_CODE_TMQ_INVALID_MSG;

×

UNCOV

873

    char *pNewTopic = taosStrdup(tmp);

×

UNCOV

874

    if (pNewTopic == NULL) {

×

875

      return terrno;

×

UNCOV

877

    removeFromTopicList(pOldConsumer->rebNewTopics, pNewTopic, pOldConsumer->consumerId, "new");

×

UNCOV

878

    bool existing = existInCurrentTopicList(pOldConsumer, pNewTopic);

×

UNCOV

879

    if (existing) {

×

880

      mError("[rebalance] consumer:0x%" PRIx64 " add new topic:%s should not in currentTopics", pOldConsumer->consumerId, pNewTopic);

×

881

      taosMemoryFree(pNewTopic);

×

UNCOV

883

      if (taosArrayPush(pOldConsumer->currentTopics, &pNewTopic) == NULL) {

×

884

        taosMemoryFree(pNewTopic);

×

885

        return TSDB_CODE_TMQ_INVALID_MSG;

×

UNCOV

887

      taosArraySort(pOldConsumer->currentTopics, taosArrayCompareString);

×

UNCOV

890

    int32_t status = pOldConsumer->status;

×

UNCOV

892

    if (taosArrayGetSize(pOldConsumer->rebNewTopics) == 0 && taosArrayGetSize(pOldConsumer->rebRemovedTopics) == 0) {

×

UNCOV

893

      pOldConsumer->status = MQ_CONSUMER_STATUS_READY;

×

UNCOV

896

    pOldConsumer->rebalanceTime = taosGetTimestampMs();

×

UNCOV

897

    (void)atomic_add_fetch_32(&pOldConsumer->epoch, 1);

×

UNCOV

899

    mInfo("[rebalance] consumer:0x%" PRIx64 " rebalance update add, state (%d)%s -> (%d)%s, new epoch:%d, reb-time:%" PRId64

×

UNCOV

906

  } else if (pNewConsumer->updateType == CONSUMER_REMOVE_REB) {

×

UNCOV

907

    char *topic = taosArrayGetP(pNewConsumer->rebRemovedTopics, 0);

×

UNCOV

908

    if (topic == NULL){

×

909

      return TSDB_CODE_TMQ_INVALID_MSG;

×

UNCOV

911

    removeFromTopicList(pOldConsumer->rebRemovedTopics, topic, pOldConsumer->consumerId, "remove");

×

UNCOV

912

    removeFromTopicList(pOldConsumer->currentTopics, topic, pOldConsumer->consumerId, "current");

×

UNCOV

914

    int32_t status = pOldConsumer->status;

×

UNCOV

916

    if (taosArrayGetSize(pOldConsumer->rebNewTopics) == 0 && taosArrayGetSize(pOldConsumer->rebRemovedTopics) == 0) {

×

UNCOV

917

      pOldConsumer->status = MQ_CONSUMER_STATUS_READY;

×

UNCOV

919

    pOldConsumer->rebalanceTime = taosGetTimestampMs();

×

UNCOV

920

    (void)atomic_add_fetch_32(&pOldConsumer->epoch, 1);

×

UNCOV

922

    mInfo("[rebalance]consumer:0x%" PRIx64 " rebalance update remove, state (%d)%s -> (%d)%s, new epoch:%d, reb-time:%" PRId64

×

UNCOV

930

  taosWUnLockLatch(&pOldConsumer->lock);

×

UNCOV

931

  return 0;

×

UNCOV

934

int32_t mndAcquireConsumer(SMnode *pMnode, int64_t consumerId, SMqConsumerObj** pConsumer) {

×

UNCOV

935

  if (pMnode == NULL || pConsumer == NULL) {

×

936

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

938

  SSdb           *pSdb = pMnode->pSdb;

×

UNCOV

939

  *pConsumer = sdbAcquire(pSdb, SDB_CONSUMER, &consumerId);

×

UNCOV

940

  if (*pConsumer == NULL) {

×

UNCOV

941

    return TSDB_CODE_MND_CONSUMER_NOT_EXIST;

×

UNCOV

943

  return 0;

×

UNCOV

946

void mndReleaseConsumer(SMnode *pMnode, SMqConsumerObj *pConsumer) {

×

UNCOV

947

  if (pMnode == NULL || pConsumer == NULL) {

×

UNCOV

948

    return;

×

UNCOV

950

  SSdb *pSdb = pMnode->pSdb;

×

UNCOV

951

  sdbRelease(pSdb, pConsumer);

×

UNCOV

954

static int32_t mndRetrieveConsumer(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rowsCapacity) {

×

UNCOV

955

  if (pReq == NULL || pShow == NULL || pBlock == NULL) {

×

956

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

958

  SMnode         *pMnode = pReq->info.node;

×

UNCOV

959

  SSdb           *pSdb = pMnode->pSdb;

×

UNCOV

960

  int32_t         numOfRows = 0;

×

UNCOV

961

  SMqConsumerObj *pConsumer = NULL;

×

UNCOV

962

  int32_t         code = 0;

×

UNCOV

963

  char           *parasStr = NULL;

×

UNCOV

964

  char           *status = NULL;

×

UNCOV

966

  while (numOfRows < rowsCapacity) {

×

UNCOV

967

    pShow->pIter = sdbFetch(pSdb, SDB_CONSUMER, pShow->pIter, (void **)&pConsumer);

×

UNCOV

968

    if (pShow->pIter == NULL) {

×

UNCOV

969

      break;

×

UNCOV

972

    if (taosArrayGetSize(pConsumer->assignedTopics) == 0) {

×

UNCOV

973

      mInfo("showing consumer:0x%" PRIx64 " no assigned topic, skip", pConsumer->consumerId);

×

UNCOV

974

      sdbRelease(pSdb, pConsumer);

×

UNCOV

975

      continue;

×

UNCOV

978

    taosRLockLatch(&pConsumer->lock);

×

UNCOV

979

    mInfo("showing consumer:0x%" PRIx64, pConsumer->consumerId);

×

UNCOV

981

    int32_t topicSz = taosArrayGetSize(pConsumer->assignedTopics);

×

UNCOV

982

    bool    hasTopic = true;

×

UNCOV

983

    if (topicSz == 0) {

×

984

      hasTopic = false;

×

985

      topicSz = 1;

×

UNCOV

988

    if (numOfRows + topicSz > rowsCapacity) {

×

989

      MND_TMQ_RETURN_CHECK(blockDataEnsureCapacity(pBlock, numOfRows + topicSz));

×

UNCOV

992

    for (int32_t i = 0; i < topicSz; i++) {

×

UNCOV

993

      SColumnInfoData *pColInfo = NULL;

×

UNCOV

994

      int32_t          cols = 0;

×

UNCOV

997

      char consumerIdHex[TSDB_CONSUMER_ID_LEN + VARSTR_HEADER_SIZE] = {0};

×

UNCOV

998

      (void)snprintf(varDataVal(consumerIdHex), TSDB_CONSUMER_ID_LEN, "0x%" PRIx64, pConsumer->consumerId);

×

UNCOV

999

      varDataSetLen(consumerIdHex, strlen(varDataVal(consumerIdHex)));

×

UNCOV

1001

      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

UNCOV

1002

      MND_TMQ_NULL_CHECK(pColInfo);

×

UNCOV

1003

      MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)consumerIdHex, false));

×

UNCOV

1006

      char cgroup[TSDB_CGROUP_LEN + VARSTR_HEADER_SIZE] = {0};

×

UNCOV

1007

      STR_TO_VARSTR(cgroup, pConsumer->cgroup);

×

UNCOV

1009

      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

UNCOV

1010

      MND_TMQ_NULL_CHECK(pColInfo);

×

UNCOV

1011

      MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)cgroup, false));

×

UNCOV

1014

      char clientId[TSDB_CLIENT_ID_LEN + VARSTR_HEADER_SIZE] = {0};

×

UNCOV

1015

      STR_TO_VARSTR(clientId, pConsumer->clientId);

×

UNCOV

1017

      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

UNCOV

1018

      MND_TMQ_NULL_CHECK(pColInfo);

×

UNCOV

1019

      MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)clientId, false));

×

UNCOV

1022

      char user[TSDB_USER_LEN + VARSTR_HEADER_SIZE] = {0};

×

UNCOV

1023

      STR_TO_VARSTR(user, pConsumer->user);

×

UNCOV

1025

      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

UNCOV

1026

      MND_TMQ_NULL_CHECK(pColInfo);

×

UNCOV

1027

      MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)user, false));

×

UNCOV

1030

      char fqdn[TSDB_FQDN_LEN + VARSTR_HEADER_SIZE] = {0};

×

UNCOV

1031

      STR_TO_VARSTR(fqdn, pConsumer->fqdn);

×

UNCOV

1033

      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

UNCOV

1034

      MND_TMQ_NULL_CHECK(pColInfo);

×

UNCOV

1035

      MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)fqdn, false));

×

UNCOV

1038

      const char *pStatusName = mndConsumerStatusName(pConsumer->status);

×

UNCOV

1039

      status = taosMemoryCalloc(1, pShow->pMeta->pSchemas[cols].bytes);

×

UNCOV

1040

      MND_TMQ_NULL_CHECK(status);

×

UNCOV

1041

      STR_TO_VARSTR(status, pStatusName);

×

UNCOV

1043

      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

UNCOV

1044

      MND_TMQ_NULL_CHECK(pColInfo);

×

UNCOV

1045

      MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)status, false));

×

UNCOV

1046

      taosMemoryFreeClear(status);

×

UNCOV

1049

      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

UNCOV

1050

      MND_TMQ_NULL_CHECK(pColInfo);

×

UNCOV

1051

      if (hasTopic) {

×

UNCOV

1052

        char        topic[TSDB_TOPIC_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};

×

UNCOV

1053

        mndTopicGetShowName(taosArrayGetP(pConsumer->assignedTopics, i), topic + VARSTR_HEADER_SIZE);

×

UNCOV

1054

        *(VarDataLenT *)(topic) = strlen(topic + VARSTR_HEADER_SIZE);

×

UNCOV

1055

        MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)topic, false));

×

1057

        MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, NULL, true));

×

UNCOV

1061

      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

UNCOV

1062

      MND_TMQ_NULL_CHECK(pColInfo);

×

UNCOV

1063

      MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)&pConsumer->createTime, false));

×

UNCOV

1066

      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

UNCOV

1067

      MND_TMQ_NULL_CHECK(pColInfo);

×

UNCOV

1068

      MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)&pConsumer->subscribeTime, false));

×

UNCOV

1071

      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

UNCOV

1072

      MND_TMQ_NULL_CHECK(pColInfo);

×

UNCOV

1073

      MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)&pConsumer->rebalanceTime, pConsumer->rebalanceTime == 0));

×

UNCOV

1075

      char         buf[TSDB_OFFSET_LEN] = {0};

×

UNCOV

1076

      STqOffsetVal pVal = {.type = pConsumer->resetOffsetCfg};

×

UNCOV

1077

      tFormatOffset(buf, TSDB_OFFSET_LEN, &pVal);

×

UNCOV

1079

      parasStr = taosMemoryCalloc(1, pShow->pMeta->pSchemas[cols].bytes);

×

UNCOV

1080

      MND_TMQ_NULL_CHECK(parasStr);

×

UNCOV

1081

      (void)snprintf(varDataVal(parasStr), pShow->pMeta->pSchemas[cols].bytes - VARSTR_HEADER_SIZE, "tbname:%d,commit:%d,interval:%dms,reset:%s", pConsumer->withTbName,

×

UNCOV

1082

              pConsumer->autoCommit, pConsumer->autoCommitInterval, buf);

×

UNCOV

1083

      varDataSetLen(parasStr, strlen(varDataVal(parasStr)));

×

UNCOV

1085

      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

UNCOV

1086

      MND_TMQ_NULL_CHECK(pColInfo);

×

UNCOV

1087

      MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)parasStr, false));

×

UNCOV

1088

      taosMemoryFreeClear(parasStr);

×

UNCOV

1089

      numOfRows++;

×

UNCOV

1092

    taosRUnLockLatch(&pConsumer->lock);

×

UNCOV

1093

    sdbRelease(pSdb, pConsumer);

×

UNCOV

1095

    pBlock->info.rows = numOfRows;

×

UNCOV

1098

  pShow->numOfRows += numOfRows;

×

UNCOV

1099

  return numOfRows;

×

1101

END:

×

1102

  taosMemoryFreeClear(status);

×

1103

  taosMemoryFreeClear(parasStr);

×

1104

  return code;

×

1107

static void mndCancelGetNextConsumer(SMnode *pMnode, void *pIter) {

×

1108

  if (pMnode == NULL || pIter == NULL) return;

×

1109

  SSdb *pSdb = pMnode->pSdb;

×

1110

  sdbCancelFetchByType(pSdb, pIter, SDB_CONSUMER);

×

UNCOV

1113

const char *mndConsumerStatusName(int status) {

×

UNCOV

1114

  switch (status) {

×

UNCOV

1115

    case MQ_CONSUMER_STATUS_READY:

×

UNCOV

1116

      return "ready";

×

UNCOV

1119

    case MQ_CONSUMER_STATUS_REBALANCE:

×

UNCOV

1120

      return "rebalancing";

×

1121

    default:

×

1122

      return "unknown";

×

taosdata / TDengine / #3562

Source File
Press 'n' to go to next uncovered line, 'b' for previous

taosdata / TDengine / #3562

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous